graph_test.py

"""Testing dependency graphs"""
import random
import pyDatalog.pyDatalog as pd

from hypothesis import strategies as st
from hypothesis import given

pd.create_terms('follows, is_root, is_command, requires, provides, X, '
                'Y, Z, N, order')

RES_COUNT = 35

range_intagers_st = st.integers(min_value=0, max_value=RES_COUNT)


@st.composite
def provide_require_st(draw, filter_=True):
    commands = draw(range_intagers_st)
    provides = draw(
        st.lists(
            st.lists(range_intagers_st),
            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)))
            else:
                requires.append([])
    else:
        requires = [[]] * commands
    return (provides, requires)


def print_example():
    example = provide_require_st().example()
    print("""
    digraph g {
         label="Command graph";
         graph [splines=line];
    """)
    for i in range(len(example[0])):
        print("    c%03d [shape=triangle];" % i)
        for provides in example[0][i]:
            print("    c%03d -> r%03d;" % (i, provides))
        for requires in example[1][i]:
            print("    r%03d -> c%03d;" % (requires, i))

    print("}")


@given(provide_require_st(), st.random_module())
def test_graph(tree, rnd):
    """Test if the resolver can handle generated graphs"""
    run_graph(tree)


def test_graph_basic():
    """Test if the resolver can handle a simple strait forward graph"""
    tree = [
        [('A'), ('B'), ('C'), ('D'), ('E'), ('F'), ('G'), ()],
        [(), ('A'), ('B'), ('C'), ('D'), ('E'), ('F'), ('G')]
    ]
    run_graph(tree)


def test_graph_multirequire():
    """Test if the resolver can handle a graph with multiple requires"""
    tree = [
        [('A'), ('B'), ('C'), ('D'), ('E'), ('F'), ('G'), ()],
        [(), ('A'), ('B'), ('C', 'A'), ('D'), ('E'), ('F'), ('G')]
    ]
    run_graph(tree)


def test_graph_multiprovide():
    """Test if the resolver can handle a graph with multiple provides"""
    tree = [
        [('A'), ('B'), ('C'), ('D', 'A'), ('E'), ('F'), ('G'), ()],
        [(), ('A'), ('B'), ('C'), ('D'), ('E'), ('F'), ('G')]
    ]
    run_graph(tree)


def run_graph(tree):
    """Run an example"""
    try:
        tree_len = len(tree[0])
        index = list(range(tree_len))
        random.shuffle(index)
        for i in index:
            + is_command(i)
            for provide in tree[0][i]:
                + provides(i, provide)
            for require in tree[1][i]:
                + requires(i, require)
        ##############################
        is_root(X) <= is_command(X) & ~requires(X, Y)
        follows(X, Z) <= (
            provides(X, Y) & requires(Z, Y) & (X != Z)
        )
        order(0, X) <= is_root(X)
        order(N, X) <= (N > 0) & order(N - 1, Y) & follows(Y, X)
        ##############################
        ordered = []
        try:
            for x in range(tree_len):
                nodes = order(x, N)
                if not nodes:
                    break
                ordered.extend([x[0] for x in nodes])
        except AttributeError:
            ordered = index
        assert len(ordered) >= tree_len
        print(ordered)
        provided = set()
        for command in ordered:
            assert set(tree[1][command]).issubset(provided)
            provided.update(tree[0][command])
    finally:
        pd.clear()