62 / 67 · 06 Observability-Driven Testing · Correlating Test Results with Production Metrics← prev⊞ allnext →☰ Read as one page
10.2The Correlation Framework
# test_production_correlation.py
"""
Correlate test suite results with production incident data to answer:
- Which tests, if they had existed, would have caught recent incidents?
- Which tests are "load-bearing" (their failure predicts production issues)?
- Which tests are "dead weight" (always pass, never catch real bugs)?
"""
def analyze_test_production_correlation(test_results: list, incidents: list) -> dict:
"""
For each production incident, determine if any test in the suite
covers the affected component and failure mode.
"""
coverage_gaps = []
load_bearing_tests = set()
for incident in incidents:
affected_component = incident["component"]
failure_mode = incident["failure_mode"]
# Find tests that cover this component
covering_tests = [
t for t in test_results
if t["component"] == affected_component
]
if not covering_tests:
coverage_gaps.append({
"incident": incident["id"],
"component": affected_component,
"failure_mode": failure_mode,
"gap_type": "no_test_coverage",
"recommendation": f"Add tests for {affected_component} {failure_mode}",
})
continue
# Check if any covering test was failing before the incident
pre_incident_failures = [
t for t in covering_tests
if t["last_failure"] and t["last_failure"] < incident["detected_at"]
and (incident["detected_at"] - t["last_failure"]).days < 7
]
if pre_incident_failures:
for test in pre_incident_failures:
load_bearing_tests.add(test["name"])
else:
coverage_gaps.append({
"incident": incident["id"],
"component": affected_component,
"failure_mode": failure_mode,
"gap_type": "tests_pass_but_incident_occurred",
"recommendation": (
f"Tests for {affected_component} may not cover "
f"the {failure_mode} failure mode"
),
})
return {
"coverage_gaps": coverage_gaps,
"load_bearing_tests": list(load_bearing_tests),
"gap_count": len(coverage_gaps),
"incident_count": len(incidents),
"coverage_percentage": (
(len(incidents) - len(coverage_gaps)) / len(incidents) * 100
if incidents else 100
),
}