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
124
125
|
# Any copyright is dedicated to the public domain.
# http://creativecommons.org/publicdomain/zero/1.0/
import json
import logging
import os
import pytest
from gecko_taskgraph.util.bugbug import BUGBUG_BASE_URL
from gecko_taskgraph.util.hg import PUSHLOG_PUSHES_TMPL
from responses import RequestsMock
from responses import logger as rsps_logger
from taskgraph.generator import TaskGraphGenerator
from taskgraph.parameters import parameters_loader
here = os.path.abspath(os.path.dirname(__file__))
@pytest.fixture(scope="session")
def responses():
rsps_logger.setLevel(logging.WARNING)
with RequestsMock(assert_all_requests_are_fired=False) as rsps:
yield rsps
@pytest.fixture(scope="session")
def datadir():
return os.path.join(here, "data")
@pytest.fixture(scope="session")
def create_tgg(responses, datadir):
def inner(parameters=None, overrides=None):
params = parameters_loader(parameters, strict=False, overrides=overrides)
tgg = TaskGraphGenerator(None, params)
# Mock out certain requests as they may depend on a revision that does
# not exist on hg.mozilla.org.
mock_requests = {}
# bugbug /push/schedules
url = BUGBUG_BASE_URL + "/push/{project}/{head_rev}/schedules".format(
**tgg.parameters
)
mock_requests[url] = "bugbug-push-schedules.json"
# files changed
url = "{head_repository}/json-automationrelevance/{head_rev}".format(
**tgg.parameters
)
mock_requests[url] = "automationrelevance.json"
url = PUSHLOG_PUSHES_TMPL.format(
repository=tgg.parameters["head_repository"],
push_id_start=int(tgg.parameters["pushlog_id"]) - 2,
push_id_end=int(tgg.parameters["pushlog_id"]) - 1,
)
mock_requests[url] = "pushes.json"
for url, filename in mock_requests.items():
with open(os.path.join(datadir, filename)) as fh:
responses.add(
responses.GET,
url,
json=json.load(fh),
status=200,
)
# Still allow other real requests.
responses.add_passthru("https://hg.mozilla.org")
responses.add_passthru("https://firefox-ci-tc.services.mozilla.com")
return tgg
return inner
@pytest.fixture(scope="module")
def tgg(request, create_tgg):
if not hasattr(request.module, "PARAMS"):
pytest.fail("'tgg' fixture requires a module-level 'PARAMS' variable")
tgg = create_tgg(overrides=request.module.PARAMS)
return tgg
@pytest.fixture(scope="module")
def params(tgg):
return tgg.parameters
@pytest.fixture(scope="module")
def full_task_graph(tgg):
return tgg.full_task_graph
@pytest.fixture(scope="module")
def optimized_task_graph(full_task_graph, tgg):
return tgg.optimized_task_graph
@pytest.fixture(scope="session")
def filter_tasks():
def inner(graph, func):
return filter(func, graph.tasks.values())
return inner
@pytest.fixture(scope="session")
def print_dependents():
def inner(graph, label, indent=""):
if indent == "":
print(f"Dependent graph for {label}:")
dependents = set()
for task in graph.tasks.values():
if label in task.dependencies.values():
dependents.add(task.label)
print(f"{indent}{label}")
if dependents:
for dep in sorted(dependents):
inner(graph, dep, indent=indent + " ")
return inner
|
