Test a reporting pipeline with failure injection

A successful run proves that the happy path worked for one input. Failure injection asks whether the workflow preserves its guarantees when something breaks at a specific point: before validation, during extraction, halfway through output or just before a completed artifact becomes visible.

Choose failures that challenge the contract. A test that only repeats the implementation's arithmetic may miss the operational behavior the report owner depends on.

Identify the boundaries that matter

The original automation lab has source validation, event selection, reconciliation, staged output and a final bundle rename. Its API extension separately handles page reads, retries and collection completeness.

Useful failures include a truncated source, conflicting event ID, stale watermark, changed API snapshot, corrupted retained output and interruption before bundle completion. Each should have an expected observable result.

For staged output, the key guarantee is that a failed preparation does not expose a directory that downstream readers can mistake for a complete report.

Inject a failure immediately before completion

import json
from pathlib import Path
from tempfile import TemporaryDirectory
from pipeline import run,ReportError,verify_bundle

config = json.loads(Path('report-config.json').read_text())
with TemporaryDirectory(prefix='failure-injection-example-') as temporary:
    output = Path(temporary)/'runs'
    try:
        run('events.csv','source-manifest.json',config,output,fail_after='before_commit')
    except ReportError as error:
        assert str(error)=='injected_failure_before_commit'
    else:
        raise AssertionError('injected failure did not occur')
    assert list(output.iterdir())==[]
    recovered,created = run('events.csv','source-manifest.json',config,output)
    assert created is True
    assert verify_bundle(recovered)['distribution']=='not_sent'
    metrics = json.loads((recovered/'metrics.json').read_text())
    assert metrics['amount_paise']==3500 and metrics['selected_events']==3
    repeated,created_again = run('events.csv','source-manifest.json',config,output)
    assert repeated==recovered and created_again is False
    print({'partial_bundle_exposed':False,'recovery_verified':True,'rerun_reused':True})

The temporary output directory is empty after the injected exception. Recovery creates one verified bundle, and the next identical rerun reuses it. The example tests local exception cleanup, not an abrupt operating-system crash or power loss.

Specify the expected state for each failure

For a source hash mismatch, no report should be prepared. For a conflicting ID, the extract should fail rather than choose an arbitrary value. For a rate-limited page, the retry should preserve the cursor and eventually complete or explicitly fail/defer.

For a corrupted existing bundle, a rerun should not silently overwrite the evidence. The lab's verification suite checks that the changed file remains visible as a discrepancy while validation fails.

Write these expected states before implementing the test. Otherwise it is easy to assert whatever the current code happens to do and call that correctness.

Test recovery as well as rejection

A pipeline that rejects bad input but cannot recover from a corrected source is incomplete. After a controlled failure, rerun with valid evidence and verify the final metrics, artifact identity and absence of duplicate effects.

Keep external side effects out of local failure tests unless the environment is explicitly designed for them. The lab uses temporary directories and simulated API functions; it sends no messages and modifies no production systems.

Injected clocks and wait functions make retry tests fast and deterministic. They establish the requested retry sequence, not the actual behavior of a remote server or network.

State the coverage boundary honestly

The supplied checks exercise local validation, reruns, two staging failures, CLI diagnostics, wrapper logging and a fictional API contract. They do not establish distributed exactly-once processing, filesystem crash durability or production concurrency behavior.

Those claims require tests and design appropriate to the real storage and execution environment. Passing a unit test should not be described as proof that every operational failure is handled.

Exercise: inject a failure after metrics are written but before selected events are written. Verify the same no-completed-bundle guarantee. Then describe a separate test needed for a process killed without normal exception cleanup.

NeuraPath's Data Analytics with Generative AI course connects Python testing with operational guarantees. A useful failure test makes the expected state and recovery path explicit.

Continue learning

This article is part of the Reliable reporting automation sequence. Use the neighbouring tasks when you need the prerequisite or the next application.

• Review the prerequisite or neighbouring task in Archive report versions without overwriting the audit trail.
• Continue with Build a runbook for a report that fails overnight.