Test pipeline changes before producing a candidate

Run the application’s failure and integration checks from one command that can also serve as a repository CI job.

Problem

When you need this. A change to arrival-time filtering can leak future events even when fitting still completes. A change to packaging can leave a model that no prediction process can load. The pipeline therefore needs checks for its data and artifact contracts, not just a successful training command.

This workshop combines the existing tests with the new local platform integrations. A single command returns a machine-readable result and a nonzero exit status when a check fails.

Concepts

Continuous integration runs checks when code changes. It answers whether the change satisfies the checks included in that job. Model evaluation asks how a candidate performs on an intended population; passing unit tests cannot answer that question.

The suite covers strict JSON and event availability, temporal splits, immutable identities, actual MLflow artifacts, Prefect retry behavior, registry selection and prediction validation. Temporary local databases keep one test run from depending on old state.

The included ci.yml is an optional GitHub Actions workflow for a repository containing the extracted lab at its root. It installs the same pinned environment and calls the same checker. The local command is executed here; the hosted workflow has not been dispatched.

Architecture

Working code

Download the shared application and tests, unzip it and enter mlops-return-workflow/. Keep the same directory throughout the course. Create the environment once. Install the pinned platform dependencies for this workshop.

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-platform.txt

On PowerShell, activate with .venv\Scripts\Activate.ps1.

Continue from Retry a failed step without repeating the experiment; keep the same lab directory and outputs.

Make check results usable by automation

The checker records test counts and returns the test runner’s success status. Failed tests remain visible in the text output.

def check():
    names=['test_workflow','test_foundations','test_tracking','test_platform']
    suite=unittest.defaultTestLoader.loadTestsFromNames(names)
    stream=io.StringIO();result=unittest.TextTestRunner(stream=stream,verbosity=2).run(suite)
    print(stream.getvalue(),file=sys.stderr)
    return {'command':'python ci.py','tests_run':result.testsRun,'failures':len(result.failures),
            'errors':len(result.errors),'skipped':len(result.skipped),'passed':result.wasSuccessful(),
            'python':sys.version.split()[0],'packages':{p:importlib.metadata.version(p) for p in ['mlflow-skinny','prefect']},
            'execution':'actual local tests; hosted GitHub Actions job not executed'}

def main():
    p=argparse.ArgumentParser();p.add_argument('--out',type=Path,required=True);a=p.parse_args()
    result=check();w.save(a.out,result);print(json.dumps(result,indent=2));sys.exit(0 if result['passed'] else 1)


if __name__=='__main__':main()

python course.py ci-checks --workspace work/course

{
  "command": "python ci.py",
  "tests_run": 55,
  "failures": 0,
  "errors": 0,
  "skipped": 0,
  "passed": true,
  "python": "3.12.8",
  "packages": {
    "mlflow-skinny": "3.17.2",
    "prefect": "3.9.0"
  },
  "execution": "actual local tests; hosted GitHub Actions job not executed"
}

The JSON result records the actual installed dependency versions as run metadata. It reports what this local test process executed, without claiming a hosted CI run.

Keep independent checks around the integration

The cohort and registry tests deliberately violate their contracts. They check effects such as an absent candidate or an unchanged selected version, rather than merely mirroring a function’s return value.

def ci_report(root):
    import ci
    need(root,'retry-state.json');result=ci.check()
    if not result['passed']:raise ValueError('local CI checks failed; inspect raw test output')
    return result

Read the named failure tests in test_platform.py alongside the earlier test_workflow.py and test_foundations.py. The downloaded lab contains all of them.

Recorded commands retain outputs in recorded-platform/; your commands use work/. Read the actual command records. Use a fresh work directory for another attempt; original outputs are preserved.

Failure cases

A green job that never discovers tests is not useful. The retained result includes tests_run, failures, errors and skipped; inspect these fields when changing discovery.

The suite deliberately produces failed Prefect and MLflow runs to verify their boundaries. Their expected failure logs do not mean the overall suite failed; the final test summary determines its exit status.

python -m unittest -v test_platform.py

Running and scaling the workshop

Use python ci.py –out work/ci-result.json for the equivalent direct check. Put ci.yml under .github/workflows/ only when adopting this lab as its own repository. Hosted execution is optional and unverified here.

Keep remote credentials out of this job. It uses synthetic data, local CPU fitting and temporary stores. Larger projects can split fast contract tests from slower integrations while retaining both before release.

Key takeaways

• CI checks code and contracts; candidate gates check evaluated artifacts.
• A useful job preserves diagnostics and propagates failures.
• A workflow file does not show that a hosted job ran.

Best practices

Test time rules and artifact loading, not only the training return code.

Keep the same entry point for local and repository checks.

References

• Prefect: flows and tasks
• Google: MLOps lifecycle and automation