10.5Implementing the Correlation Pipeline

Step 1: Tag Tests with Component Metadata

# conftest.py -- Pytest markers for component tagging
import pytest

def pytest_configure(config):
    config.addinivalue_line("markers", "component(name): tag test with component")
    config.addinivalue_line("markers", "failure_mode(mode): tag with failure mode")

# test_checkout.py
@pytest.mark.component("checkout-service")
@pytest.mark.failure_mode("payment_timeout")
def test_checkout_handles_payment_timeout():
    """Verify checkout gracefully handles payment service timeout."""
    pass

Step 2: Export Test Results to a Database

# test_result_exporter.py
import json
from datetime import datetime

def export_test_results(pytest_json_report: dict, build_id: str) -> list:
    """Convert pytest JSON report to correlation-ready records."""
    records = []
    for test in pytest_json_report["tests"]:
        markers = {m["name"]: m.get("args", []) for m in test.get("markers", [])}

        records.append({
            "test_name": test["nodeid"],
            "component": markers.get("component", ["unknown"])[0],
            "failure_mode": markers.get("failure_mode", ["general"])[0],
            "outcome": test["outcome"],  # passed, failed, skipped
            "duration_ms": test["duration"] * 1000,
            "build_id": build_id,
            "timestamp": datetime.utcnow().isoformat(),
            "last_failure": (
                datetime.utcnow().isoformat() if test["outcome"] == "failed" else None
            ),
        })

    return records

Step 3: Run Correlation Analysis After Every Incident

# post_incident_correlation.py
def post_incident_analysis(incident: dict, test_db, metric_db) -> dict:
    """Run after every production incident to identify test gaps."""

    # Get all tests for the affected component
    component_tests = test_db.query(
        "SELECT * FROM test_results WHERE component = %s ORDER BY timestamp DESC",
        [incident["component"]]
    )

    # Were any tests failing in the 7 days before the incident?
    recent_failures = [
        t for t in component_tests
        if t["outcome"] == "failed"
        and t["timestamp"] > incident["detected_at"] - timedelta(days=7)
        and t["timestamp"] < incident["detected_at"]
    ]

    # Get the production metrics during the incident
    metrics_during = metric_db.query(
        "SELECT * FROM metrics WHERE service = %s AND timestamp BETWEEN %s AND %s",
        [incident["component"], incident["started_at"], incident["resolved_at"]]
    )

    return {
        "incident_id": incident["id"],
        "component": incident["component"],
        "total_component_tests": len(set(t["test_name"] for t in component_tests)),
        "tests_failing_before_incident": len(recent_failures),
        "was_predictable": len(recent_failures) > 0,
        "failing_tests": [t["test_name"] for t in recent_failures],
        "recommendation": (
            "Investigate why failing tests were not acted upon"
            if recent_failures
            else f"Add tests covering {incident['failure_mode']} for {incident['component']}"
        ),
        "metrics_during_incident": {
            "peak_error_rate": max(m["error_rate"] for m in metrics_during),
            "peak_latency_p99": max(m["latency_p99"] for m in metrics_during),
            "duration_minutes": (
                incident["resolved_at"] - incident["started_at"]
            ).total_seconds() / 60,
        },
    }