taskcluster/test/conftest.py

   1 # Any copyright is dedicated to the public domain.
   2 # http://creativecommons.org/publicdomain/zero/1.0/
   3
   4 import json
   5 import logging
   6 import os
   7
   8 import pytest
   9 from gecko_taskgraph.util.bugbug import BUGBUG_BASE_URL
  10 from gecko_taskgraph.util.hg import PUSHLOG_PUSHES_TMPL
  11 from responses import RequestsMock
  12 from responses import logger as rsps_logger
  13 from taskgraph.generator import TaskGraphGenerator
  14 from taskgraph.parameters import parameters_loader
  15
  16 here = os.path.abspath(os.path.dirname(__file__))
  17
  18
  19 @pytest.fixture(scope="session")
  20 def responses():
  21     rsps_logger.setLevel(logging.WARNING)
  22     with RequestsMock(assert_all_requests_are_fired=False) as rsps:
  23         yield rsps
  24
  25
  26 @pytest.fixture(scope="session")
  27 def datadir():
  28     return os.path.join(here, "data")
  29
  30
  31 @pytest.fixture(scope="session")
  32 def create_tgg(responses, datadir):
  33     def inner(parameters=None, overrides=None):
  34         params = parameters_loader(parameters, strict=False, overrides=overrides)
  35         tgg = TaskGraphGenerator(None, params)
  36
  37         # Mock out certain requests as they may depend on a revision that does
  38         # not exist on hg.mozilla.org.
  39         mock_requests = {}
  40
  41         # bugbug /push/schedules
  42         url = BUGBUG_BASE_URL + "/push/{project}/{head_rev}/schedules".format(
  43             **tgg.parameters
  44         )
  45         mock_requests[url] = "bugbug-push-schedules.json"
  46
  47         # files changed
  48         url = "{head_repository}/json-automationrelevance/{head_rev}".format(
  49             **tgg.parameters
  50         )
  51         mock_requests[url] = "automationrelevance.json"
  52
  53         url = PUSHLOG_PUSHES_TMPL.format(
  54             repository=tgg.parameters["head_repository"],
  55             push_id_start=int(tgg.parameters["pushlog_id"]) - 2,
  56             push_id_end=int(tgg.parameters["pushlog_id"]) - 1,
  57         )
  58         mock_requests[url] = "pushes.json"
  59
  60         for url, filename in mock_requests.items():
  61             with open(os.path.join(datadir, filename)) as fh:
  62                 responses.add(
  63                     responses.GET,
  64                     url,
  65                     json=json.load(fh),
  66                     status=200,
  67                 )
  68
  69         # Still allow other real requests.
  70         responses.add_passthru("https://hg.mozilla.org")
  71         responses.add_passthru("https://firefox-ci-tc.services.mozilla.com")
  72         return tgg
  73
  74     return inner
  75
  76
  77 @pytest.fixture(scope="module")
  78 def tgg(request, create_tgg):
  79     if not hasattr(request.module, "PARAMS"):
  80         pytest.fail("'tgg' fixture requires a module-level 'PARAMS' variable")
  81
  82     tgg = create_tgg(overrides=request.module.PARAMS)
  83     return tgg
  84
  85
  86 @pytest.fixture(scope="module")
  87 def tgg_new_config(request, create_tgg):
  88     if not hasattr(request.module, "PARAMS_NEW_CONFIG"):
  89         pytest.fail(
  90             "'tgg_new_config' fixture requires a module-level 'PARAMS' variable"
  91         )
  92
  93     tgg = create_tgg(overrides=request.module.PARAMS_NEW_CONFIG)
  94     return tgg
  95
  96
  97 @pytest.fixture(scope="module")
  98 def params(tgg):
  99     return tgg.parameters
 100
 101
 102 @pytest.fixture(scope="module")
 103 def full_task_graph(tgg):
 104     return tgg.full_task_graph
 105
 106
 107 @pytest.fixture(scope="module")
 108 def optimized_task_graph(full_task_graph, tgg):
 109     return tgg.optimized_task_graph
 110
 111
 112 @pytest.fixture(scope="module")
 113 def full_task_graph_new_config(tgg_new_config):
 114     return tgg_new_config.full_task_graph
 115
 116
 117 @pytest.fixture(scope="module")
 118 def optimized_task_graph_new_config(full_task_graph, tgg_new_config):
 119     return tgg_new_config.optimized_task_graph
 120
 121
 122 @pytest.fixture(scope="session")
 123 def filter_tasks():
 124     def inner(graph, func):
 125         return filter(func, graph.tasks.values())
 126
 127     return inner
 128
 129
 130 @pytest.fixture(scope="session")
 131 def print_dependents():
 132     def inner(graph, label, indent=""):
 133         if indent == "":
 134             print(f"Dependent graph for {label}:")
 135
 136         dependents = set()
 137         for task in graph.tasks.values():
 138             if label in task.dependencies.values():
 139                 dependents.add(task.label)
 140
 141         print(f"{indent}{label}")
 142         if dependents:
 143             for dep in sorted(dependents):
 144                 inner(graph, dep, indent=indent + "  ")
 145
 146     return inner