Testing YAML Driven Jobs With pytest and Hypothesis

A pipeline described in YAML is easy to change, and so it changes often and often by people who do not run it first. The cheapest protection is a test that runs on every pull request, takes under a second and breaks when the file is wrong. The trick is knowing what to test, since a runner that starts a database, shells out and sleeps is slow and flaky to exercise as a whole. This post separates the job into the part that is deterministic (reading the spec and planning the order) and the part that is not (running the steps), tests the first thoroughly and gives the second a thin test of its own. It belongs to the cluster on YAML in data engineering workflows, and extends the runner from declaring pipelines in YAML for data jobs.

Make the job testable first

The first move is structural. A runner that reads a file, validates it, orders the steps and runs each one in a single function can only be tested by running it. Split it:

from graphlib import CycleError, TopologicalSorter

def plan(spec):
    steps = {s["id"]: s for s in spec["steps"]}
    ids = [s["id"] for s in spec["steps"]]
    if len(ids) != len(set(ids)):
        raise ValueError("duplicate step id")
    for s in steps.values():
        for d in s.get("needs", []):
            if d not in steps:
                raise ValueError(f"step {s['id']!r} needs unknown step {d!r}")
    try:
        return list(TopologicalSorter({i: s.get("needs", []) for i, s in steps.items()}).static_order())
    except CycleError as e:
        raise ValueError("cycle: " + " -> ".join(e.args[1]))

plan takes a dictionary, returns a list of step ids or raises ValueError with a message that names the problem. It touches no files, no clock and no network. That purity is the property you are buying: the tests are fast, they cannot fail because a port is in use, and a failure points at the logic. Loading the YAML (yaml.safe_load) and executing the steps stay outside, in a thin main that is tested separately. Note that the duplicate check runs on the list, before the dictionary is built, because a dictionary built from a list with duplicate ids silently keeps the last one, the same trap as duplicate keys in YAML.

Example tests

Write the specs inline as YAML strings. It keeps each test readable by someone who only knows the file format, and it exercises the same parse the runner does:

def spec(text):
    return yaml.safe_load(text)

def test_order_respects_needs():
    order = plan(spec("steps:\n  - {id: a, run: x}\n  - {id: b, needs: [a], run: y}\n  - {id: c, needs: [a, b], run: z}\n"))
    assert order.index("a") < order.index("b") < order.index("c")

def test_unknown_dependency():
    with pytest.raises(ValueError, match="unknown step 'joins'"):
        plan(spec("steps:\n  - {id: join, run: x}\n  - {id: report, needs: [joins], run: y}\n"))

def test_cycle_reported():
    with pytest.raises(ValueError, match="cycle"):
        plan(spec("steps:\n  - {id: a, needs: [b], run: x}\n  - {id: b, needs: [a], run: y}\n"))

def test_duplicate_ids_rejected():
    with pytest.raises(ValueError, match="duplicate"):
        plan(spec("steps:\n  - {id: a, run: x}\n  - {id: a, run: y}\n"))

Each test pins one rule. The order test asserts relative positions and not the whole list, because a graph with independent branches has several valid orders and a test that compares against one of them breaks when the sorting library changes its tie-breaking. The unknown dependency test uses a typo (joins for join) because that is how the failure happens in real files, and it asserts the message contains the typo, since an error that does not name the culprit makes the author search a long file. The test names read as the rules of the format; a new teammate can learn the constraints from the test file.

A property over random graphs

Examples cover what you thought of. A property test covers shapes you did not. Hypothesis generates many random graphs and checks one statement that must hold for all of them:

dag = st.integers(min_value=1, max_value=15).flatmap(
    lambda n: st.tuples(*[st.lists(st.integers(0, i - 1), max_size=3, unique=True) if i else st.just([]) for i in range(n)]))

@settings(max_examples=200, deadline=None)
@given(dag)
def test_any_dag_plans_in_valid_order(deps):
    s = {"steps": [{"id": f"s{i}", "run": "x", "needs": [f"s{j}" for j in d]} for i, d in enumerate(deps)]}
    order = plan(s)
    pos = {name: k for k, name in enumerate(order)}
    assert len(order) == len(deps)
    for step in s["steps"]:
        for need in step["needs"]:
            assert pos[need] < pos[step["id"]]

The strategy builds graphs that are acyclic by construction: step i may depend only on steps with a lower number, up to three of them. The assertions say the plan contains every step once and puts every dependency first. Two hundred random graphs passed, and the whole file ran in 0.28 seconds with the four examples.

My first version of this strategy was wrong: it raised InvalidArgument inside Hypothesis because a step with nothing earlier to depend on was given an empty choice, and the test failed before checking anything. That is worth knowing about property tests. A failure may be in the generator, not the code under test, so read the traceback to see which one it is before you change plan. It is also a reason to keep the strategy short and obvious.

Do the tests bite

A test that cannot fail is decoration. The check is to break the code on purpose and watch for red. Two mutations, each removing a validation:

skip unknown-need check -> 1 failed, 4 passed in 0.28s
skip duplicate check -> 1 failed, 4 passed in 0.26s

Each removal fails exactly one test, the one named for the rule. That is the result you want: one rule, one test, nothing redundant and nothing missing. A mutation that leaves all tests green would show a rule nobody tests. Tools like mutmut automate this for a whole project, and a manual pass on the three or four rules that matter costs a few minutes.

The parts around the plan

The plan is the part worth the most tests, and it is not the whole job. Three layers remain:

• The loader and the schema. A test that loads the real pipeline files from the repository and runs plan on each catches an actual typo in an actual file, which is the failure that matters. Add a JSON Schema check for types and required keys (validating YAML with JSON Schema, testing config loading in CI).

• The handlers. Each step's code is an ordinary function. Test it with ordinary fixtures; a handler that loads a table is tested against a throwaway database, as in YAML fixtures for database tests.

• One end to end run. A single test that runs a small real pipeline against a scratch database and checks the output proves the pieces connect. Keep it to one or two, because it is the slow one.

Run the first two groups on every commit and the end to end test on every merge, or whenever the runner changes.

Keep the tests close to the format

When someone adds a feature to the format (a new key such as retries, a condition), the test file should grow with a case for it before the change merges. Reject unknown keys in the schema so a typo like need: for needs fails the load and not silently disappears into a step with no dependency. That typo is the most likely mistake in this format, and it deserves an explicit test that feeds a step with need: to the schema check and asserts it reports an error.

Without the schema, need is ignored, the step runs too early, and the failure appears at 3 a.m. as a missing table.

When tests on a YAML driven job give false comfort

The plan is right and the step is wrong. Ordering tests prove the sequence, not that load_table loads the right rows. A green plan suite next to untested handlers is the most common false comfort.

The tests use friendly specs. If every test file is a tidy three-step chain, the suite says little about the real 400-step file. Run plan over the production specs in CI.

Environment differences. A spec that plans correctly can still fail where a variable is unset or a path differs. That is for the end to end run and the deploy checks.

Time and external state. If steps depend on the date or a remote service, pass them in as parameters so tests control them, or the suite becomes flaky and people stop trusting it.

Hypothesis found nothing. A property that passes 200 times shows no counterexample among the shapes the strategy builds. It does not prove the code, and a strategy that never builds a cycle cannot test cycle handling; that is why the cycle case is an example.

The libraries are documented in the pytest documentation and the Hypothesis documentation. Check a spec's syntax in the YAML formatter.