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

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 add_bools(list_of_lists):
    """
    Given recursive list that can contain other lists, return tuple of that plus
    a booleans strategy for each list.
    """
    l = []
    def count(recursive):
        l.append(1)
        for child in recursive:
            if isinstance(child, list):
                count(child)
    count(list_of_lists)
    return st.tuples(st.just(list_of_lists), st.tuples(*[st.sampled_from([True, False]) for i in l]))

def simple_classes(defaults=None):
    """
    Return a strategy that yields tuples of simple classes and values to
    instantiate them.
    """
    return lists_of_attrs(defaults).flatmap(_create_hyp_class)


# Ok, so st.recursive works by taking a base strategy (in this case,
# simple_classes) and a special function. This function receives a strategy,
# and returns another strategy (building on top of the base strategy).

def files(draw):
    # TODO: use st.recursive to generate files in folders
    return draw(st.lists(valid_file, average_size=5, max_size=MAX_FILES)\
            .filter(has_no_duplicate))

#valid_vault = vault().filter(has_no_duplicate)

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