nice graph tests

strategies.py

"""
Hypothesis strategies for testing ClusterHQ jira tools.
"""

import random

import networkx

from hypothesis.strategies import (
    dictionaries,
    lists,
    frozensets,
    sampled_from,
    text,
)


def _digraphs_without_data(nodes):
    """
    Generate a graph given nodes.

    'Graph' is a list of of tuples from node to node.
    """

    def flatten_graph_dict(d):
        """Map k [v] -> [(k, v)]"""
        output = []
        for k in d:
            for v in d[k]:
                output.append((k, v))
        return output

    return (
        dictionaries(sampled_from(nodes), frozensets(sampled_from(nodes)))
        .map(flatten_graph_dict)
    )


def fixed_lists(elements, size):
    """
    Generate a list of the given size, with members drawn from the 'elements'
    strategy.
    """
    return lists(elements, min_size=size, max_size=size)


def annotate_sequence(sequence, annotations):
    """
    Given a sequence and a strategy for generating annotations, return a
    strategy that generates a list of tuples of sequence members and
    annotations.

    e.g.

    >>> annotate_sequence(['foo', 'bar', 'baz'], integers()).example()
    [('foo', 4), ('bar', 0), ('baz', -1)]
    """
    return fixed_lists(annotations, len(sequence)).map(lambda notes: zip(sequence, notes))


# Strategy for generating arbitrary data associated with a graph edge.
#
# The idea is to have a small dictionary mapping keywords to some other data
# that we don't care about very much.
edge_data = dictionaries(keys=text(), values=text(), average_size=3)


def _arbitrary_digraphs(nodes):
    """
    Generate arbitrary directed graphs, based on the given nodes.
    """
    return (
        _digraphs_without_data(nodes)
        .flatmap(lambda edges: annotate_sequence(edges, edge_data))
        .map(lambda xs: networkx.DiGraph([(a, b, c) for ((a, b), c) in xs])))


# More complex strategy for generating arbitrary digraphs.
arbitrary_digraphs = _arbitrary_digraphs(list(range(10)))


# Always-true predicate
true = lambda _: True


# Always-false predicate
false = lambda _: False


def randomp(_):
    return random.choice([True, False])


def even_edge_keys(data):
    return len(data) % 2


# Example predicate functions to be used in our tests.
example_predicates = [
    true,
    false,
    randomp,
    even_edge_keys,
]


# Generate an arbitrary predicate, suitable for filtering graph edge data.
#
# That is, a predicate on an arbitrary dictionary.
predicates = sampled_from(example_predicates)

test_blockers.py

"""Tests for blocker code."""


from hypothesis import assume, given
import networkx
from testtools import TestCase
from testtools.matchers import (
    AfterPreprocessing,
    AllMatch,
    Equals,
    Matcher,
    MatchesAll,
    MatchesPredicate,
    Mismatch,
)

from ..blockers import filter_edges

from .strategies import arbitrary_digraphs, predicates, randomp, true, false


class SubsetOf(Matcher):

    def __init__(self, superset):
        self._superset = superset

    def match(self, subset):
        if not subset.issubset(self._superset):
            return Mismatch(
                '%s is not subset of %s, extra: %s'
                % (subset, self._superset, subset - self._superset)
            )


def Subgraph(supergraph):
    return MatchesAll(
        AfterPreprocessing(lambda x: set(x.nodes()), SubsetOf(set(supergraph.nodes()))),
        AfterPreprocessing(lambda x: set(x.edges()), SubsetOf(set(supergraph.edges()))),
    )


def EqualsGraph(graph):
    return AfterPreprocessing(lambda x: sorted(x.edges()), Equals(sorted(graph.edges())))


class TestFilterEdges(TestCase):

    @given(arbitrary_digraphs)
    def test_all(self, graph):
        """An always-true predicate returns the original graph."""
        new_graph = filter_edges(true, graph)
        self.assertThat(new_graph, EqualsGraph(graph))

    @given(arbitrary_digraphs)
    def test_none(self, graph):
        """An always-false predicate returns an empty graph."""
        new_graph = filter_edges(false, graph)
        self.assertThat(new_graph, EqualsGraph(networkx.DiGraph()))

    @given(predicates, arbitrary_digraphs)
    def test_subgraph(self, predicate, graph):
        """Regardless of predicate, the new graph is a sub-graph of the old."""
        new_graph = filter_edges(predicate, graph)
        self.assertThat(new_graph, Subgraph(graph))

    @given(predicates, arbitrary_digraphs)
    def test_predicate_satisfied(self, predicate, graph):
        """The predicate is satisfied for all edge data."""
        assume(predicate is not randomp)
        new_graph = filter_edges(predicate, graph)
        data = []
        for src in new_graph:
            for dst in new_graph[src]:
                data.append(new_graph[src][dst])
        predicate_matcher = MatchesPredicate(predicate, 'p(%s) does not hold')
        self.assertThat(data, AllMatch(predicate_matcher))