def _create_hyp_class(attrs_and_strategy):
"""
A helper function for Hypothesis to generate attrs classes.
The result is a tuple: an attrs class, and a tuple of values to
instantiate it.
"""
def key(t):
return t[0].default is not NOTHING
attrs_and_strat = sorted(attrs_and_strategy, key=key)
attrs = [a[0] for a in attrs_and_strat]
for i, a in enumerate(attrs):
a.counter = i
vals = tuple((a[1]) for a in attrs_and_strat)
return st.tuples(
st.just(make_class('HypClass',
OrderedDict(zip(gen_attr_names(), attrs)))),
st.tuples(*vals))
python类just()的实例源码
def _create_hyp_nested_strategy(simple_class_strategy):
"""
Create a recursive attrs class.
Given a strategy for building (simpler) classes, create and return
a strategy for building classes that have as an attribute:
* just the simpler class
* a list of simpler classes
* a dict mapping the string "cls" to a simpler class.
"""
# A strategy producing tuples of the form ([list of attributes], <given
# class strategy>).
attrs_and_classes = st.tuples(lists_of_attrs(defaults=True),
simple_class_strategy)
return (attrs_and_classes.flatmap(just_class) |
attrs_and_classes.flatmap(list_of_class) |
attrs_and_classes.flatmap(dict_of_class))
def pattern_to_statements(pattern):
if isinstance(pattern, template):
return lists(just(pattern), min_size=1, max_size=1)
rule, value = pattern
if rule == 'sequence':
return tuples(*map(pattern_to_statements, value)).map(unpack_list).map(list)
elif rule == 'alternates':
return one_of(*map(pattern_to_statements, value))
elif rule == 'zeroOrMore':
return lists(pattern_to_statements(value)).map(unpack_list).map(list)
elif rule == 'oneOrMore':
return lists(pattern_to_statements(value), min_size=1).map(unpack_list).map(list)
elif rule == 'optional':
return lists(pattern_to_statements(value), min_size=0, max_size=1).map(unpack_list).map(list)
else:
raise Exception("impossible!", rule)
# this replicates the current scorm pattern, a realistic example of medium
# complexity. Note it has repeated elements, just not in ambiguous ways.
def test_overrides_respected():
"""Ensure provided overrides are respected."""
protobuf_strategies = modules_to_strategies(im_pb2, **{
full_field_name(im_pb2.InstantMessage, 'message'): st.just('test message')
})
instant_message_strategy = protobuf_strategies[im_pb2.InstantMessage]
instant_message_example = instant_message_strategy.example()
assert instant_message_example.message == 'test message'
def random_references(draw):
'''generates random sorted lists of references with the same prefix like ['IASDHAH1', 'IASDHAH1569', ...]'''
prefix = st.just(draw(random_prefix()))
parts = draw(st.lists(random_reference(prefix = prefix)))
parts.sort()
return list(map(toRef, parts))
def just_class(tup):
nested_cl = tup[1][0]
default = attr.Factory(nested_cl)
combined_attrs = list(tup[0])
combined_attrs.append((attr.ib(default=default), st.just(nested_cl())))
return _create_hyp_class(combined_attrs)
def list_of_class(tup):
nested_cl = tup[1][0]
default = attr.Factory(lambda: [nested_cl()])
combined_attrs = list(tup[0])
combined_attrs.append((attr.ib(default=default),
st.just([nested_cl()])))
return _create_hyp_class(combined_attrs)
def dict_of_class(tup):
nested_cl = tup[1][0]
default = attr.Factory(lambda: {"cls": nested_cl()})
combined_attrs = list(tup[0])
combined_attrs.append((attr.ib(default=default),
st.just({'cls': nested_cl()})))
return _create_hyp_class(combined_attrs)
def strategy(self):
'Returns resulting strategy that generates configured char set'
max_codepoint = None if self._unicode else 127
strategies = []
if self._negate:
if self._categories or self._whitelist_chars:
strategies.append(
hs.characters(
blacklist_categories=self._categories | set(['Cc', 'Cs']),
blacklist_characters=self._whitelist_chars,
max_codepoint=max_codepoint,
)
)
if self._blacklist_chars:
strategies.append(
hs.sampled_from(
list(self._blacklist_chars - self._whitelist_chars)
)
)
else:
if self._categories or self._blacklist_chars:
strategies.append(
hs.characters(
whitelist_categories=self._categories,
blacklist_characters=self._blacklist_chars,
max_codepoint=max_codepoint,
)
)
if self._whitelist_chars:
strategies.append(
hs.sampled_from(
list(self._whitelist_chars - self._blacklist_chars)
)
)
return hs.one_of(*strategies) if strategies else hs.just(u'')
def test_agents(agent_type, attributes):
agents = testing.agents(size_strategy=st.just(10),
agent_type=agent_type,
attributes=attributes)
assert agents.example().shape == (10,)
def comprehension_node(draw, target=None, iter=None,
ifs=hs.just([])):
target = target or const_node(valid_identifier())
iter = iter or list_node()
node = astroid.Comprehension()
node.postinit(draw(target), draw(iter), draw(ifs))
return node
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 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)
