tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

conftest.py (4300B)


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