1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
# Any copyright is dedicated to the public domain.
# http://creativecommons.org/publicdomain/zero/1.0/
import os
import gecko_taskgraph
import pytest
from gecko_taskgraph.util.verify import verifications
from mozunit import main
from taskgraph.generator import TaskGraphGenerator
from taskgraph.parameters import Parameters
from taskgraph.taskgraph import TaskGraph
from taskgraph.util.taskcluster import get_artifact
def pytest_generate_tests(metafunc):
if "verification" in metafunc.fixturenames:
name = metafunc.function.__name__.split("_", 1)[1]
verification_objs = verifications._verifications.get(name, [])
ids = [v.func.__name__ for v in verification_objs]
metafunc.parametrize("verification", verification_objs, ids=ids)
@pytest.fixture(scope="module")
def parameters():
if "TASK_GROUP_ID" not in os.environ:
pytest.skip(reason="requires a Decision taskId to test against")
return Parameters(
**get_artifact(os.environ["TASK_GROUP_ID"], "public/parameters.yml")
)
@pytest.fixture(scope="module")
def tgg(parameters):
root = os.path.dirname(os.path.dirname(gecko_taskgraph.__file__))
return TaskGraphGenerator(root, parameters)
@pytest.fixture(scope="module")
def graph_config(tgg):
return tgg.graph_config
@pytest.fixture(scope="module")
def kinds(tgg):
return {kind.name: kind for kind in tgg._load_kinds(tgg.graph_config, [])}
@pytest.fixture(scope="module")
def full_task_graph():
if "TASK_GROUP_ID" not in os.environ:
pytest.skip(reason="requires a Decision taskId to test against")
return TaskGraph.from_json(
get_artifact(os.environ["TASK_GROUP_ID"], "public/full-task-graph.json")
)[1]
@pytest.fixture(scope="module")
def target_task_graph():
if "TASK_GROUP_ID" not in os.environ:
pytest.skip(reason="requires a Decision taskId to test against")
return TaskGraph.from_json(
get_artifact(os.environ["TASK_GROUP_ID"], "public/target-tasks.json")
)[1]
@pytest.fixture(scope="module")
def morphed_task_graph():
if "TASK_GROUP_ID" not in os.environ:
pytest.skip(reason="requires a Decision taskId to test against")
return TaskGraph.from_json(
get_artifact(os.environ["TASK_GROUP_ID"], "public/task-graph.json")
)[1]
def test_initial(verification):
verification.verify()
def test_graph_config(verification, graph_config):
verification.verify(graph_config)
def test_kinds(verification, kinds):
verification.verify(kinds)
def test_parameters(verification, parameters):
verification.verify(parameters)
def test_full_task_set(verification, full_task_graph, graph_config, parameters):
# We don't write out the full_task_set as a decision task artifact, but
# the full_task_graph is functionally equivalent.
verification.verify(full_task_graph, graph_config, parameters)
def test_full_task_graph(verification, full_task_graph, graph_config, parameters):
verification.verify(full_task_graph, graph_config, parameters)
def test_target_task_graph(verification, target_task_graph, graph_config, parameters):
verification.verify(target_task_graph, graph_config, parameters)
def test_optimized_task_graph(
verification, morphed_task_graph, graph_config, parameters
):
# We don't write out the optimized graph as a decision task artifact, but
# the morphed graph is a good enough stand-in.
verification.verify(morphed_task_graph, graph_config, parameters)
def test_morphed_task_graph(verification, morphed_task_graph, graph_config, parameters):
verification.verify(morphed_task_graph, graph_config, parameters)
if __name__ == "__main__":
main()
|
