def __init__(self, max_epochs, min_proportional_decrease=0.0):
        '''
        max_epochs: int
          Stop training if the monitored value doesn't decrease for
          this many epochs.

        min_proportional_decrease: float
          If this value is T, the monitored value is V, and the last known
          minimum of V is Vm, then V is considered a decrease only if
          V < (1.0 - T) * Vm
        '''
        super(StopsOnStagnation, self).__init__()

        assert_greater(max_epochs, 0)
        assert_true(numpy.issubdtype(type(max_epochs), numpy.integer))

        assert_greater_equal(min_proportional_decrease, 0.0)

        self._max_epochs_since_min = max_epochs
        self._min_proportional_decrease = min_proportional_decrease
        self._epochs_since_min = 0

        # This gets set to self._min_value at each siginificant decrese.
        # A "significant decrease" is a decrease in self._min_value
        # by more than min_proportional_decrease relative to
        # _significant_min_value.
        self._significant_min_value = None

def init_session_retry(session, max_retries):
    from requests.adapters import HTTPAdapter
    from nose.tools import assert_greater_equal
    assert_greater_equal(max_retries, 0)
    session.mount('http://', HTTPAdapter(max_retries=max_retries))
    session.mount('https://', HTTPAdapter(max_retries=max_retries))
    return session

 def check_descriptor_between(self, catchment, descr, lower, upper):
     nt.assert_greater_equal(getattr(catchment.descriptors, descr), lower,
                             msg="Catchment {} does not have a `descriptors.`{}>={}"
                             .format(catchment.id, descr, lower))
     nt.assert_less_equal(getattr(catchment.descriptors, descr), upper,
                          msg="Catchment {} does not have a `descriptors.`{}<={}"
                          .format(catchment.id, descr, upper))

def test_upload_chunk__expired_url():
    upload_parts = [{'uploadPresignedUrl': 'https://www.fake.url/fake/news',
                     'partNumber': 420},
                    {'uploadPresignedUrl': 'https://www.google.com',
                     'partNumber': 421},
                    {'uploadPresignedUrl': 'https://rito.pls/',
                     'partNumber': 422},
                    {'uploadPresignedUrl': 'https://never.lucky.gg',
                     'partNumber': 423}
                    ]

    value_doesnt_matter = None
    expired = Value(c_bool, False)
    mocked_get_chunk_function = MagicMock(side_effect=[1, 2, 3, 4])

    with patch.object(multipart_upload, "_put_chunk",
                      side_effect=SynapseHTTPError("useless message",
                                                   response=MagicMock(status_code=403))) as mocked_put_chunk, \
         patch.object(warnings, "warn") as mocked_warn:
        def chunk_upload(part):
            return _upload_chunk(part, completed=value_doesnt_matter, status=value_doesnt_matter, syn=syn,
                                 filename=value_doesnt_matter, get_chunk_function=mocked_get_chunk_function,
                                 fileSize=value_doesnt_matter, partSize=value_doesnt_matter,
                                 t0=value_doesnt_matter, expired=expired, bytes_already_uploaded=value_doesnt_matter)
        # 2 threads both with urls that have expired
        mp = Pool(4)
        mp.map(chunk_upload, upload_parts)
        assert_true(expired.value)

        # assert warnings.warn was only called once
        mocked_warn.assert_called_once_with("The pre-signed upload URL has expired. Restarting upload...\n")

        # assert _put_chunk was called at least once
        assert_greater_equal(len(mocked_put_chunk.call_args_list), 1)

def test_incentive_process(lim=1e-14):
    """
    Compare stationary distribution computations to known analytic form for
    neutral landscape for the Moran process.
    """

    for n, N in [(2, 10), (2, 40), (3, 10), (3, 20), (4, 10)]:
        mu = (n - 1.) / n * 1./ (N + 1)
        alpha = N * mu / (n - 1. - n * mu)

        # Neutral landscape is the default
        edges = incentive_process.compute_edges(N, num_types=n,
                                                incentive_func=replicator, mu=mu)
        for logspace in [False, True]:
            stationary_1 = incentive_process.neutral_stationary(
                N, alpha, n, logspace=logspace)
            for exact in [False, True]:
                stationary_2 = stationary_distribution(
                    edges, lim=lim, logspace=logspace, exact=exact)
                for key in stationary_1.keys():
                    assert_almost_equal(
                        stationary_1[key], stationary_2[key], places=4)

        # Check that the stationary distribution satisfies balance conditions
        check_detailed_balance(edges, stationary_1)
        check_global_balance(edges, stationary_1)
        check_eigenvalue(edges, stationary_1)

        # Test Entropy Rate bounds
        er = entropy_rate(edges, stationary_1)
        h = (2. * n - 1) / n * numpy.log(n)
        assert_less_equal(er, h)
        assert_greater_equal(er, 0)

    def __init__(self,
                 model,
                 training_state,
                 filepath,
                 overwrite=True,
                 epochs_seen=0):

        def check_filepath(filepath):
            if os.path.isdir(filepath):
                path = filepath
                filename = ""
            else:
                path, filename = os.path.split(filepath)

            assert_true(os.path.isdir(path),
                        "{} isn't a directory".format(path))
            assert_equal(os.path.splitext(filename)[1], '.h5')

        assert_is_instance(model, H5Saveable)
        assert_is_instance(training_state, H5Saveable)
        check_filepath(filepath)
        assert_is_instance(overwrite, bool)
        assert_greater_equal(epochs_seen, 0)

        self._filepath = filepath
        self._model = model
        self._training_state = training_state
        self._overwrite = overwrite

def check_sum_of_calls(object_, methods, maximum_calls, minimum_calls=1):
    """
    Instruments the given methods on the given object to verify that the total sum of calls made to the
    methods falls between minumum_calls and maximum_calls.
    """
    mocks = {
        method: Mock(wraps=getattr(object_, method))
        for method in methods
    }

    with patch.multiple(object_, **mocks):
        yield

    call_count = sum(mock.call_count for mock in mocks.values())
    calls = pprint.pformat({
        method_name: mock.call_args_list
        for method_name, mock in mocks.items()
    })

    # Assertion errors don't handle multi-line values, so pretty-print to std-out instead
    if not minimum_calls <= call_count <= maximum_calls:
        print "Expected between {} and {} calls, {} were made. Calls: {}".format(
            minimum_calls,
            maximum_calls,
            call_count,
            calls,
        )

    # verify the counter actually worked by ensuring we have counted greater than (or equal to) the minimum calls
    assert_greater_equal(call_count, minimum_calls)

    # now verify the number of actual calls is less than (or equal to) the expected maximum
    assert_less_equal(call_count, maximum_calls)

    def test_get_next_candidate(self):
        """
        Tests the get next candidate function.
        Tests:
            - The candidate's parameters are acceptable
        """

        cand = None
        counter = 0
        while cand is None and counter < 20:
            cand = self.EAss.get_next_candidate()
            time.sleep(0.1)
            counter += 1
        if counter == 20:
            raise Exception("Received no result in the first 2 seconds.")
        assert_is_none(cand.result)
        params = cand.params
        assert_less_equal(params["x"], 1)
        assert_greater_equal(params["x"], 0)
        assert_in(params["name"], self.param_defs["name"].values)
        self.EAss.update(cand, "pausing")
        time.sleep(1)
        new_cand = None
        while new_cand is None and counter < 20:
            new_cand = self.EAss.get_next_candidate()
            time.sleep(0.1)
            counter += 1
        if counter == 20:
            raise Exception("Received no result in the first 2 seconds.")
        assert_equal(new_cand, cand)

 def t(s, n, expected):
     result = M.ltrim(s, n)
     assert_greater_equal(
         max(1, n),
         len(result)
     )
     assert_equal(result, expected)

    def init_sparse_linear(shared_variable, num_nonzeros, rng):
        params = shared_variable.get_value()
        params[...] = 0.0

        assert_greater_equal(num_nonzeros, 0)
        assert_less_equal(num_nonzeros, params.shape[0])

        for c in xrange(params.shape[1]):
            indices = rng.choice(params.shape[0], size=num_nonzeros, replace=False)

            # normal dist with stddev=1.0, divided by 255.0
            #
            # We need to divide by 255 for convergence. This is because
            # we're using unnormalized (i.e. 0 to 255) pixel values, unlike the
            # 0.0-to-1.0 pixels in
            # pylearn2.scripts.tutorials.multilayer_perceptron/
            #
            # We could just do as the above tutorial does and normalize the
            # pixels to [0.0, 1.0], and not rescale the weights. However,
            # experiments show that this converges to a higher error, and also
            # makes mnist_visualizer.py's results look very "staticky", without
            # any recognizable digit hallucinations.
            params[indices, c] = rng.randn(num_nonzeros) / 255.0

        shared_variable.set_value(params)

    def init_sparse_bias(shared_variable, num_nonzeros, rng):
        """
        Mimics the sparse initialization in
        pylearn2.models.mlp.Linear.set_input_space()
        """

        params = shared_variable.get_value()
        assert_equal(params.shape[0], 1)

        assert_greater_equal(num_nonzeros, 0)
        assert_less_equal(num_nonzeros, params.shape[1])

        params[...] = 0.0

        indices = rng.choice(params.size, size=num_nonzeros, replace=False)

        # normal dist with stddev=1.0
        params[0, indices] = rng.randn(num_nonzeros)

        # Found that for biases, this didn't help (it increased the
        # final misclassification rate by .001)
        # if num_nonzeros > 0:
        #     params /= float(num_nonzeros)

        shared_variable.set_value(params)

def test_wright_fisher(N=20, lim=1e-10, n=2):
    """Test 2 dimensional Wright-Fisher process."""
    for n in [2, 3]:
        mu = (n - 1.) / n * 1. / (N + 1)
        m = numpy.ones((n, n)) # neutral landscape
        fitness_landscape = linear_fitness_landscape(m)
        incentive = replicator(fitness_landscape)

        # Wright-Fisher
        for low_memory in [True, False]:
            edge_func = wright_fisher.multivariate_transitions(
                N, incentive, mu=mu, num_types=n, low_memory=low_memory)
            states = list(simplex_generator(N, d=n-1))
            for logspace in [False, True]:
                s = stationary_distribution(
                    edge_func, states=states, iterations=200, lim=lim,
                    logspace=logspace)
                wf_edges = edge_func_to_edges(edge_func, states)

                er = entropy_rate(wf_edges, s)
                assert_greater_equal(er, 0)

                # Check that the stationary distribution satistifies balance
                # conditions
                check_detailed_balance(wf_edges, s, places=2)
                check_global_balance(wf_edges, s, places=4)
                check_eigenvalue(wf_edges, s, places=2)

def test_external_versions_basic():
    ev = ExternalVersions()
    assert_equal(ev._versions, {})
    assert_equal(ev["duecredit"], __version__)
    # and it could be compared
    assert_greater_equal(ev["duecredit"], __version__)
    assert_greater(ev["duecredit"], "0.1")

    # For non-existing one we get None
    assert_equal(ev["duecreditnonexisting"], None)
    # and nothing gets added to _versions for nonexisting
    assert_equal(set(ev._versions.keys()), {"duecredit"})

    # but if it is a module without version, we get it set to UNKNOWN
    assert_equal(ev["os"], ev.UNKNOWN)
    # And get a record on that inside
    assert_equal(ev._versions.get("os"), ev.UNKNOWN)
    # And that thing is "True", i.e. present
    assert ev["os"]
    # but not comparable with anything besides itself (was above)
    assert_raises(TypeError, cmp, ev["os"], "0")
    assert_raises(TypeError, assert_greater, ev["os"], "0")

    # And we can get versions based on modules themselves
    from duecredit.tests import mod

    assert_equal(ev[mod], mod.__version__)

    def init_sparse_linear(shared_variable, num_nonzeros, rng):
        params = shared_variable.get_value()
        params[...] = 0.0

        assert_greater_equal(num_nonzeros, 0)
        assert_less_equal(num_nonzeros, params.shape[0])

        for c in xrange(params.shape[1]):
            indices = rng.choice(params.shape[0],
                                 size=num_nonzeros,
                                 replace=False)

            # normal dist with stddev=1.0
            params[indices, c] = rng.randn(num_nonzeros)

        # TODO: it's somewhat worrisome that the tutorial in
        # pylearn2.scripts.tutorials.multilayer_perceptron/
        #   multilayer_perceptron.ipynb
        # seems to do fine without scaling the weights like this
        if num_nonzeros > 0:
            params /= float(num_nonzeros)
            # Interestingly, while this seems more correct (normalize
            # columns to norm=1), it prevents the NN from converging.
            # params /= numpy.sqrt(float(num_nonzeros))

        shared_variable.set_value(params)

    def elev_label_to_elev(elev_label):
        assert_greater_equal(elev_label, -1)
        elev_degrees = 30 if elev_label == -1 else (elev_label * 5 + 30)

        assert_greater_equal(elev_degrees, 30)
        assert_less_equal(elev_degrees, 90)
        return deg_to_rad(elev_degrees)

 def test_local_inputs_contents(self):
     xs = self.mws._local_search_xs(0, 20, 20)
     random.seed(1)
     # this is stochastic, so run it 100 times & hope any errors are caught
     for _ in xrange(100):
         for i, x in enumerate(xs):
             assert_greater_equal(i + 1, x)
             assert_less_equal(i, x)

 def test_get_gzh_article_by_hot_real(self):
     gzh_articles = ws_api.get_gzh_article_by_hot(WechatSogouConst.hot_index.gaoxiao,
                                                  identify_image_callback=self.identify_image_callback_sogou)
     for gzh_article in gzh_articles:
         assert_in('gzh', gzh_article)
         assert_in('article', gzh_article)
         assert_in('http://mp.weixin.qq.com/s?src=', gzh_article['article']['url'])
     assert_greater_equal(len(gzh_articles), 10)

 def test_get_gzh_article_by_history_real(self):
     gzh_article = ws_api.get_gzh_article_by_history(gaokao_keyword,
                                                     identify_image_callback_sogou=self.identify_image_callback_sogou,
                                                     identify_image_callback_weixin=self.identify_image_callback_ruokuai_weixin)
     assert_in('gzh', gzh_article)
     assert_in('article', gzh_article)
     assert_in('wx.qlogo.cn', gzh_article['gzh']['headimage'])
     assert_greater_equal(len(gzh_article['article']), 1)

def _all_pairs_connectivity(G, cc, k, memo):
    # Brute force check
    for u, v in it.combinations(cc, 2):
        # Use a memoization dict to save on computation
        connectivity = _memo_connectivity(G, u, v, memo)
        if G.is_directed():
            connectivity = min(connectivity, _memo_connectivity(G, v, u, memo))
        assert_greater_equal(connectivity, k)

            def check_all_sessions(idx, n, val):
                write_nodes, read_nodes, strong_consistency = self.get_num_nodes(idx)
                results = []
                for s in sessions:
                    results.append(outer.query_counter(s, n, val, read_cl, check_ret=strong_consistency))

                assert_greater_equal(results.count(val), write_nodes, "Failed to read value from sufficient number of nodes, required {} nodes to have a counter "
                                     "value of {} at key {}, instead got these values: {}".format(write_nodes, val, n, results))