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)
python类integers()的实例源码
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)