Python hypothesis.strategies 模块，integers() 实例源码

def spark_application(app_id):
    """Mock of the Spark jobs REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'jobId': st.integers(0),
        'name': st.text(),
        'submissionTime': st.text(),
        'completionTime': st.text(),
        'stageIds': st.lists(st.integers(0), average_size=3),
        'status': st.sampled_from(['SUCCEEDED', 'RUNNING', 'FAILED']),
        'numTasks': st.integers(0),
        'numActiveTasks': st.integers(0),
        'numCompletedTasks': st.integers(0),
        'numSkippedTasks': st.integers(0),
        'numFailedTasks': st.integers(0),
        'numActiveStages': st.integers(0),
        'numCompletedStages': st.integers(0),
        'numSkippedStages': st.integers(0),
        'numFailedStages': st.integers(0),
    })
    result = json.dumps(st.lists(d, average_size=3).example())
    redis.set(request.base_url, result)
    return jsonify(result)

def test_len(self, length: int) -> None:
        """
        Tests that the length of the job set is whatever the number of
        entries in the SQL query backing this job says it is.

        .. note::

            For some reason, ``unittest.mock`` won't let me set an integer
            bigger than the ``max_value`` specified in this test as a return
            value. Because of that, the max value on the integers needs to
            be constrained.

        :param length: A randomly-generated length
        """
        self.root_query.count = mock.MagicMock(return_value=length)
        self.assertEqual(length, len(self.job_list))

def arguments_node(draw, annotated=False):
    n = draw(hs.integers(min_value=1, max_value=5))
    args = draw(hs.lists(name_node(None), min_size=n, max_size=n))
    if annotated:
        annotations = draw(hs.lists(name_node(annotation), min_size=n, max_size=n))
    else:
        annotations = None
    node = astroid.Arguments()
    node.postinit(
        args,
        None,
        None,
        None,
        annotations
    )
    return node

def test_to_byte_array_list(self, list_val):
        """
        to_byte_array() with param:
        :param list_val: randomly list of postive integers (within byte range).
        """
        pointer, leng = to_byte_array(list_val)
        self.verify_c_type(pointer, leng)

        py_bytes = self.reverse_case(pointer, leng, to_byte_array)

        # Create list from returned byte-string
        py_list = []
        for i in range(0, len(py_bytes), 2):
            py_list.append(int(py_bytes[i:i + 2], 16))

        assert py_list == list_val

def gfywords(draw):
    x = adjectives[draw(st.integers(min_value=0, max_value=len(adjectives)-1))]
    y = adjectives[draw(st.integers(min_value=0, max_value=len(adjectives)-1))]
    z = animals[draw(st.integers(min_value=0, max_value=len(animals)-1))]
    assume(x != y)
    return ' '.join((x, y, z))

def gfycodes(draw):
    x = adjectives[draw(st.integers(min_value=0, max_value=len(adjectives)))]
    y = adjectives[draw(st.integers(min_value=0, max_value=len(adjectives)))]
    z = animals[draw(st.integers(min_value=0, max_value=len(animals)))]
    assume(x != y)
    return ''.join((x, y, z))

def random_reference(draw, prefix = random_prefix()):
    number = st.integers(min_value = 0)
    return draw(st.tuples(prefix, number))

def test_positive_integers_are_allowed(self, value):
        """Positive integers go through `check_positive_integer` unmodified."""
        assert cli.check_positive_integer(name='test', value=value) == value

def test_non_positive_integers_are_valueerror(self, value):
        """Negative or zero integers are rejected with a `ValueError`."""
        with pytest.raises(ValueError) as exc:
            cli.check_positive_integer(name='test', value=value)
        assert 'should be positive' in exc.value.args[0]

def color_strategy():
    """Hypothesis strategy for generating instances of `RGBColor`."""
    return st.builds(
        RGBColor,
        st.integers(min_value=0, max_value=255),
        st.integers(min_value=0, max_value=255),
        st.integers(min_value=0, max_value=255),
    )

def enums_of_primitives(draw):
    """Generate enum classes with primitive values."""
    if is_py2:
        names = draw(st.sets(st.text(alphabet=string.ascii_letters,
                                     min_size=1),
                             min_size=1))
    else:
        names = draw(st.sets(st.text(min_size=1), min_size=1))
    n = len(names)
    vals = draw(st.one_of(st.sets(st.one_of(
            st.integers(),
            st.floats(allow_nan=False),
            st.text(min_size=1)),
        min_size=n, max_size=n)))
    return Enum('HypEnum', list(zip(names, vals)))

def int_attrs(draw, defaults=None):
    """
    Generate a tuple of an attribute and a strategy that yields ints for that
    attribute.
    """
    default = NOTHING
    if defaults is True or (defaults is None and draw(st.booleans())):
        default = draw(st.integers())
    return ((attr.ib(default=default), st.integers()))

def dict_attrs(draw, defaults=None):
    """
    Generate a tuple of an attribute and a strategy that yields dictionaries
    for that attribute. The dictionaries map strings to integers.
    """
    default = NOTHING
    val_strat = st.dictionaries(keys=st.text(), values=st.integers())
    if defaults is True or (defaults is None and draw(st.booleans())):
        default_val = draw(val_strat)
        default = attr.Factory(lambda: default_val)
    return ((attr.ib(default=default), val_strat))

def metrics():
    """Mock of the YARN cluster metrics REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'activeNodes': st.integers(0),
        'allocatedMB': st.integers(0),
        'allocatedVirtualCores': st.integers(0),
        'appsCompleted': st.integers(0),
        'appsFailed': st.integers(0),
        'appsKilled': st.integers(0),
        'appsPending': st.integers(0),
        'appsRunning': st.integers(0),
        'appsSubmitted': st.integers(0),
        'availableMB': st.integers(0),
        'availableVirtualCores': st.integers(0),
        'containersAllocated': st.integers(0),
        'containersPending': st.integers(0),
        'containersReserved': st.integers(0),
        'decommissionedNodes': st.integers(0),
        'lostNodes': st.integers(0),
        'rebootedNodes': st.integers(0),
        'reservedMB': st.integers(0),
        'reservedVirtualCores': st.integers(0),
        'totalMB': st.integers(0),
        'totalNodes': st.integers(0),
        'totalVirtualCores': st.integers(0),
        'unhealthyNodes': st.integers(0)
    })
    result = json.dumps({
        'clusterMetrics': d.example()
    })
    redis.set(request.base_url, result)
    return jsonify(result)

def api_exceptions(
        draw,
        status_codes=integers(min_value=400, max_value=599),
        titles=text(),
        details=text()
) -> APIExceptionInterface:
    return _APIException(
        draw(status_codes), draw(titles), draw(details)
    )

def list_of_strings(draw):
        return [draw(text(min_size=1, average_size=70))
                for i in range(draw(integers(min_value=0, max_value=100)))]

def test_prefix_complex(data, stuff):
    pattern, statements = stuff
    quantity = data.draw(integers(min_value=0, max_value=max(0, len(statements) - 1)))
    statements = statements[:quantity]

    works, unhandled = matches(statements, pattern)
    assert works in [partial, success]

def list_of_strings(draw):
        return [draw(text(min_size=1, average_size=70))
                for i in range(draw(integers(min_value=0, max_value=100)))]

def index_node(draw, value=const_node(hs.integers())):
    node = astroid.Index()
    node.postinit(draw(value))
    return node

def slice_node(draw):
    lower = draw(hs.one_of(const_node(hs.integers()), hs.none()))
    upper = draw(hs.one_of(const_node(hs.integers()), hs.none()))
    step = draw(hs.one_of(const_node(hs.integers()), hs.none()))
    node = astroid.Slice()
    node.postinit(lower, upper, step)
    return node

def bipartite_graph(draw):
    m = draw(st.integers(min_value=1, max_value=4))
    n = draw(st.integers(min_value=m, max_value=5))

    graph = BipartiteGraph()
    for i in range(n):
        for j in range(m):
            b = draw(st.booleans())
            if b:
                graph[i, j] = b

    return graph

def sequence_vars(draw):
    num_vars = draw(st.integers(min_value=1, max_value=4))

    variables = []
    for i in range(num_vars):
        name = 'var{:d}'.format(i)
        count = draw(st.integers(min_value=1, max_value=4))
        minimum = draw(st.integers(min_value=0, max_value=2))

        variables.append(VariableWithCount(name, count, minimum, None))

    return variables

def panicing_cert(draw, now, panic):
    server_name = draw(ts.dns_names())
    offset = timedelta(seconds=draw(
        s.integers(
                min_value=-1000,
                max_value=int(panic.total_seconds()))))
    return (server_name,
            _generate_cert(
                server_name,
                not_valid_before=now + offset - timedelta(seconds=1),
                not_valid_after=now + offset))

def panicing_certs_fixture(draw):
    now = draw(datetimes(min_year=1971, max_year=2030, timezones=[]))
    panic = timedelta(seconds=draw(
        s.integers(min_value=60, max_value=60 * 60 * 24)))
    certs = dict(
        draw(
            s.lists(
                panicing_cert(now, panic),
                min_size=1,
                unique_by=lambda i: i[0])))
    return AcmeFixture(now=now, panic_interval=panic, certs=certs)

def test_validation_transform(integer):
    MonadTransformTester(monad=Validation.success, value=integer).test(run_to_validation_test=False)

    Validation.fail([integer]).to_maybe() == Maybe.nothing()
    Validation.fail([integer]).to_either() == Left([integers])

def test_fold(integer, text, boolean):

    dictionary = {'key': 'value'}

    assert First(text).fold(identity) is text
    assert All(boolean).fold(identity) is boolean
    assert Sum(integers).fold(identity) is integers
    assert Map(dictionary).fold(identity) is dictionary

def test_play_with_queue_size(self):
        self._run_argument_numeric("queue_size", "queue", hst.integers)

def durations(draw):
    num = draw(st.integers(1, 100))
    suffix = draw(st.sampled_from('ymd'))
    return f'{num}{suffix}'

def dataset_shape(draw):
    crs = draw(DatacubeCRSStrategy)
    height = draw(st.integers(10, 20))
    width = draw(st.integers(10, 20))
    ntimes = draw(st.integers(1, 10))
    times = draw(ordered_dates(ntimes))
    return crs, height, width, times

def test_to_byte_array_list_fail_big(self, list_val):
        """
        to_byte_array() with incompatible param:
        :param list_val: random list of integers > 256 -ValueError
        """
        with pytest.raises(ValueError):
            pointer, leng = to_byte_array(list_val)

def test_to_byte_array_list_fail_neg(self, list_val):
        """
        to_byte_array() with incompatible param:
        :param list_val: random list of negative integers. -ValueError
        """
        with pytest.raises(ValueError):
            pointer, leng = to_byte_array(list_val)

def test_split_string_into_list(self, data):
        """
        _split_string_into_list() w/ random text and block size
        :param data:
        """
        txt = data.draw(text(alphabet=ascii, min_size=1))
        block = data.draw(integers(min_value=1, max_value=len(txt)))

        txt_list = [txt[i:i + block] for i in range(0, len(txt), block)]
        assert encrypt._split_string_into_list(txt, block) == txt_list

def applications():
    """Mock of the YARN cluster apps REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'allocatedMB': st.integers(-1),
        'allocatedVCores': st.integers(-1),
        'amContainerLogs': st.text(),
        'amHostHttpAddress': st.text(),
        'applicationTags': st.text(),
        'applicationType': st.sampled_from(['MAPREDUCE', 'SPARK']),
        'clusterId': st.integers(0),
        'diagnostics': st.text(),
        'elapsedTime': st.integers(0),
        'finalStatus': st.sampled_from(['UNDEFINED', 'SUCCEEDED', 'FAILED', 'KILLED']),
        'finishedTime': st.integers(0),
        'id': st.text(string.ascii_letters, min_size=5, max_size=25),
        'memorySeconds': st.integers(0),
        'name': st.text(min_size=5),
        'numAMContainerPreempted': st.integers(0),
        'numNonAMContainerPreempted': st.integers(0),
        'preemptedResourceMB': st.integers(0),
        'preemptedResourceVCores': st.integers(0),
        'progress': st.floats(0, 100),
        'queue': st.text(),
        'runningContainers': st.integers(-1),
        'startedTime': st.integers(0),
        'state': st.sampled_from(['NEW', 'NEW_SAVING', 'SUBMITTED', 'ACCEPTED', 'RUNNING', 'FINISHED', 'FAILED', 'KILLED']),
        'trackingUI': st.text(),
        'trackingUrl': st.just(os.environ['YARN_ENDPOINT']),
        'user': st.text(),
        'vcoreSeconds': st.integers(0)
    })

    result = json.dumps({
        'apps': {
            'app': st.lists(d, min_size=4, average_size=10).example()
        }
    })
    redis.set(request.base_url, result)
    return jsonify(result)

def mapreduce_application():
    """Mock of the mapreduce jobs REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'startTime': st.integers(0),
        'finishTime': st.integers(0),
        'elapsedTime': st.integers(0),
        'id': st.integers(0),
        'name': st.text(),
        'user': st.text(),
        'state': st.sampled_from(['NEW', 'SUCCEEDED', 'RUNNING', 'FAILED', 'KILLED']),
        'mapsTotal': st.integers(0),
        'mapsCompleted': st.integers(0),
        'reducesTotal': st.integers(0),
        'reducesCompleted': st.integers(0),
        'mapProgress': st.floats(0, 100),
        'reduceProgress': st.floats(0, 100),
        'mapsPending': st.integers(0),
        'mapsRunning': st.integers(0),
        'reducesPending': st.integers(0),
        'reducesRunning': st.integers(0),
        'uberized': st.booleans(),
        'diagnostics': st.text(),
        'newReduceAttempts': st.integers(0),
        'runningReduceAttempts': st.integers(0),
        'failedReduceAttempts': st.integers(0),
        'killedReduceAttempts': st.integers(0),
        'successfulReduceAttempts': st.integers(0),
        'newMapAttempts': st.integers(0),
        'runningMapAttempts': st.integers(0),
        'failedMapAttempts': st.integers(0),
        'killedMapAttempts': st.integers(0),
        'successfulMapAttempts': st.integers(0)
    })
    result = json.dumps({
        'jobs': {
            'job': st.lists(d, average_size=3).example()
        }
    })
    redis.set(request.base_url, result)
    return jsonify(result)