taskcluster/test/test_generate_params.py

   1 import json
   2 import os
   3 import subprocess
   4
   5 import pytest
   6 from gecko_taskgraph import GECKO
   7 from mozunit import main
   8 from taskgraph.taskgraph import TaskGraph
   9
  10 pytestmark = pytest.mark.slow
  11 PARAMS_DIR = os.path.join(GECKO, "taskcluster", "test", "params")
  12
  13
  14 @pytest.fixture(scope="module")
  15 def get_graph_from_spec(tmpdir_factory):
  16     outdir = tmpdir_factory.mktemp("graphs")
  17
  18     # Use a mach subprocess to leverage the auto parallelization of
  19     # parameters when specifying a directory.
  20     cmd = [
  21         "./mach",
  22         "taskgraph",
  23         "morphed",
  24         "--json",
  25         f"--parameters={PARAMS_DIR}",
  26         f"--output-file={outdir}/graph.json",
  27     ]
  28     # unset MOZ_AUTOMATION so we don't attempt to optimize out the graph
  29     # entirely as having already run
  30     env = os.environ.copy()
  31     env.pop("MOZ_AUTOMATION", None)
  32     subprocess.run(cmd, cwd=GECKO, env=env)
  33     assert len(outdir.listdir()) > 0
  34
  35     def inner(param_spec):
  36         outfile = f"{outdir}/graph_{param_spec}.json"
  37         with open(outfile) as fh:
  38             output = fh.read()
  39             try:
  40                 return TaskGraph.from_json(json.loads(output))[1]
  41             except ValueError:
  42                 return output
  43
  44     return inner
  45
  46
  47 @pytest.mark.parametrize(
  48     "param_spec",
  49     [os.path.splitext(p)[0] for p in os.listdir(PARAMS_DIR) if p.endswith(".yml")],
  50 )
  51 def test_generate_graphs(get_graph_from_spec, param_spec):
  52     ret = get_graph_from_spec(param_spec)
  53     if isinstance(ret, str):
  54         print(ret)
  55         pytest.fail("An exception was raised during graph generation!")
  56
  57     assert isinstance(ret, TaskGraph)
  58     assert len(ret.tasks) > 0
  59
  60
  61 if __name__ == "__main__":
  62     main()