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     subprocess.run(cmd, cwd=GECKO)
  29     assert len(outdir.listdir()) > 0
  30
  31     def inner(param_spec):
  32         outfile = f"{outdir}/graph_{param_spec}.json"
  33         with open(outfile) as fh:
  34             output = fh.read()
  35             try:
  36                 return TaskGraph.from_json(json.loads(output))[1]
  37             except ValueError:
  38                 return output
  39
  40     return inner
  41
  42
  43 @pytest.mark.parametrize(
  44     "param_spec", [os.path.splitext(p)[0] for p in os.listdir(PARAMS_DIR)]
  45 )
  46 def test_generate_graphs(get_graph_from_spec, param_spec):
  47     ret = get_graph_from_spec(param_spec)
  48     if isinstance(ret, str):
  49         print(ret)
  50         pytest.fail("An exception was raised during graph generation!")
  51
  52     assert isinstance(ret, TaskGraph)
  53     assert len(ret.tasks) > 0
  54
  55
  56 if __name__ == "__main__":
  57     main()