File size: 6,030 Bytes
cc036ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """
Flaky Test Tracker - Tracks test results across multiple runs to identify unstable tests.
A test is considered flaky if:
- It fails in at least 1 of the last N runs (default: 5)
- Its pass rate is below threshold (default: 80%)
- It has been run at least minimum times (default: 3)
Usage:
from scripts.flaky_test_tracker import FlakyTestTracker
tracker = FlakyTestTracker()
tracker.update_from_pytest_report("pytest_report.json")
flaky_tests = tracker.get_flaky_tests(threshold=0.8)
"""
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
class FlakyTestTracker:
"""Tracks test results across runs to identify flaky tests."""
def __init__(self, data_file: str = "backend/tests/e2e_ui/data/flaky_tests.json"):
"""Initialize tracker with data file path."""
self.data_file = Path(data_file)
self.data: Dict = self._load_data()
def _load_data(self) -> Dict:
"""Load historical test data from JSON file."""
if self.data_file.exists():
with open(self.data_file, 'r') as f:
return json.load(f)
return {
"tests": {},
"last_updated": None,
"total_runs": 0
}
def _save_data(self) -> None:
"""Save test data to JSON file."""
self.data["last_updated"] = datetime.now().isoformat()
self.data_file.parent.mkdir(parents=True, exist_ok=True)
with open(self.data_file, 'w') as f:
json.dump(self.data, f, indent=2)
def update_from_pytest_report(self, report_file: str) -> None:
"""Update tracker data from pytest JSON report."""
with open(report_file, 'r') as f:
report = json.load(f)
summary = report.get("summary", {})
self.data["total_runs"] += 1
# Process each test result
for test_name, test_result in report.get("tests", {}).items():
if test_name not in self.data["tests"]:
self.data["tests"][test_name] = {
"runs": [],
"total_runs": 0,
"passed": 0,
"failed": 0,
"skipped": 0,
"errors": 0
}
test_data = self.data["tests"][test_name]
test_data["total_runs"] += 1
# Record outcome
outcome = test_result.get("outcome", "unknown")
test_data["runs"].append({
"run": self.data["total_runs"],
"outcome": outcome,
"timestamp": datetime.now().isoformat()
})
if outcome == "passed":
test_data["passed"] += 1
elif outcome == "failed":
test_data["failed"] += 1
elif outcome == "skipped":
test_data["skipped"] += 1
else:
test_data["errors"] += 1
# Keep only last N runs (default: 10)
test_data["runs"] = test_data["runs"][-10:]
self._save_data()
def get_flaky_tests(
self,
pass_threshold: float = 0.8,
min_runs: int = 3,
last_n_runs: Optional[int] = None
) -> List[Dict]:
"""
Get list of flaky tests.
Args:
pass_threshold: Minimum pass rate (0.0-1.0)
min_runs: Minimum runs before considering test
last_n_runs: Only consider last N runs (None = all runs)
Returns:
List of flaky test info dicts
"""
flaky = []
for test_name, test_data in self.data["tests"].items():
# Skip tests with insufficient runs
if test_data["total_runs"] < min_runs:
continue
# Calculate pass rate
runs_to_consider = test_data["runs"][-last_n_runs:] if last_n_runs else test_data["runs"]
if not runs_to_consider:
continue
passed = sum(1 for r in runs_to_consider if r["outcome"] == "passed")
total = len(runs_to_consider)
pass_rate = passed / total if total > 0 else 0
# Check if flaky (below threshold AND has at least one failure)
has_failures = any(r["outcome"] in ["failed", "error"] for r in runs_to_consider)
if pass_rate < pass_threshold and has_failures:
flaky.append({
"test": test_name,
"pass_rate": pass_rate,
"total_runs": total,
"passed": passed,
"failed": total - passed,
"last_outcome": runs_to_consider[-1]["outcome"]
})
# Sort by pass rate (most flaky first)
flaky.sort(key=lambda x: x["pass_rate"])
return flaky
def get_test_history(self, test_name: str) -> Dict:
"""Get historical data for a specific test."""
return self.data["tests"].get(test_name, {})
def get_summary(self) -> Dict:
"""Get summary of tracked test data."""
total_tests = len(self.data["tests"])
flaky = self.get_flaky_tests()
return {
"total_tests": total_tests,
"total_runs": self.data["total_runs"],
"flaky_count": len(flaky),
"flaky_tests": [t["test"] for t in flaky[:10]], # Top 10
"last_updated": self.data["last_updated"]
}
if __name__ == "__main__":
# CLI interface for manual testing
import sys
tracker = FlakyTestTracker()
if len(sys.argv) > 1:
# Update from report file
report_file = sys.argv[1]
tracker.update_from_pytest_report(report_file)
print(f"Updated tracker from {report_file}")
# Print flaky tests
flaky = tracker.get_flaky_tests()
print(f"\nFound {len(flaky)} flaky tests:")
for test in flaky:
print(f" - {test['test']}: {test['pass_rate']:.1%} pass rate ({test['passed']}/{test['total_runs']} runs)")
|