Generating IQ/OQ/PQ Protocols from Python Tests

The gap shows up the first time a validation lead asks for the OQ protocol behind a pipeline release and the answer is a Word document that nobody has re-executed since it was written: the tests that actually run in CI and the protocol that claims to qualify the system have drifted apart, and under inspection that gap is a finding. Python ETL engineers and quality teams hit this the moment they try to make a pytest suite serve double duty as GxP evidence — the suite runs green, but it emits nothing an auditor can sign. This deep-dive, part of Validation and CSV Frameworks for Clinical Data Pipelines within the broader Clinical Data Architecture & EDC Standards program, shows how to instrument the suite so a single run produces Installation, Operational, and Performance Qualification evidence — Computerized System Validation (CSV) evidence, not comma-separated data — bound to requirements, rendered into a human-readable protocol, and sealed into a tamper-evident bundle. Here CSV means Computerized System Validation.

Evidence Generation Flow #

A qualified run captures three evidence classes from one suite: the environment snapshot (IQ), the requirement-tagged test outcomes (OQ), and the representative-load timings (PQ), then bundles and seals them.

Root Cause: Why Green Tests Are Not Yet Evidence #

A passing pytest run and a qualification protocol answer different questions. The suite answers “does the code behave correctly right now, on this machine?” A protocol answers “was this specific, specified system proven to work, by whom, on what date, against which requirement, and can that be reproduced?” The default test run discards almost everything the second question needs: it does not record which requirement each test verifies, it does not capture the exact installed environment, it does not run against production-scale data, and its console output is ephemeral. Worse, an ordinary suite is often quietly non-deterministic — it reaches for datetime.now(), an unseeded shuffle, or a live endpoint — so even if you saved the output, re-running it would produce a different result, which is the opposite of reproducible evidence.

The fix is to treat the run as an evidence-generating event. Tag every test with the requirement it verifies, capture the environment as IQ evidence, exercise representative load as PQ evidence, emit machine-readable OQ records, and seal the whole set so it cannot be edited after the fact. Because the same build produces the evidence and ships to production, the qualified system and the deployed system are provably identical — the gap manual protocols leave open simply does not exist. The requirement tagging also feeds the traceability matrix described in the parent Validation and CSV Frameworks guide.

Step-by-Step Implementation #

Step 1 — Tag every test with the requirement it verifies #

A test that does not cite a requirement cannot serve as OQ evidence, because OQ is a per-requirement claim. Register a req marker and attach it to each test; a pytest hook then records, at execution time, which requirement each test verified and its outcome. Recording at execution time (not from a static list) keeps the evidence contemporaneous — it reflects what actually ran.

# 21 CFR Part 11 relevance: 11.10(a) operational verification is per-requirement.
# The marker binds each test to a URS id; the hook records the outcome as it runs,
# so the evidence is contemporaneous rather than transcribed after the fact.
import json

def pytest_configure(config):
    config.addinivalue_line("markers", "req(id): the URS requirement this test verifies")

_OQ_RECORDS = []

def pytest_runtest_logreport(report):
    if report.when != "call":
        return
    marker = report.keywords.get("req")
    req_id = getattr(marker, "args", ["UNTRACED"])[0] if marker else "UNTRACED"
    _OQ_RECORDS.append({
        "requirement": req_id,
        "test": report.nodeid,
        "outcome": report.outcome,          # passed | failed | skipped
        "duration_s": round(report.duration, 4),
    })

def pytest_sessionfinish(session):
    with open("oq_records.json", "w") as fh:
        json.dump(_OQ_RECORDS, fh, indent=2, sort_keys=True)

Step 2 — Capture IQ evidence: the exact installed environment #

IQ proves the system is installed as specified. The evidence is a snapshot of the resolved dependency set, the interpreter build, and — where the pipeline ships as a container — the image digest, which pins the OS layer and system libraries an ordinary pip freeze misses. Hash the manifest so any later drift is detectable by comparing a single value.

# 21 CFR Part 11 relevance: 11.10(a) installation verification. The pip-freeze hash
# plus container digest fix the qualified installed state; a mismatch later proves
# the running system diverged from what was validated.
import hashlib, os, platform, subprocess

def capture_iq_evidence() -> dict:
    frozen = subprocess.run(["pip", "freeze"], capture_output=True, text=True).stdout
    return {
        "python_build": platform.python_version(),
        "platform": platform.platform(),
        "container_image_digest": os.environ.get("IMAGE_DIGEST", "not-containerized"),
        "dependency_manifest_sha256": hashlib.sha256(frozen.encode()).hexdigest(),
        "frozen_dependencies": frozen.splitlines(),
    }

Step 3 — Exercise representative load for PQ evidence #

PQ proves the system performs under production-like conditions, so a toy fixture will not do — use a representative dataset at production scale (realistic subject counts, SDTM domain volumes, and audit-record cardinality). Capture throughput and latency, and assert against the specified performance requirement so the PQ record carries a pass/fail, not just a number.

# 21 CFR Part 11 relevance: PQ demonstrates fitness for intended use at scale.
# Timing is captured against a representative ODM export so the evidence reflects
# production conditions, and the assertion turns the measurement into a verdict.
import time, pytest

@pytest.mark.req("URS-030")   # "The pipeline SHALL ingest 50k audit records within 90s"
def test_pq_representative_load(pipeline, representative_odm_50k):
    start = time.perf_counter()
    result = pipeline.ingest(representative_odm_50k)   # ~50,000 audit records
    elapsed = time.perf_counter() - start
    with open("pq_timings.json", "w") as fh:
        import json; json.dump({"records": result.count, "elapsed_s": elapsed}, fh)
    assert result.count == 50_000
    assert elapsed < 90, f"PQ throughput requirement missed: {elapsed:.1f}s"

Step 4 — Render JUnit XML into a human-readable protocol #

Run the suite with pytest --junitxml=oq_evidence.xml. The JUnit XML is the raw, machine-readable OQ record; a reviewer, however, signs a legible protocol. Render the XML (joined with the requirement records from Step 1) into a page-numbered protocol table showing each requirement, its test, the outcome, and the run timestamp, so a validation lead can review and approve it.

# 21 CFR Part 11 relevance: 11.10(a) reviewed evidence. The renderer joins raw
# JUnit results with requirement ids into a legible, reviewable protocol without
# mutating the underlying XML, which remains the tamper-evidence source.
import xml.etree.ElementTree as ET
from datetime import datetime, timezone

def render_oq_protocol(junit_path: str, out_path: str) -> None:
    root = ET.parse(junit_path).getroot()
    lines = [f"OQ PROTOCOL — generated {datetime.now(timezone.utc).isoformat()}", "=" * 60]
    for case in root.iter("testcase"):
        failed = case.find("failure") is not None
        status = "FAIL" if failed else "PASS"
        lines.append(f"{case.get('classname')}::{case.get('name')}  [{status}]  "
                     f"{float(case.get('time', 0)):.3f}s")
    with open(out_path, "w") as fh:
        fh.write("\n".join(lines))

Step 5 — Bundle and seal the evidence tamper-evidently #

Collect the IQ snapshot, OQ records, JUnit XML, PQ timings, and rendered protocol into one bundle, then hash the bundle so any post-hoc edit is detectable. Sealing (a signature or a write to WORM storage) makes the evidence enduring and attributable to the run that produced it.

# 21 CFR Part 11 relevance: 11.10(c)/(e) protect records and make them tamper-evident.
# A manifest hash over every artifact means altering any file after sealing changes
# the manifest digest, which verification will surface immediately.
import hashlib, json, pathlib

def seal_bundle(artifact_paths: list[str], manifest_out: str) -> str:
    manifest = {}
    for p in sorted(artifact_paths):
        data = pathlib.Path(p).read_bytes()
        manifest[p] = hashlib.sha256(data).hexdigest()
    body = json.dumps(manifest, sort_keys=True).encode()
    bundle_hash = hashlib.sha256(body).hexdigest()
    pathlib.Path(manifest_out).write_text(
        json.dumps({"artifacts": manifest, "bundle_sha256": bundle_hash}, indent=2))
    return bundle_hash   # record this in the release log / sign it

Verification and Audit Trail #

Verification confirms the bundle is complete, reproducible, and unaltered. Re-run the suite on a clean checkout of the same commit and confirm the OQ outcomes match; recompute each artifact hash and confirm it equals the sealed manifest; and confirm the IQ manifest hash matches the qualified lockfile. Capture these fields per qualification run as the inspection evidence.

Edge Cases and Vendor-Specific Gotchas #

Floating dependencies void the IQ. If the lockfile is not hash-locked, pip install can resolve a different transitive version tomorrow, so the environment you qualified is not the one that runs. Compile the lockfile with pip-compile --generate-hashes, assert its hash in an IQ test, and treat any bump as a change-control trigger that re-runs qualification.

Flaky, non-deterministic tests cannot be evidence. A test that reads the wall clock, shuffles without a seed, or depends on dict ordering yields a result that is not reproducible, so its pass proves nothing under inspection. Inject a frozen clock, seed all randomness, canonicalize serialization, and mock every external system before the suite is allowed to generate OQ evidence.

The evidence bundle itself must be tamper-evident. A protocol saved to an editable share can be altered after signing, which defeats the purpose. Hash the bundle, record bundle_sha256 in the release log, and write the sealed set to WORM storage or attach a cryptographic signature so any later change is detectable — the same discipline used for the pipeline’s own 21 CFR Part 11 audit log.

Frequently Asked Questions #

Related #

• Validation and CSV Frameworks for Clinical Data Pipelines — the parent guide defining the V-model, GAMP 5 categorization, and traceability this evidence feeds.
• Designing a 21 CFR Part 11 Audit Log Schema — the tamper-evident logging pattern the evidence bundle reuses.
• Audit Trail Boundaries in EDC Systems — the provenance discipline the qualified pipeline must preserve.
• Clinical Data Architecture & EDC Standards — the architecture program this qualification work sits within.