test_examples.py 2.02 KB
Newer Older
Wout De Nolf's avatar
Wout De Nolf committed
1
2
3
4
5
import pytest
import itertools
from .examples.graphs import graph_names
from .examples.graphs import get_graph
from .utils import assert_taskgraph_result
6
from .utils import assert_workflow_result
Wout De Nolf's avatar
Wout De Nolf committed
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from ewokscore import load_graph


@pytest.mark.parametrize(
    "graph_name,persist", itertools.product(graph_names(), (True, False))
)
def test_execute_graph(graph_name, persist, tmpdir):
    g, expected = get_graph(graph_name)
    ewoksgraph = load_graph(g)
    if persist:
        varinfo = {"root_uri": str(tmpdir)}
    else:
        varinfo = None
    if ewoksgraph.is_cyclic or ewoksgraph.has_conditional_links:
        with pytest.raises(RuntimeError):
            ewoksgraph.execute(varinfo=varinfo)
    else:
24
        tasks = ewoksgraph.execute(varinfo=varinfo, results_of_all_nodes=True)
Wout De Nolf's avatar
Wout De Nolf committed
25
26
27
28
        assert_taskgraph_result(ewoksgraph, expected, tasks=tasks)
        if persist:
            assert_taskgraph_result(ewoksgraph, expected, varinfo=varinfo)

29
30
31
32
33
34
        end_tasks = ewoksgraph.execute(varinfo=varinfo, results_of_all_nodes=False)
        end_nodes = ewoksgraph.end_nodes()
        assert end_tasks.keys() == end_nodes
        expected = {k: v for k, v in expected.items() if k in end_nodes}
        assert_workflow_result(end_tasks, expected, varinfo=varinfo)

Wout De Nolf's avatar
Wout De Nolf committed
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

def test_graph_cyclic():
    g, _ = get_graph("empty")
    taskgraph = load_graph(g)
    assert not taskgraph.is_cyclic
    g, _ = get_graph("acyclic1")
    taskgraph = load_graph(g)
    assert not taskgraph.is_cyclic
    g, _ = get_graph("cyclic1")
    taskgraph = load_graph(g)
    assert taskgraph.is_cyclic


def test_start_nodes():
    g, _ = get_graph("acyclic1")
    taskgraph = load_graph(g)
    assert taskgraph.start_nodes() == {"task1", "task2"}

    g, _ = get_graph("acyclic2")
    taskgraph = load_graph(g)
    assert taskgraph.start_nodes() == {"task1"}

    g, _ = get_graph("cyclic1")
    taskgraph = load_graph(g)
    assert taskgraph.start_nodes() == {"task1"}

    g, _ = get_graph("triangle1")
    taskgraph = load_graph(g)
    assert taskgraph.start_nodes() == {"task1"}