def setcomp_node(draw, elt=const_node(),
generators=hs.lists(comprehension_node(),
min_size=1, average_size=1)):
node = astroid.SetComp()
node.postinit(draw(elt), draw(generators))
return node
python类lists()的实例源码
def list_node(draw, elt=const_node(), **kwargs):
"""Return a List node with elements drawn from elt.
"""
node = astroid.List()
node.postinit(draw(hs.lists(elt, **kwargs)))
return node
def tuple_node(draw, elt=const_node, **kwargs):
"""Return a Tuple node with elements drawn from elt.
"""
elts = draw(hs.lists(elt(), **kwargs))
node = astroid.Tuple()
node.postinit(elts)
return node
def func_wrap_strategy(args, func):
min_size = func.arity[0]
max_size = func.arity[1] and func.arity[0] or 4
return st.lists(args, min_size=min_size, max_size=max_size).map(lambda a: func(*a))
def func_wrap_strategy(args, func):
min_size = func.arity[0]
max_size = func.arity[1] and func.arity[0] or 4
return st.lists(args, min_size=min_size, max_size=max_size).map(lambda a: func(*a))
def dns_names():
"""
Strategy for generating limited charset DNS names.
"""
return (
s.lists(dns_labels(), min_size=1, max_size=10)
.map(u'.'.join))
def urls():
"""
Strategy for generating ``twisted.python.url.URL``\s.
"""
return s.builds(
URL,
scheme=s.just(u'https'),
host=dns_names(),
path=s.lists(s.text(
max_size=64,
alphabet=s.characters(blacklist_characters=u'/?#',
blacklist_categories=('Cs',))
), min_size=1, max_size=10))
def pem_objects(draw):
"""
Strategy for generating ``pem`` objects.
"""
key = RSAPrivateKey((
b'-----BEGIN RSA PRIVATE KEY-----\n' +
encodebytes(draw(s.binary(min_size=1))) +
b'-----END RSA PRIVATE KEY-----\n'))
return [key] + [
Certificate((
b'-----BEGIN CERTIFICATE-----\n' +
encodebytes(cert) +
b'-----END CERTIFICATE-----\n'))
for cert in draw(s.lists(s.binary(min_size=1), min_size=1))]
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_unit_interval(self):
"""lists of random values in [0, 1] chosen. Minimum distance cannot be greater than 1."""
n = numpy.random.rand(5, 3)
m = numpy.random.rand(10, 3)
min_dist = isambard.geometry.closest_distance(points1=n, points2=m)
self.assertLessEqual(min_dist, 1.0)
def ordered_dates(num):
return st.lists(st.datetimes(datetime(1970, 1, 1), datetime(2050, 1, 1)),
min_size=num, max_size=num)
def provide_require_st(draw, filter_=True): # pragma: no cover
commands = draw(range_intagers_st)
provides = draw(
st.lists(
st.lists(range_intagers_st, max_size=10),
min_size = commands,
max_size = commands
),
)
is_func = draw(
st.lists(
st.booleans(),
min_size = commands,
max_size = commands
)
)
provides_set = set()
for command in provides:
provides_set.update(command)
requires = []
if provides_set:
for command in provides:
if command:
max_prov = max(command)
else:
max_prov = 0
if filter_:
provides_filter = [x for x in provides_set if x > max_prov]
else:
provides_filter = provides_set
if provides_filter:
sample = st.sampled_from(provides_filter)
requires.append(draw(st.lists(sample, max_size=10)))
else:
requires.append([])
else:
requires = [[]] * commands
return (provides, requires, is_func)
def test_nestedInteractions(self, values):
"""
Nested interactions operate independently of parent interactions.
:param values: a two-tuple composed of:
- a recursive list of unicode and other recursive lists - list start
means begin interaction, string means node resolve, list end means
finish interaction.
- list of False/True; True means failed interaction
"""
requested_interactions, failures = values
failures = iter(failures)
assume(not isinstance(requested_interactions, unicode))
self.init()
ws_actor = self.connector.expectSocket()
self.connector.connect(ws_actor)
failures = iter(failures)
created_services = {}
expected_success_nodes = Counter()
expected_failed_nodes = Counter()
def run_interaction(children):
should_fail = next(failures)
failed = []
succeeded = []
self.session.start_interaction()
for child in children:
if isinstance(child, unicode):
# Make sure disco knows about the node:
if child in created_services:
node = created_services[child]
else:
node = create_node(child, child)
created_services[child] = node
self.disco.onMessage(None, NodeActive(node))
# Make sure the child Node is resolved in the interaction
self.session.resolve(node.service, "1.0")
if should_fail:
expected_failed_nodes[node] += 1
failed.append(node)
else:
expected_success_nodes[node] += 1
succeeded.append(node)
else:
run_interaction(child)
if should_fail:
self.session.fail_interaction("OHNO")
self.session.finish_interaction()
self.connector.advance_time(5.0) # Make sure interaction is sent
ws_actor.swallowLogMessages()
self.connector.expectInteraction(
self, ws_actor, self.session, failed, succeeded)
run_interaction(requested_interactions)
for node in set(expected_failed_nodes) | set(expected_success_nodes):
policy = self.disco.failurePolicy(node)
self.assertEqual((policy.successes, policy.failures),
(expected_success_nodes[node],
expected_failed_nodes[node]))
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)
