Merge remote-tracking branch 'origin/burn/171-1776263896'
This commit is contained in:
Alexander Whitestone
551
scripts/perf_bottleneck_finder.py
Normal file
551
scripts/perf_bottleneck_finder.py
Normal file
@@ -0,0 +1,551 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Performance Bottleneck Finder — Identify slow tests, builds, and CI steps.
|
||||
|
||||
Analyzes:
|
||||
1. Pytest output for slow tests
|
||||
2. Build logs for slow steps
|
||||
3. CI workflow durations
|
||||
4. File system for large/slow artifacts
|
||||
|
||||
Usage:
|
||||
python3 scripts/perf_bottleneck_finder.py --repo /path/to/repo
|
||||
python3 scripts/perf_bottleneck_finder.py --repo /path/to/repo --json
|
||||
python3 scripts/perf_bottleneck_finder.py --repo /path/to/repo --report metrics/perf_report.md
|
||||
|
||||
Weekly cron:
|
||||
0 9 * * 1 cd /path/to/compounding-intelligence && python3 scripts/perf_bottleneck_finder.py --repo /path/to/target --report metrics/perf_report.md
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
# ── Configuration ──────────────────────────────────────────────────
|
||||
|
||||
SLOW_TEST_THRESHOLD_S = 2.0 # Tests slower than this are flagged
|
||||
SLOW_BUILD_STEP_THRESHOLD_S = 10.0
|
||||
TOP_N_BOTTLENECKS = 10 # Report top N bottlenecks
|
||||
PYTEST_DURATIONS_COUNT = 20 # Number of slow tests to collect
|
||||
LOG_EXTENSIONS = {".log", ".txt"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Bottleneck:
|
||||
"""A single performance bottleneck."""
|
||||
category: str # "test", "build", "ci", "artifact", "import"
|
||||
name: str # What's slow
|
||||
duration_s: float # How long it takes
|
||||
severity: str # "critical", "warning", "info"
|
||||
recommendation: str # How to fix
|
||||
file_path: Optional[str] = None
|
||||
line_number: Optional[int] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class PerfReport:
|
||||
"""Full performance report."""
|
||||
timestamp: str
|
||||
repo_path: str
|
||||
bottlenecks: List[Bottleneck] = field(default_factory=list)
|
||||
summary: Dict[str, Any] = field(default_factory=dict)
|
||||
test_stats: Dict[str, Any] = field(default_factory=dict)
|
||||
build_stats: Dict[str, Any] = field(default_factory=dict)
|
||||
ci_stats: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = asdict(self)
|
||||
return d
|
||||
|
||||
|
||||
# ── Test Analysis ──────────────────────────────────────────────────
|
||||
|
||||
def find_slow_tests_pytest(repo_path: str) -> List[Bottleneck]:
|
||||
"""Run pytest --durations and parse slow tests."""
|
||||
bottlenecks = []
|
||||
|
||||
# Try to run pytest with durations
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["python3", "-m", "pytest", "--co", "-q", "--durations=0"],
|
||||
cwd=repo_path, capture_output=True, text=True, timeout=30
|
||||
)
|
||||
# If tests exist, try to get durations from last run
|
||||
durations_file = os.path.join(repo_path, ".pytest_cache", "v", "cache", "durations")
|
||||
if os.path.exists(durations_file):
|
||||
with open(durations_file) as f:
|
||||
for line in f:
|
||||
parts = line.strip().split()
|
||||
if len(parts) >= 2:
|
||||
try:
|
||||
duration = float(parts[0])
|
||||
test_name = " ".join(parts[1:])
|
||||
if duration > SLOW_TEST_THRESHOLD_S:
|
||||
severity = "critical" if duration > 10 else "warning"
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="test",
|
||||
name=test_name,
|
||||
duration_s=duration,
|
||||
severity=severity,
|
||||
recommendation=f"Test takes {duration:.1f}s. Consider mocking slow I/O, using fixtures, or marking with @pytest.mark.slow."
|
||||
))
|
||||
except ValueError:
|
||||
continue
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError):
|
||||
pass
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
def find_slow_tests_by_scan(repo_path: str) -> List[Bottleneck]:
|
||||
"""Scan test files for patterns that indicate slow tests."""
|
||||
bottlenecks = []
|
||||
test_patterns = [
|
||||
(r"time\.sleep\((\d+(?:\.\d+)?)\)", "Contains time.sleep() — consider using mock or async wait"),
|
||||
(r"subprocess\.run\(.*timeout=(\d+)", "Subprocess with timeout — may block test"),
|
||||
(r"requests\.(get|post|put|delete)\(", "Real HTTP call — mock with responses or httpretty"),
|
||||
(r"open\([^)]*['"]w['"]", "File I/O in test — use tmp_path fixture"),
|
||||
]
|
||||
|
||||
for root, dirs, files in os.walk(repo_path):
|
||||
# Skip hidden and cache dirs
|
||||
dirs[:] = [d for d in dirs if not d.startswith(('.', '__pycache__', 'node_modules', '.git'))]
|
||||
|
||||
for fname in files:
|
||||
if not (fname.startswith("test_") or fname.endswith("_test.py")):
|
||||
continue
|
||||
if not fname.endswith(".py"):
|
||||
continue
|
||||
|
||||
fpath = os.path.join(root, fname)
|
||||
rel_path = os.path.relpath(fpath, repo_path)
|
||||
|
||||
try:
|
||||
with open(fpath) as f:
|
||||
lines = f.readlines()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
for pattern, recommendation in test_patterns:
|
||||
match = re.search(pattern, line)
|
||||
if match:
|
||||
duration = 1.0 # Default estimate
|
||||
if "sleep" in pattern:
|
||||
try:
|
||||
duration = float(match.group(1))
|
||||
except (ValueError, IndexError):
|
||||
duration = 1.0
|
||||
elif "timeout" in pattern:
|
||||
try:
|
||||
duration = float(match.group(1))
|
||||
except (ValueError, IndexError):
|
||||
duration = 10.0
|
||||
else:
|
||||
duration = 2.0 # Estimated
|
||||
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="test",
|
||||
name=f"{rel_path}:{i+1}",
|
||||
duration_s=duration,
|
||||
severity="warning" if duration < 5 else "critical",
|
||||
recommendation=recommendation,
|
||||
file_path=rel_path,
|
||||
line_number=i + 1
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
# ── Build Analysis ─────────────────────────────────────────────────
|
||||
|
||||
def analyze_build_artifacts(repo_path: str) -> List[Bottleneck]:
|
||||
"""Find large build artifacts that slow down builds."""
|
||||
bottlenecks = []
|
||||
large_dirs = {
|
||||
"node_modules": "Consider using npm ci --production or yarn --production",
|
||||
"__pycache__": "Consider .gitignore and cleaning before builds",
|
||||
".tox": "Consider caching tox environments",
|
||||
".pytest_cache": "Consider cleaning between CI runs",
|
||||
"dist": "Check if dist/ artifacts are being rebuilt unnecessarily",
|
||||
"build": "Check if build/ artifacts are being rebuilt unnecessarily",
|
||||
".next": "Next.js cache — consider incremental builds",
|
||||
"venv": "Virtual env in repo — move outside or use Docker",
|
||||
}
|
||||
|
||||
for dirname, recommendation in large_dirs.items():
|
||||
dirpath = os.path.join(repo_path, dirname)
|
||||
if os.path.isdir(dirpath):
|
||||
total_size = 0
|
||||
file_count = 0
|
||||
for root, dirs, files in os.walk(dirpath):
|
||||
for f in files:
|
||||
try:
|
||||
fpath = os.path.join(root, f)
|
||||
total_size += os.path.getsize(fpath)
|
||||
file_count += 1
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
if total_size > 10 * 1024 * 1024: # > 10MB
|
||||
size_mb = total_size / (1024 * 1024)
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="build",
|
||||
name=f"{dirname}/ ({size_mb:.1f}MB, {file_count} files)",
|
||||
duration_s=size_mb * 0.5, # Rough estimate
|
||||
severity="critical" if size_mb > 100 else "warning",
|
||||
recommendation=recommendation
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
def analyze_makefile_targets(repo_path: str) -> List[Bottleneck]:
|
||||
"""Analyze Makefile for potentially slow targets."""
|
||||
bottlenecks = []
|
||||
makefiles = []
|
||||
|
||||
for root, dirs, files in os.walk(repo_path):
|
||||
dirs[:] = [d for d in dirs if not d.startswith(('.', '__pycache__'))]
|
||||
for f in files:
|
||||
if f in ("Makefile", "makefile", "GNUmakefile"):
|
||||
makefiles.append(os.path.join(root, f))
|
||||
|
||||
slow_patterns = [
|
||||
(r"pip install", "pip install without --no-deps or constraints"),
|
||||
(r"npm install(?!.*--production)", "npm install without --production flag"),
|
||||
(r"docker build", "Docker build — consider multi-stage and layer caching"),
|
||||
(r"pytest(?!.*-x|--maxfail)", "pytest without early exit on failure"),
|
||||
(r"mypy|mypy --strict", "Type checking — consider incremental mode"),
|
||||
]
|
||||
|
||||
for mfile in makefiles:
|
||||
rel_path = os.path.relpath(mfile, repo_path)
|
||||
try:
|
||||
with open(mfile) as f:
|
||||
content = f.read()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
for pattern, recommendation in slow_patterns:
|
||||
if re.search(pattern, content):
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="build",
|
||||
name=f"{rel_path}: {pattern}",
|
||||
duration_s=5.0,
|
||||
severity="info",
|
||||
recommendation=recommendation,
|
||||
file_path=rel_path
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
# ── CI Analysis ────────────────────────────────────────────────────
|
||||
|
||||
def analyze_github_actions(repo_path: str) -> List[Bottleneck]:
|
||||
"""Analyze GitHub Actions workflow files for inefficiencies."""
|
||||
bottlenecks = []
|
||||
workflow_dir = os.path.join(repo_path, ".github", "workflows")
|
||||
|
||||
if not os.path.isdir(workflow_dir):
|
||||
return bottlenecks
|
||||
|
||||
slow_patterns = [
|
||||
(r"runs-on:\s*ubuntu-latest", 0, "Consider caching dependencies between runs"),
|
||||
(r"npm install", 2, "Use npm ci instead of npm install for reproducible builds"),
|
||||
(r"pip install(?!.*--cache-dir)", 2, "Add --cache-dir or use pip cache action"),
|
||||
(r"docker build(?!.*--cache-from)", 5, "Use Docker layer caching"),
|
||||
(r"python -m pytest(?!.*-n|--numprocesses)", 3, "Consider pytest-xdist for parallel test execution"),
|
||||
]
|
||||
|
||||
for fname in os.listdir(workflow_dir):
|
||||
if not fname.endswith(('.yml', '.yaml')):
|
||||
continue
|
||||
|
||||
fpath = os.path.join(workflow_dir, fname)
|
||||
try:
|
||||
with open(fpath) as f:
|
||||
content = f.read()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
for pattern, est_savings, recommendation in slow_patterns:
|
||||
if re.search(pattern, content):
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="ci",
|
||||
name=f"{fname}: {pattern}",
|
||||
duration_s=est_savings,
|
||||
severity="info",
|
||||
recommendation=recommendation,
|
||||
file_path=f".github/workflows/{fname}"
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
def analyze_gitea_ci(repo_path: str) -> List[Bottleneck]:
|
||||
"""Analyze Gitea/Drone CI config files."""
|
||||
bottlenecks = []
|
||||
ci_files = [".gitea/workflows", ".drone.yml", ".woodpecker.yml"]
|
||||
|
||||
for ci_path in ci_files:
|
||||
full_path = os.path.join(repo_path, ci_path)
|
||||
if os.path.isfile(full_path):
|
||||
try:
|
||||
with open(full_path) as f:
|
||||
content = f.read()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
if "pip install" in content and "--cache-dir" not in content:
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="ci",
|
||||
name=f"{ci_path}: pip without cache",
|
||||
duration_s=5.0,
|
||||
severity="warning",
|
||||
recommendation="Add --cache-dir or mount pip cache volume",
|
||||
file_path=ci_path
|
||||
))
|
||||
|
||||
elif os.path.isdir(full_path):
|
||||
for fname in os.listdir(full_path):
|
||||
if not fname.endswith(('.yml', '.yaml')):
|
||||
continue
|
||||
fpath = os.path.join(full_path, fname)
|
||||
try:
|
||||
with open(fpath) as f:
|
||||
content = f.read()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
if "pip install" in content and "--cache-dir" not in content:
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="ci",
|
||||
name=f"{ci_path}/{fname}: pip without cache",
|
||||
duration_s=5.0,
|
||||
severity="warning",
|
||||
recommendation="Add --cache-dir or mount pip cache volume",
|
||||
file_path=f"{ci_path}/{fname}"
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
# ── Import Analysis ────────────────────────────────────────────────
|
||||
|
||||
def find_slow_imports(repo_path: str) -> List[Bottleneck]:
|
||||
"""Find Python files with heavy import chains."""
|
||||
bottlenecks = []
|
||||
heavy_imports = {
|
||||
"pandas": 0.5,
|
||||
"numpy": 0.3,
|
||||
"torch": 2.0,
|
||||
"tensorflow": 3.0,
|
||||
"scipy": 0.5,
|
||||
"matplotlib": 0.8,
|
||||
"sklearn": 0.5,
|
||||
"transformers": 1.5,
|
||||
}
|
||||
|
||||
for root, dirs, files in os.walk(repo_path):
|
||||
dirs[:] = [d for d in dirs if not d.startswith(('.', '__pycache__', 'node_modules'))]
|
||||
for fname in files:
|
||||
if not fname.endswith(".py"):
|
||||
continue
|
||||
|
||||
fpath = os.path.join(root, fname)
|
||||
rel_path = os.path.relpath(fpath, repo_path)
|
||||
|
||||
try:
|
||||
with open(fpath) as f:
|
||||
lines = f.readlines()
|
||||
except (PermissionError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("import ") or stripped.startswith("from "):
|
||||
for heavy, est_time in heavy_imports.items():
|
||||
if heavy in stripped:
|
||||
bottlenecks.append(Bottleneck(
|
||||
category="import",
|
||||
name=f"{rel_path}:{i+1}: import {heavy}",
|
||||
duration_s=est_time,
|
||||
severity="info" if est_time < 1.0 else "warning",
|
||||
recommendation=f"Heavy import ({heavy} ~{est_time}s). Consider lazy import or conditional import.",
|
||||
file_path=rel_path,
|
||||
line_number=i + 1
|
||||
))
|
||||
|
||||
return bottlenecks
|
||||
|
||||
|
||||
# ── Report Generation ──────────────────────────────────────────────
|
||||
|
||||
def severity_sort_key(b: Bottleneck) -> Tuple[int, float]:
|
||||
"""Sort by severity then duration."""
|
||||
sev_order = {"critical": 0, "warning": 1, "info": 2}
|
||||
return (sev_order.get(b.severity, 3), -b.duration_s)
|
||||
|
||||
|
||||
def generate_report(repo_path: str) -> PerfReport:
|
||||
"""Run all analyses and generate a performance report."""
|
||||
report = PerfReport(
|
||||
timestamp=datetime.now(timezone.utc).isoformat(),
|
||||
repo_path=os.path.abspath(repo_path)
|
||||
)
|
||||
|
||||
# Collect all bottlenecks
|
||||
all_bottlenecks = []
|
||||
|
||||
print("Scanning for slow tests (pytest cache)...")
|
||||
all_bottlenecks.extend(find_slow_tests_pytest(repo_path))
|
||||
|
||||
print("Scanning for slow test patterns...")
|
||||
all_bottlenecks.extend(find_slow_tests_by_scan(repo_path))
|
||||
|
||||
print("Analyzing build artifacts...")
|
||||
all_bottlenecks.extend(analyze_build_artifacts(repo_path))
|
||||
|
||||
print("Analyzing Makefiles...")
|
||||
all_bottlenecks.extend(analyze_makefile_targets(repo_path))
|
||||
|
||||
print("Analyzing CI workflows...")
|
||||
all_bottlenecks.extend(analyze_github_actions(repo_path))
|
||||
all_bottlenecks.extend(analyze_gitea_ci(repo_path))
|
||||
|
||||
print("Scanning for heavy imports...")
|
||||
all_bottlenecks.extend(find_slow_imports(repo_path))
|
||||
|
||||
# Sort by severity and duration
|
||||
all_bottlenecks.sort(key=severity_sort_key)
|
||||
report.bottlenecks = all_bottlenecks[:TOP_N_BOTTLENECKS * 2] # Keep more for stats
|
||||
|
||||
# Compute summary
|
||||
by_category = defaultdict(list)
|
||||
for b in all_bottlenecks:
|
||||
by_category[b.category].append(b)
|
||||
|
||||
report.summary = {
|
||||
"total_bottlenecks": len(all_bottlenecks),
|
||||
"critical": sum(1 for b in all_bottlenecks if b.severity == "critical"),
|
||||
"warning": sum(1 for b in all_bottlenecks if b.severity == "warning"),
|
||||
"info": sum(1 for b in all_bottlenecks if b.severity == "info"),
|
||||
"estimated_total_slowdown_s": sum(b.duration_s for b in all_bottlenecks),
|
||||
"by_category": {cat: len(items) for cat, items in by_category.items()},
|
||||
}
|
||||
|
||||
report.test_stats = {
|
||||
"slow_tests": len(by_category.get("test", [])),
|
||||
"total_estimated_s": sum(b.duration_s for b in by_category.get("test", [])),
|
||||
}
|
||||
|
||||
report.build_stats = {
|
||||
"build_issues": len(by_category.get("build", [])),
|
||||
"total_estimated_s": sum(b.duration_s for b in by_category.get("build", [])),
|
||||
}
|
||||
|
||||
report.ci_stats = {
|
||||
"ci_issues": len(by_category.get("ci", [])),
|
||||
"total_estimated_s": sum(b.duration_s for b in by_category.get("ci", [])),
|
||||
}
|
||||
|
||||
return report
|
||||
|
||||
|
||||
def format_markdown(report: PerfReport) -> str:
|
||||
"""Format report as markdown."""
|
||||
lines = []
|
||||
lines.append(f"# Performance Bottleneck Report")
|
||||
lines.append(f"")
|
||||
lines.append(f"Generated: {report.timestamp}")
|
||||
lines.append(f"Repository: {report.repo_path}")
|
||||
lines.append(f"")
|
||||
|
||||
# Summary
|
||||
s = report.summary
|
||||
lines.append(f"## Summary")
|
||||
lines.append(f"")
|
||||
lines.append(f"- **Total bottlenecks:** {s['total_bottlenecks']}")
|
||||
lines.append(f"- **Critical:** {s['critical']} | **Warning:** {s['warning']} | **Info:** {s['info']}")
|
||||
lines.append(f"- **Estimated total slowdown:** {s['estimated_total_slowdown_s']:.1f}s")
|
||||
lines.append(f"- **By category:** {', '.join(f'{k}: {v}' for k, v in s['by_category'].items())}")
|
||||
lines.append(f"")
|
||||
|
||||
# Top bottlenecks
|
||||
lines.append(f"## Top {min(TOP_N_BOTTLENECKS, len(report.bottlenecks))} Bottlenecks")
|
||||
lines.append(f"")
|
||||
|
||||
for i, b in enumerate(report.bottlenecks[:TOP_N_BOTTLENECKS], 1):
|
||||
icon = {"critical": "🔴", "warning": "🟡", "info": "🔵"}.get(b.severity, "⚪")
|
||||
loc = f" ({b.file_path}:{b.line_number})" if b.file_path else ""
|
||||
lines.append(f"{i}. {icon} **{b.category}** — {b.name}{loc}")
|
||||
lines.append(f" - Duration: ~{b.duration_s:.1f}s | Severity: {b.severity}")
|
||||
lines.append(f" - Fix: {b.recommendation}")
|
||||
lines.append(f"")
|
||||
|
||||
# Category breakdowns
|
||||
for cat in ["test", "build", "ci", "import"]:
|
||||
items = [b for b in report.bottlenecks if b.category == cat]
|
||||
if items:
|
||||
lines.append(f"## {cat.title()} Bottlenecks")
|
||||
lines.append(f"")
|
||||
for b in items:
|
||||
icon = {"critical": "🔴", "warning": "🟡", "info": "🔵"}.get(b.severity, "⚪")
|
||||
loc = f" ({b.file_path}:{b.line_number})" if b.file_path else ""
|
||||
lines.append(f"- {icon} {b.name}{loc} — ~{b.duration_s:.1f}s — {b.recommendation}")
|
||||
lines.append(f"")
|
||||
|
||||
return "
|
||||
".join(lines)
|
||||
|
||||
|
||||
# ── Main ───────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Performance Bottleneck Finder")
|
||||
parser.add_argument("--repo", default=".", help="Path to repository to analyze")
|
||||
parser.add_argument("--json", action="store_true", help="Output as JSON")
|
||||
parser.add_argument("--report", help="Write markdown report to file")
|
||||
parser.add_argument("--threshold", type=float, default=SLOW_TEST_THRESHOLD_S,
|
||||
help="Slow test threshold in seconds")
|
||||
args = parser.parse_args()
|
||||
|
||||
global SLOW_TEST_THRESHOLD_S
|
||||
SLOW_TEST_THRESHOLD_S = args.threshold
|
||||
|
||||
if not os.path.isdir(args.repo):
|
||||
print(f"Error: {args.repo} is not a directory", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
report = generate_report(args.repo)
|
||||
|
||||
if args.json:
|
||||
print(json.dumps(report.to_dict(), indent=2))
|
||||
else:
|
||||
md = format_markdown(report)
|
||||
if args.report:
|
||||
os.makedirs(os.path.dirname(args.report) or ".", exist_ok=True)
|
||||
with open(args.report, "w") as f:
|
||||
f.write(md)
|
||||
print(f"Report written to {args.report}")
|
||||
else:
|
||||
print(md)
|
||||
|
||||
# Exit code: 1 if critical bottlenecks found
|
||||
if report.summary.get("critical", 0) > 0:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
319
tests/test_perf_bottleneck_finder.py
Normal file
319
tests/test_perf_bottleneck_finder.py
Normal file
@@ -0,0 +1,319 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tests for Performance Bottleneck Finder.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Add scripts to path
|
||||
import sys
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
|
||||
|
||||
from perf_bottleneck_finder import (
|
||||
Bottleneck,
|
||||
PerfReport,
|
||||
find_slow_tests_by_scan,
|
||||
analyze_build_artifacts,
|
||||
analyze_makefile_targets,
|
||||
find_slow_imports,
|
||||
generate_report,
|
||||
format_markdown,
|
||||
severity_sort_key,
|
||||
)
|
||||
|
||||
|
||||
class TestBottleneck:
|
||||
"""Test Bottleneck dataclass."""
|
||||
|
||||
def test_creation(self):
|
||||
b = Bottleneck(
|
||||
category="test",
|
||||
name="test_foo",
|
||||
duration_s=5.0,
|
||||
severity="warning",
|
||||
recommendation="Mock it"
|
||||
)
|
||||
assert b.category == "test"
|
||||
assert b.name == "test_foo"
|
||||
assert b.duration_s == 5.0
|
||||
assert b.severity == "warning"
|
||||
assert b.recommendation == "Mock it"
|
||||
assert b.file_path is None
|
||||
assert b.line_number is None
|
||||
|
||||
def test_with_location(self):
|
||||
b = Bottleneck(
|
||||
category="test",
|
||||
name="test_bar",
|
||||
duration_s=2.0,
|
||||
severity="info",
|
||||
recommendation="Consider",
|
||||
file_path="tests/test_bar.py",
|
||||
line_number=42
|
||||
)
|
||||
assert b.file_path == "tests/test_bar.py"
|
||||
assert b.line_number == 42
|
||||
|
||||
def test_to_dict(self):
|
||||
b = Bottleneck("test", "x", 1.0, "info", "y")
|
||||
d = b.__dict__
|
||||
assert "category" in d
|
||||
assert "duration_s" in d
|
||||
|
||||
|
||||
class TestPerfReport:
|
||||
"""Test PerfReport dataclass."""
|
||||
|
||||
def test_creation(self):
|
||||
report = PerfReport(
|
||||
timestamp="2026-01-01T00:00:00Z",
|
||||
repo_path="/tmp/repo"
|
||||
)
|
||||
assert report.timestamp == "2026-01-01T00:00:00Z"
|
||||
assert report.bottlenecks == []
|
||||
assert report.summary == {}
|
||||
|
||||
def test_to_dict(self):
|
||||
report = PerfReport(
|
||||
timestamp="2026-01-01T00:00:00Z",
|
||||
repo_path="/tmp/repo",
|
||||
bottlenecks=[Bottleneck("test", "x", 1.0, "info", "y")]
|
||||
)
|
||||
d = report.to_dict()
|
||||
assert "bottlenecks" in d
|
||||
assert len(d["bottlenecks"]) == 1
|
||||
|
||||
|
||||
class TestSeveritySort:
|
||||
"""Test severity sorting."""
|
||||
|
||||
def test_critical_first(self):
|
||||
items = [
|
||||
Bottleneck("test", "a", 1.0, "info", ""),
|
||||
Bottleneck("test", "b", 0.5, "critical", ""),
|
||||
Bottleneck("test", "c", 2.0, "warning", ""),
|
||||
]
|
||||
items.sort(key=severity_sort_key)
|
||||
assert items[0].severity == "critical"
|
||||
assert items[1].severity == "warning"
|
||||
assert items[2].severity == "info"
|
||||
|
||||
def test_duration_within_severity(self):
|
||||
items = [
|
||||
Bottleneck("test", "slow", 10.0, "warning", ""),
|
||||
Bottleneck("test", "fast", 1.0, "warning", ""),
|
||||
]
|
||||
items.sort(key=severity_sort_key)
|
||||
assert items[0].name == "slow" # Higher duration first within same severity
|
||||
|
||||
|
||||
class TestSlowTestScan:
|
||||
"""Test slow test pattern scanning."""
|
||||
|
||||
def test_finds_sleep(self, tmp_path):
|
||||
test_file = tmp_path / "test_sleepy.py"
|
||||
test_file.write_text(textwrap.dedent('''
|
||||
import time
|
||||
|
||||
def test_slow():
|
||||
time.sleep(5)
|
||||
assert True
|
||||
'''))
|
||||
|
||||
bottlenecks = find_slow_tests_by_scan(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
assert any("sleep" in b.recommendation.lower() for b in bottlenecks)
|
||||
|
||||
def test_finds_http_calls(self, tmp_path):
|
||||
test_file = tmp_path / "test_http.py"
|
||||
test_file.write_text(textwrap.dedent('''
|
||||
import requests
|
||||
|
||||
def test_api():
|
||||
resp = requests.get("https://example.com")
|
||||
assert resp.status_code == 200
|
||||
'''))
|
||||
|
||||
bottlenecks = find_slow_tests_by_scan(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
assert any("HTTP" in b.recommendation or "mock" in b.recommendation.lower() for b in bottlenecks)
|
||||
|
||||
def test_skips_non_test_files(self, tmp_path):
|
||||
src_file = tmp_path / "main.py"
|
||||
src_file.write_text("import time\ntime.sleep(10)\n")
|
||||
|
||||
bottlenecks = find_slow_tests_by_scan(str(tmp_path))
|
||||
assert len(bottlenecks) == 0
|
||||
|
||||
def test_handles_missing_dir(self):
|
||||
bottlenecks = find_slow_tests_by_scan("/nonexistent/path")
|
||||
assert bottlenecks == []
|
||||
|
||||
def test_file_path_populated(self, tmp_path):
|
||||
test_file = tmp_path / "test_example.py"
|
||||
test_file.write_text("import time\n\ndef test_it():\n time.sleep(2)\n")
|
||||
|
||||
bottlenecks = find_slow_tests_by_scan(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
assert bottlenecks[0].file_path is not None
|
||||
assert bottlenecks[0].line_number is not None
|
||||
|
||||
|
||||
class TestBuildArtifacts:
|
||||
"""Test build artifact analysis."""
|
||||
|
||||
def test_finds_large_node_modules(self, tmp_path):
|
||||
nm = tmp_path / "node_modules"
|
||||
nm.mkdir()
|
||||
# Create a file > 10MB
|
||||
big_file = nm / "big.txt"
|
||||
big_file.write_bytes(b"x" * (11 * 1024 * 1024))
|
||||
|
||||
bottlenecks = analyze_build_artifacts(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
assert any("node_modules" in b.name for b in bottlenecks)
|
||||
|
||||
def test_ignores_small_dirs(self, tmp_path):
|
||||
nm = tmp_path / "node_modules"
|
||||
nm.mkdir()
|
||||
small_file = nm / "small.txt"
|
||||
small_file.write_bytes(b"x" * 100)
|
||||
|
||||
bottlenecks = analyze_build_artifacts(str(tmp_path))
|
||||
assert not any("node_modules" in b.name for b in bottlenecks)
|
||||
|
||||
def test_finds_pycache(self, tmp_path):
|
||||
cache = tmp_path / "__pycache__"
|
||||
cache.mkdir()
|
||||
big_file = cache / "big.pyc"
|
||||
big_file.write_bytes(b"x" * (11 * 1024 * 1024))
|
||||
|
||||
bottlenecks = analyze_build_artifacts(str(tmp_path))
|
||||
assert any("__pycache__" in b.name for b in bottlenecks)
|
||||
|
||||
|
||||
class TestMakefileAnalysis:
|
||||
"""Test Makefile analysis."""
|
||||
|
||||
def test_finds_pip_install(self, tmp_path):
|
||||
makefile = tmp_path / "Makefile"
|
||||
makefile.write_text(textwrap.dedent('''
|
||||
install:
|
||||
pip install -r requirements.txt
|
||||
|
||||
test:
|
||||
pytest
|
||||
'''))
|
||||
|
||||
bottlenecks = analyze_makefile_targets(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
|
||||
def test_no_makefile(self, tmp_path):
|
||||
bottlenecks = analyze_makefile_targets(str(tmp_path))
|
||||
assert bottlenecks == []
|
||||
|
||||
|
||||
class TestImportAnalysis:
|
||||
"""Test heavy import detection."""
|
||||
|
||||
def test_finds_pandas(self, tmp_path):
|
||||
src = tmp_path / "analysis.py"
|
||||
src.write_text("import pandas as pd\n")
|
||||
|
||||
bottlenecks = find_slow_imports(str(tmp_path))
|
||||
assert len(bottlenecks) >= 1
|
||||
assert any("pandas" in b.name for b in bottlenecks)
|
||||
|
||||
def test_finds_torch(self, tmp_path):
|
||||
src = tmp_path / "model.py"
|
||||
src.write_text("import torch\n")
|
||||
|
||||
bottlenecks = find_slow_imports(str(tmp_path))
|
||||
assert any("torch" in b.name for b in bottlenecks)
|
||||
|
||||
def test_skips_light_imports(self, tmp_path):
|
||||
src = tmp_path / "utils.py"
|
||||
src.write_text("import json\nimport os\nimport sys\n")
|
||||
|
||||
bottlenecks = find_slow_imports(str(tmp_path))
|
||||
assert len(bottlenecks) == 0
|
||||
|
||||
|
||||
class TestGenerateReport:
|
||||
"""Test full report generation."""
|
||||
|
||||
def test_empty_repo(self, tmp_path):
|
||||
report = generate_report(str(tmp_path))
|
||||
assert report.summary["total_bottlenecks"] >= 0
|
||||
assert "critical" in report.summary
|
||||
assert "warning" in report.summary
|
||||
|
||||
def test_with_findings(self, tmp_path):
|
||||
# Create a test file with issues
|
||||
test_file = tmp_path / "test_slow.py"
|
||||
test_file.write_text(textwrap.dedent('''
|
||||
import time
|
||||
import requests
|
||||
|
||||
def test_sleepy():
|
||||
time.sleep(3)
|
||||
|
||||
def test_http():
|
||||
requests.get("https://example.com")
|
||||
'''))
|
||||
|
||||
report = generate_report(str(tmp_path))
|
||||
assert report.summary["total_bottlenecks"] >= 2
|
||||
assert len(report.bottlenecks) > 0
|
||||
|
||||
def test_summary_categories(self, tmp_path):
|
||||
report = generate_report(str(tmp_path))
|
||||
assert "by_category" in report.summary
|
||||
|
||||
|
||||
class TestMarkdownReport:
|
||||
"""Test markdown output."""
|
||||
|
||||
def test_format(self):
|
||||
report = PerfReport(
|
||||
timestamp="2026-01-01T00:00:00Z",
|
||||
repo_path="/tmp/repo",
|
||||
bottlenecks=[
|
||||
Bottleneck("test", "slow_test", 5.0, "critical", "Fix it")
|
||||
],
|
||||
summary={
|
||||
"total_bottlenecks": 1,
|
||||
"critical": 1,
|
||||
"warning": 0,
|
||||
"info": 0,
|
||||
"estimated_total_slowdown_s": 5.0,
|
||||
"by_category": {"test": 1},
|
||||
}
|
||||
)
|
||||
md = format_markdown(report)
|
||||
assert "# Performance Bottleneck Report" in md
|
||||
assert "slow_test" in md
|
||||
assert "🔴" in md
|
||||
assert "Fix it" in md
|
||||
|
||||
def test_empty_report(self):
|
||||
report = PerfReport(
|
||||
timestamp="2026-01-01T00:00:00Z",
|
||||
repo_path="/tmp/repo",
|
||||
summary={
|
||||
"total_bottlenecks": 0,
|
||||
"critical": 0,
|
||||
"warning": 0,
|
||||
"info": 0,
|
||||
"estimated_total_slowdown_s": 0,
|
||||
"by_category": {},
|
||||
}
|
||||
)
|
||||
md = format_markdown(report)
|
||||
assert "Total bottlenecks:** 0" in md
|
||||
Reference in New Issue
Block a user