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

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))

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)