1318 lines
43 KiB
Python
1318 lines
43 KiB
Python
"""Tests for the evaluation framework (scorer, reporter, runner, benchmarks)."""
|
|
|
|
import csv
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from code_review_graph.eval.reporter import (
|
|
generate_full_report,
|
|
generate_markdown_report,
|
|
generate_readme_tables,
|
|
)
|
|
|
|
try:
|
|
import yaml as _yaml # noqa: F401
|
|
|
|
from code_review_graph.eval.runner import load_all_configs, load_config, write_csv
|
|
_HAS_YAML = True
|
|
except ImportError:
|
|
_HAS_YAML = False
|
|
load_all_configs = None # type: ignore[assignment]
|
|
load_config = None # type: ignore[assignment]
|
|
write_csv = None # type: ignore[assignment]
|
|
from code_review_graph.eval.scorer import (
|
|
compute_mrr,
|
|
compute_precision_recall,
|
|
compute_token_efficiency,
|
|
)
|
|
|
|
# --- Existing scorer tests ---
|
|
|
|
|
|
def test_token_efficiency():
|
|
result = compute_token_efficiency(10000, 3000)
|
|
assert result["raw_tokens"] == 10000
|
|
assert result["graph_tokens"] == 3000
|
|
assert result["ratio"] == 0.3
|
|
assert result["reduction_percent"] == 70.0
|
|
|
|
|
|
def test_token_efficiency_zero_raw():
|
|
result = compute_token_efficiency(0, 100)
|
|
assert result["ratio"] == 0.0
|
|
assert result["reduction_percent"] == 0.0
|
|
|
|
|
|
def test_mrr_found_at_rank_2():
|
|
result = compute_mrr("b", ["a", "b", "c"])
|
|
assert result == 0.5
|
|
|
|
|
|
def test_mrr_found_at_rank_1():
|
|
result = compute_mrr("a", ["a", "b", "c"])
|
|
assert result == 1.0
|
|
|
|
|
|
def test_mrr_not_found():
|
|
result = compute_mrr("z", ["a", "b", "c"])
|
|
assert result == 0.0
|
|
|
|
|
|
def test_precision_recall():
|
|
predicted = {"a", "b", "c", "d"}
|
|
actual = {"b", "c", "e"}
|
|
result = compute_precision_recall(predicted, actual)
|
|
assert result["precision"] == 0.5
|
|
assert result["recall"] == round(2 / 3, 4)
|
|
expected_f1 = round(2 * 0.5 * (2 / 3) / (0.5 + 2 / 3), 4)
|
|
assert result["f1"] == expected_f1
|
|
|
|
|
|
def test_precision_recall_empty_sets():
|
|
result = compute_precision_recall(set(), set())
|
|
assert result["precision"] == 1.0
|
|
assert result["recall"] == 1.0
|
|
assert result["f1"] == 1.0
|
|
|
|
|
|
def test_precision_recall_no_overlap():
|
|
result = compute_precision_recall({"a"}, {"b"})
|
|
assert result["precision"] == 0.0
|
|
assert result["recall"] == 0.0
|
|
assert result["f1"] == 0.0
|
|
|
|
|
|
def test_generate_markdown_report():
|
|
results = [
|
|
{
|
|
"benchmark": "token_efficiency",
|
|
"ratio": 0.3,
|
|
"reduction_percent": 70.0,
|
|
},
|
|
{
|
|
"benchmark": "search_mrr",
|
|
"ratio": "-",
|
|
"reduction_percent": "-",
|
|
},
|
|
]
|
|
report = generate_markdown_report(results)
|
|
assert "# Evaluation Report" in report
|
|
assert "## Summary" in report
|
|
assert "token_efficiency" in report
|
|
assert "search_mrr" in report
|
|
assert "70.0" in report
|
|
assert "| Benchmark |" in report
|
|
|
|
|
|
def test_generate_markdown_report_empty():
|
|
report = generate_markdown_report([])
|
|
assert "No benchmark results" in report
|
|
|
|
|
|
# --- New tests ---
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_load_config():
|
|
"""Load a temp YAML config and verify structure."""
|
|
import yaml
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
) as f:
|
|
yaml.dump(
|
|
{
|
|
"name": "test-repo",
|
|
"url": "https://example.com/repo.git",
|
|
"commit": "HEAD",
|
|
"language": "python",
|
|
"size_category": "small",
|
|
"test_commits": [{"sha": "abc123", "description": "test"}],
|
|
"entry_points": ["main.py::main"],
|
|
"search_queries": [
|
|
{"query": "hello", "expected": "main.py::greet"}
|
|
],
|
|
},
|
|
f,
|
|
)
|
|
tmp_path = f.name
|
|
|
|
try:
|
|
import yaml as _yaml
|
|
|
|
with open(tmp_path) as fh:
|
|
config = _yaml.safe_load(fh)
|
|
|
|
assert config["name"] == "test-repo"
|
|
assert config["language"] == "python"
|
|
assert len(config["test_commits"]) == 1
|
|
assert len(config["entry_points"]) == 1
|
|
assert len(config["search_queries"]) == 1
|
|
finally:
|
|
os.unlink(tmp_path)
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_shipped_eval_configs_pin_the_latest_test_commit():
|
|
"""The cloned snapshot must contain every configured benchmark commit."""
|
|
for config in load_all_configs():
|
|
test_commits = config.get("test_commits", [])
|
|
if test_commits:
|
|
assert config["commit"] == test_commits[-1]["sha"], config["name"]
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_load_config_rejects_a_pin_before_the_latest_test_commit(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
"""An inconsistent snapshot must fail instead of yielding invalid metrics."""
|
|
import yaml
|
|
|
|
config_path = tmp_path / "bad.yaml"
|
|
config_path.write_text(
|
|
yaml.safe_dump({
|
|
"name": "bad",
|
|
"commit": "older",
|
|
"test_commits": [
|
|
{"sha": "older", "changed_files": 10},
|
|
{"sha": "newer", "changed_files": 12},
|
|
],
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr("code_review_graph.eval.runner.CONFIGS_DIR", tmp_path)
|
|
|
|
with pytest.raises(ValueError, match="latest test_commit newer"):
|
|
load_config("bad")
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_write_csv():
|
|
"""Write results to CSV and read back."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "results" / "test.csv"
|
|
results = [
|
|
{"repo": "foo", "tokens": 100, "ratio": 2.5},
|
|
{"repo": "bar", "tokens": 200, "ratio": 1.5},
|
|
]
|
|
write_csv(results, path)
|
|
|
|
assert path.exists()
|
|
with open(path, newline="") as f:
|
|
reader = csv.DictReader(f)
|
|
rows = list(reader)
|
|
|
|
assert len(rows) == 2
|
|
assert rows[0]["repo"] == "foo"
|
|
assert rows[1]["tokens"] == "200"
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_write_csv_empty():
|
|
"""Writing empty results should be a no-op."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "empty.csv"
|
|
write_csv([], path)
|
|
assert not path.exists()
|
|
|
|
|
|
def test_generate_readme_tables():
|
|
"""Feed sample CSV data and verify table format."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
results_dir = Path(tmpdir)
|
|
|
|
# Write token efficiency CSV
|
|
te_path = results_dir / "test_token_efficiency_2026-01-01.csv"
|
|
with open(te_path, "w", newline="") as f:
|
|
w = csv.DictWriter(
|
|
f,
|
|
fieldnames=[
|
|
"repo", "commit", "description", "changed_files",
|
|
"naive_tokens", "standard_tokens", "graph_tokens",
|
|
"naive_to_graph_ratio", "standard_to_graph_ratio",
|
|
],
|
|
)
|
|
w.writeheader()
|
|
w.writerow({
|
|
"repo": "myrepo", "commit": "abc", "description": "test",
|
|
"changed_files": "3", "naive_tokens": "1000",
|
|
"standard_tokens": "500", "graph_tokens": "200",
|
|
"naive_to_graph_ratio": "5.0",
|
|
"standard_to_graph_ratio": "2.5",
|
|
})
|
|
|
|
tables = generate_readme_tables(results_dir)
|
|
assert "### Token Efficiency" in tables
|
|
assert "myrepo" in tables
|
|
assert "1000" in tables
|
|
|
|
|
|
def test_generate_full_report():
|
|
"""Feed sample CSV data and verify report sections."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
results_dir = Path(tmpdir)
|
|
|
|
# Write a build_performance CSV
|
|
bp_path = results_dir / "test_build_performance_2026-01-01.csv"
|
|
with open(bp_path, "w", newline="") as f:
|
|
w = csv.DictWriter(
|
|
f,
|
|
fieldnames=[
|
|
"repo", "file_count", "node_count", "edge_count",
|
|
"flow_detection_seconds", "community_detection_seconds",
|
|
"search_avg_ms", "nodes_per_second",
|
|
],
|
|
)
|
|
w.writeheader()
|
|
w.writerow({
|
|
"repo": "testrepo", "file_count": "10", "node_count": "50",
|
|
"edge_count": "30", "flow_detection_seconds": "0.1",
|
|
"community_detection_seconds": "0.2",
|
|
"search_avg_ms": "5.0", "nodes_per_second": "500",
|
|
})
|
|
|
|
report = generate_full_report(results_dir)
|
|
assert "# Evaluation Report" in report
|
|
assert "## Methodology" in report
|
|
assert "## Build Performance" in report
|
|
assert "testrepo" in report
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_runner_with_mock_repo():
|
|
"""Create a tiny git repo with 2 Python files, run benchmarks, verify output."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "mock_repo"
|
|
repo_path.mkdir()
|
|
|
|
# Init git repo
|
|
subprocess.run(
|
|
["git", "init"], cwd=str(repo_path), capture_output=True
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "[email protected]"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
# Create two Python files
|
|
(repo_path / "main.py").write_text(
|
|
'from helper import greet\n\ndef main():\n greet("world")\n',
|
|
encoding="utf-8",
|
|
)
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hello {name}")\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "initial"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
# Second commit: modify helper.py
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hi {name}!")\n',
|
|
encoding="utf-8",
|
|
)
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "update greeting"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
# Build graph
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import full_build, get_db_path
|
|
|
|
db_path = get_db_path(repo_path)
|
|
store = GraphStore(db_path)
|
|
full_build(repo_path, store)
|
|
|
|
config = {
|
|
"name": "mock",
|
|
"language": "python",
|
|
"test_commits": [
|
|
{"sha": "HEAD", "description": "update greeting"},
|
|
],
|
|
"entry_points": ["main.py::main"],
|
|
"search_queries": [
|
|
{"query": "greet", "expected": "helper.py::greet"},
|
|
],
|
|
}
|
|
|
|
# Run token_efficiency
|
|
from code_review_graph.eval.benchmarks import token_efficiency
|
|
te_results = token_efficiency.run(repo_path, store, config)
|
|
assert len(te_results) >= 1
|
|
assert "naive_tokens" in te_results[0]
|
|
assert "graph_tokens" in te_results[0]
|
|
|
|
# Run impact_accuracy
|
|
from code_review_graph.eval.benchmarks import impact_accuracy
|
|
ia_results = impact_accuracy.run(repo_path, store, config)
|
|
assert len(ia_results) >= 1
|
|
assert "precision" in ia_results[0]
|
|
assert "f1" in ia_results[0]
|
|
|
|
# Run search_quality
|
|
from code_review_graph.eval.benchmarks import search_quality
|
|
sq_results = search_quality.run(repo_path, store, config)
|
|
assert len(sq_results) == 1
|
|
assert "reciprocal_rank" in sq_results[0]
|
|
|
|
# Run build_performance
|
|
from code_review_graph.eval.benchmarks import build_performance
|
|
bp_results = build_performance.run(repo_path, store, config)
|
|
assert len(bp_results) == 1
|
|
assert "node_count" in bp_results[0]
|
|
assert bp_results[0]["node_count"] > 0
|
|
|
|
store.close()
|
|
|
|
|
|
# --- Token benchmark tests ---
|
|
|
|
|
|
def test_estimate_tokens_basic():
|
|
"""estimate_tokens should return a reasonable approximation."""
|
|
from code_review_graph.eval.token_benchmark import estimate_tokens
|
|
|
|
# Simple string: "hello" => JSON '"hello"' (7 chars) => 7 // 4 = 1
|
|
assert estimate_tokens("hello") == 1
|
|
|
|
# Dict: {"a": 1} => '{"a": 1}' (8 chars) => 8 // 4 = 2
|
|
assert estimate_tokens({"a": 1}) == 2
|
|
|
|
# Longer content should scale proportionally
|
|
long_text = "x" * 400
|
|
tokens = estimate_tokens(long_text)
|
|
# JSON adds 2 quote chars: (400 + 2) // 4 = 100
|
|
assert tokens == 100
|
|
|
|
|
|
def test_estimate_tokens_nested():
|
|
"""estimate_tokens handles nested structures."""
|
|
from code_review_graph.eval.token_benchmark import estimate_tokens
|
|
|
|
nested = {"nodes": [{"name": "foo"}, {"name": "bar"}], "count": 2}
|
|
tokens = estimate_tokens(nested)
|
|
assert tokens > 0
|
|
assert isinstance(tokens, int)
|
|
|
|
|
|
def test_estimate_tokens_non_serializable():
|
|
"""estimate_tokens uses default=str for non-serializable objects."""
|
|
from pathlib import Path
|
|
|
|
from code_review_graph.eval.token_benchmark import estimate_tokens
|
|
|
|
# Path objects are not JSON-serializable but default=str handles them
|
|
tokens = estimate_tokens({"path": Path("/tmp/test")})
|
|
assert tokens > 0
|
|
|
|
|
|
def test_benchmark_review_workflow():
|
|
"""benchmark_review_workflow completes and returns expected structure."""
|
|
from code_review_graph.eval.token_benchmark import benchmark_review_workflow
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "bench_repo"
|
|
repo_path.mkdir()
|
|
|
|
# Init git repo with two commits
|
|
subprocess.run(
|
|
["git", "init"], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "[email protected]"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
(repo_path / "main.py").write_text(
|
|
'from helper import greet\n\ndef main():\n greet("world")\n',
|
|
encoding="utf-8",
|
|
)
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hello {name}")\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "initial"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
# Second commit
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hi {name}!")\n',
|
|
encoding="utf-8",
|
|
)
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "update greeting"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
# Build graph
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import full_build, get_db_path
|
|
|
|
db_path = get_db_path(repo_path)
|
|
store = GraphStore(db_path)
|
|
full_build(repo_path, store)
|
|
store.close()
|
|
|
|
# Run the review benchmark
|
|
result = benchmark_review_workflow(
|
|
repo_root=str(repo_path), base="HEAD~1",
|
|
)
|
|
|
|
assert result["workflow"] == "review"
|
|
assert result["total_tokens"] > 0
|
|
assert result["tool_calls"] == 2
|
|
assert len(result["calls"]) == 2
|
|
assert result["calls"][0]["tool"] == "get_minimal_context"
|
|
assert result["calls"][1]["tool"] == "detect_changes_minimal"
|
|
for call in result["calls"]:
|
|
assert call["tokens"] >= 0
|
|
|
|
|
|
def test_run_all_benchmarks():
|
|
"""run_all_benchmarks returns results for all workflows."""
|
|
from code_review_graph.eval.token_benchmark import run_all_benchmarks
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "all_bench_repo"
|
|
repo_path.mkdir()
|
|
|
|
subprocess.run(
|
|
["git", "init"], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "[email protected]"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
(repo_path / "app.py").write_text(
|
|
'def main():\n print("hello")\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "initial"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
(repo_path / "app.py").write_text(
|
|
'def main():\n print("hi")\n',
|
|
encoding="utf-8",
|
|
)
|
|
subprocess.run(
|
|
["git", "add", "."], cwd=str(repo_path), capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "update"],
|
|
cwd=str(repo_path), capture_output=True,
|
|
)
|
|
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import full_build, get_db_path
|
|
|
|
db_path = get_db_path(repo_path)
|
|
store = GraphStore(db_path)
|
|
full_build(repo_path, store)
|
|
store.close()
|
|
|
|
results = run_all_benchmarks(repo_root=str(repo_path), base="HEAD~1")
|
|
|
|
# Should have one result per workflow (5 total)
|
|
assert len(results) == 5
|
|
|
|
workflow_names = {r["workflow"] for r in results}
|
|
assert workflow_names == {
|
|
"review", "architecture", "debug", "onboard", "pre_merge",
|
|
}
|
|
|
|
# Each successful result should have total_tokens
|
|
for r in results:
|
|
if "error" not in r:
|
|
assert r["total_tokens"] >= 0
|
|
assert "calls" in r
|
|
|
|
|
|
# --- Failure-inflation regression tests + agent_baseline + co-change mode ---
|
|
|
|
|
|
def _git(repo_path, *args):
|
|
subprocess.run(["git", *args], cwd=str(repo_path), capture_output=True)
|
|
|
|
|
|
def _make_repo(tmpdir, two_file_commit=False):
|
|
"""Tiny git repo: initial commit, then a second commit touching 1 or 2 files."""
|
|
repo_path = Path(tmpdir) / "mock_repo"
|
|
repo_path.mkdir()
|
|
_git(repo_path, "init")
|
|
_git(repo_path, "config", "user.email", "[email protected]")
|
|
_git(repo_path, "config", "user.name", "Test")
|
|
|
|
(repo_path / "main.py").write_text(
|
|
'from helper import greet\n\ndef main():\n greet("world")\n',
|
|
encoding="utf-8",
|
|
)
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hello {name}")\n',
|
|
encoding="utf-8",
|
|
)
|
|
_git(repo_path, "add", ".")
|
|
_git(repo_path, "commit", "-m", "initial")
|
|
|
|
(repo_path / "helper.py").write_text(
|
|
'def greet(name):\n print(f"Hi {name}!")\n',
|
|
encoding="utf-8",
|
|
)
|
|
if two_file_commit:
|
|
(repo_path / "main.py").write_text(
|
|
'from helper import greet\n\ndef main():\n greet("there")\n',
|
|
encoding="utf-8",
|
|
)
|
|
_git(repo_path, "add", ".")
|
|
_git(repo_path, "commit", "-m", "update greeting")
|
|
return repo_path
|
|
|
|
|
|
def _build_store(repo_path):
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import full_build, get_db_path
|
|
|
|
store = GraphStore(get_db_path(repo_path))
|
|
full_build(repo_path, store)
|
|
return store
|
|
|
|
|
|
def _mock_config(**extra):
|
|
config = {
|
|
"name": "mock",
|
|
"language": "python",
|
|
"test_commits": [{"sha": "HEAD", "description": "update greeting"}],
|
|
"entry_points": ["main.py::main"],
|
|
"search_queries": [{"query": "greet", "expected": "helper.py::greet"}],
|
|
}
|
|
config.update(extra)
|
|
return config
|
|
|
|
|
|
def test_token_efficiency_failure_marked_error_not_inflated(monkeypatch):
|
|
"""A thrown get_review_context must yield status=error, not ratio=naive/1."""
|
|
from code_review_graph.eval.benchmarks import token_efficiency
|
|
|
|
def _boom(**kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("code_review_graph.tools.get_review_context", _boom)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = token_efficiency.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) >= 1
|
|
for row in results:
|
|
assert row["status"] == "error"
|
|
assert "boom" in row["error"]
|
|
# Failed measurements must not look like valid (inflated) ratios.
|
|
assert row["graph_tokens"] == ""
|
|
assert row["naive_to_graph_ratio"] == ""
|
|
assert row["standard_to_graph_ratio"] == ""
|
|
|
|
agg = token_efficiency.aggregate(results)
|
|
assert agg["ok_rows"] == 0
|
|
assert agg["error_rows"] == len(results)
|
|
assert agg["median_naive_to_graph_ratio"] is None
|
|
|
|
|
|
def test_token_efficiency_success_rows_status_ok():
|
|
from code_review_graph.eval.benchmarks import token_efficiency
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = token_efficiency.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) >= 1
|
|
for row in results:
|
|
assert row["status"] == "ok"
|
|
assert row["error"] == ""
|
|
assert isinstance(row["graph_tokens"], int)
|
|
assert isinstance(row["naive_to_graph_ratio"], float)
|
|
|
|
agg = token_efficiency.aggregate(results)
|
|
assert agg["ok_rows"] == len(results)
|
|
assert agg["error_rows"] == 0
|
|
assert isinstance(agg["median_naive_to_graph_ratio"], float)
|
|
|
|
|
|
def test_impact_accuracy_failure_marked_error_not_perfect_recall(monkeypatch):
|
|
"""A thrown analyze_changes must not silently score recall 1.0."""
|
|
from code_review_graph.eval.benchmarks import impact_accuracy
|
|
|
|
def _boom(*args, **kwargs):
|
|
raise RuntimeError("analysis exploded")
|
|
|
|
monkeypatch.setattr("code_review_graph.changes.analyze_changes", _boom)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir, two_file_commit=True)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = impact_accuracy.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) >= 2 # both modes attempted, both failed
|
|
for row in results:
|
|
assert row["status"] == "error"
|
|
assert "analysis exploded" in row["error"]
|
|
assert row["recall"] == "" # NOT 1.0
|
|
assert row["precision"] == ""
|
|
assert row["f1"] == ""
|
|
|
|
agg = impact_accuracy.aggregate(results)
|
|
assert agg["graph_derived"]["ok_rows"] == 0
|
|
assert agg["co_change"]["ok_rows"] == 0
|
|
assert agg["graph_derived"]["mean_recall"] is None
|
|
assert agg["error_rows"] == len(results)
|
|
|
|
|
|
def test_impact_accuracy_emits_both_ground_truth_modes():
|
|
from code_review_graph.eval.benchmarks import impact_accuracy
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir, two_file_commit=True)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = impact_accuracy.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
modes = {r["ground_truth_mode"] for r in results}
|
|
assert impact_accuracy.MODE_GRAPH_DERIVED in modes
|
|
assert impact_accuracy.MODE_CO_CHANGE in modes
|
|
|
|
graph_rows = [
|
|
r for r in results
|
|
if r["ground_truth_mode"] == impact_accuracy.MODE_GRAPH_DERIVED
|
|
]
|
|
co_rows = [
|
|
r for r in results
|
|
if r["ground_truth_mode"] == impact_accuracy.MODE_CO_CHANGE
|
|
]
|
|
|
|
for row in graph_rows:
|
|
assert row["status"] == "ok"
|
|
assert 0.0 <= row["recall"] <= 1.0
|
|
assert row["seed_file"] == ""
|
|
|
|
# Commit touched helper.py + main.py: seed is the sorted-first file and
|
|
# the ground truth is the *other* co-changed file — independent of the graph.
|
|
assert len(co_rows) == 1
|
|
co = co_rows[0]
|
|
assert co["status"] == "ok"
|
|
assert co["seed_file"] == "helper.py"
|
|
assert co["actual_files"] == 1
|
|
assert 0.0 <= co["precision"] <= 1.0
|
|
assert 0.0 <= co["recall"] <= 1.0
|
|
|
|
|
|
def test_impact_accuracy_co_change_skipped_for_single_file_commit():
|
|
from code_review_graph.eval.benchmarks import impact_accuracy
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir, two_file_commit=False)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = impact_accuracy.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
co_rows = [
|
|
r for r in results
|
|
if r["ground_truth_mode"] == impact_accuracy.MODE_CO_CHANGE
|
|
]
|
|
assert len(co_rows) == 1
|
|
assert co_rows[0]["status"] == "skipped"
|
|
assert "co-changed" in co_rows[0]["error"]
|
|
|
|
agg = impact_accuracy.aggregate(results)
|
|
assert agg["skipped_rows"] == 1
|
|
assert agg["co_change"]["ok_rows"] == 0
|
|
|
|
|
|
# --- agent_baseline benchmark ---
|
|
|
|
|
|
def test_derive_search_terms_extracts_identifiers_and_keywords():
|
|
from code_review_graph.eval.benchmarks.agent_baseline import derive_search_terms
|
|
|
|
terms = derive_search_terms("How does Client.request send an HTTP request?")
|
|
assert "client.request" in terms
|
|
assert "how" not in terms # stopword
|
|
assert "does" not in terms # stopword
|
|
assert all(t == t.lower() for t in terms)
|
|
|
|
|
|
def test_grep_rank_orders_by_match_count_and_takes_top_k():
|
|
from code_review_graph.eval.benchmarks.agent_baseline import grep_rank
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
corpus = Path(tmpdir)
|
|
(corpus / "a.py").write_text("greet()\ngreet()\ngreet()\n", encoding="utf-8")
|
|
(corpus / "b.py").write_text("greet()\n", encoding="utf-8")
|
|
(corpus / "c.py").write_text("nothing here\n", encoding="utf-8")
|
|
(corpus / "d.txt").write_text("greet greet greet greet\n", encoding="utf-8")
|
|
sub = corpus / "node_modules"
|
|
sub.mkdir()
|
|
(sub / "e.py").write_text("greet greet greet greet greet\n", encoding="utf-8")
|
|
|
|
ranked = grep_rank(corpus, ["greet"], k=3)
|
|
# d.txt (non-source ext) and node_modules/e.py (skipped dir) excluded
|
|
assert ranked == [("a.py", 3), ("b.py", 1)]
|
|
|
|
top1 = grep_rank(corpus, ["greet"], k=1)
|
|
assert top1 == [("a.py", 3)]
|
|
|
|
assert grep_rank(corpus, [], k=3) == []
|
|
|
|
|
|
def test_grep_rank_tie_breaks_on_path():
|
|
from code_review_graph.eval.benchmarks.agent_baseline import grep_rank
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
corpus = Path(tmpdir)
|
|
(corpus / "zz.py").write_text("token token\n", encoding="utf-8")
|
|
(corpus / "aa.py").write_text("token token\n", encoding="utf-8")
|
|
ranked = grep_rank(corpus, ["token"], k=2)
|
|
assert ranked == [("aa.py", 2), ("zz.py", 2)]
|
|
|
|
|
|
def test_agent_baseline_run_with_mock_repo():
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
config = _mock_config(
|
|
agent_questions=["How does greet print a greeting"],
|
|
)
|
|
try:
|
|
results = agent_baseline.run(repo_path, store, config)
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) == 1
|
|
row = results[0]
|
|
assert row["question"] == "How does greet print a greeting"
|
|
assert "greet" in row["terms"]
|
|
assert row["files_matched"] >= 1
|
|
assert "helper.py" in row["top_files"]
|
|
assert row["baseline_tokens"] > 0
|
|
assert row["status"] in ("ok", "no_graph_results")
|
|
if row["status"] == "ok":
|
|
assert isinstance(row["baseline_to_graph_ratio"], float)
|
|
|
|
|
|
def test_agent_baseline_falls_back_to_search_queries():
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
results = agent_baseline.run(repo_path, store, _mock_config())
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["question"] == "greet"
|
|
|
|
|
|
def test_agent_baseline_search_failure_marked_error(monkeypatch):
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
def _boom(*args, **kwargs):
|
|
raise RuntimeError("search down")
|
|
|
|
monkeypatch.setattr("code_review_graph.search.hybrid_search", _boom)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
config = _mock_config(agent_questions=["How does greet work"])
|
|
try:
|
|
results = agent_baseline.run(repo_path, store, config)
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["status"] == "error"
|
|
assert "search down" in results[0]["error"]
|
|
assert results[0]["baseline_to_graph_ratio"] == ""
|
|
|
|
agg = agent_baseline.aggregate(results)
|
|
assert agg["ok_rows"] == 0
|
|
assert agg["error_rows"] == 1
|
|
assert agg["median_baseline_to_graph_ratio"] is None
|
|
|
|
|
|
def test_agent_baseline_aggregate_excludes_non_ok_rows():
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
rows = [
|
|
{"status": "ok", "baseline_to_graph_ratio": 4.0},
|
|
{"status": "ok", "baseline_to_graph_ratio": 8.0},
|
|
{"status": "error", "baseline_to_graph_ratio": ""},
|
|
{"status": "no_graph_results", "baseline_to_graph_ratio": ""},
|
|
]
|
|
agg = agent_baseline.aggregate(rows)
|
|
assert agg["total_rows"] == 4
|
|
assert agg["ok_rows"] == 2
|
|
assert agg["error_rows"] == 1
|
|
assert agg["median_baseline_to_graph_ratio"] == 6.0
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_agent_baseline_registered_in_runner():
|
|
from code_review_graph.eval.runner import BENCHMARK_REGISTRY
|
|
|
|
assert "agent_baseline" in BENCHMARK_REGISTRY
|
|
|
|
|
|
def test_reporter_impact_f1_skips_error_and_co_change_rows():
|
|
"""Table B must aggregate only ok graph-derived rows."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
results_dir = Path(tmpdir)
|
|
ia_path = results_dir / "mock_impact_accuracy_2026-01-01.csv"
|
|
fieldnames = [
|
|
"repo", "commit", "ground_truth_mode", "seed_file",
|
|
"predicted_files", "actual_files", "true_positives",
|
|
"precision", "recall", "f1", "status", "error",
|
|
]
|
|
with open(ia_path, "w", newline="") as f:
|
|
w = csv.DictWriter(f, fieldnames=fieldnames)
|
|
w.writeheader()
|
|
w.writerow({
|
|
"repo": "mock", "commit": "abc",
|
|
"ground_truth_mode": "graph-derived (circular — upper bound)",
|
|
"seed_file": "", "predicted_files": "2", "actual_files": "2",
|
|
"true_positives": "1", "precision": "0.5", "recall": "0.5",
|
|
"f1": "0.5", "status": "ok", "error": "",
|
|
})
|
|
w.writerow({
|
|
"repo": "mock", "commit": "def",
|
|
"ground_truth_mode": "graph-derived (circular — upper bound)",
|
|
"seed_file": "", "predicted_files": "", "actual_files": "",
|
|
"true_positives": "", "precision": "", "recall": "",
|
|
"f1": "", "status": "error", "error": "boom",
|
|
})
|
|
w.writerow({
|
|
"repo": "mock", "commit": "abc",
|
|
"ground_truth_mode": "co-change (same commit, seed excluded)",
|
|
"seed_file": "a.py", "predicted_files": "1", "actual_files": "1",
|
|
"true_positives": "1", "precision": "1.0", "recall": "1.0",
|
|
"f1": "0.9", "status": "ok", "error": "",
|
|
})
|
|
|
|
tables = generate_readme_tables(results_dir)
|
|
|
|
# 0.5 comes only from the single ok graph-derived row; the error row and
|
|
# the co-change row (different metric) must not pollute the column.
|
|
assert "0.5" in tables
|
|
assert "0.9" not in tables
|
|
|
|
|
|
def test_eval_embed_bootstraps_vectors_and_returns_real_graph_results(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
"""The public eval path must build vectors that its semantic benchmark can use."""
|
|
from code_review_graph.eval import runner
|
|
|
|
repo_path = _make_repo(tmp_path)
|
|
helper = repo_path / "helper.py"
|
|
helper.write_text(
|
|
"# salutation_marker appears only in source text, not in graph node names\n"
|
|
+ helper.read_text(encoding="utf-8"),
|
|
encoding="utf-8",
|
|
)
|
|
config = _mock_config(
|
|
agent_questions=["Where is salutation_marker handled?"],
|
|
)
|
|
monkeypatch.setattr(runner, "load_config", lambda _name: config)
|
|
monkeypatch.setattr(runner, "clone_or_update", lambda _config: repo_path)
|
|
monkeypatch.setenv("CRG_SERIAL_PARSE", "1")
|
|
|
|
state_dir = tmp_path / "state"
|
|
monkeypatch.setenv("CRG_HOME", str(state_dir))
|
|
from code_review_graph import registry as registry_module
|
|
|
|
monkeypatch.setattr(
|
|
registry_module,
|
|
"_REGISTRY_PATH",
|
|
state_dir / "registry.json",
|
|
raising=False,
|
|
)
|
|
|
|
class _StubProvider:
|
|
dimension = 2
|
|
|
|
def __init__(self, name):
|
|
self.name = name
|
|
|
|
@staticmethod
|
|
def embed(texts):
|
|
return [[float(len(text)), 1.0] for text in texts]
|
|
|
|
@staticmethod
|
|
def embed_query(_text):
|
|
return [1.0, 0.0]
|
|
|
|
monkeypatch.setattr(
|
|
"code_review_graph.embeddings.get_provider",
|
|
lambda provider=None, model=None: _StubProvider(
|
|
f"{provider or 'local'}:{model or 'default'}",
|
|
),
|
|
)
|
|
|
|
results = runner.run_eval(
|
|
repos=["mock"],
|
|
benchmarks=["agent_baseline"],
|
|
output_dir=tmp_path / "results",
|
|
embed=True,
|
|
embedding_provider="local",
|
|
embedding_model="eval-test",
|
|
)
|
|
|
|
rows = results["mock_agent_baseline"]
|
|
assert len(rows) == 1
|
|
assert rows[0]["status"] == "ok"
|
|
assert rows[0]["graph_tokens"] > 0
|
|
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import get_db_path
|
|
|
|
store = GraphStore(get_db_path(repo_path))
|
|
try:
|
|
assert runner._embedding_count(store) > 0
|
|
finally:
|
|
store.close()
|
|
|
|
|
|
def test_search_quality_uses_the_index_provider_and_model(monkeypatch, tmp_path):
|
|
"""A custom eval index is useless unless benchmark queries select that identity."""
|
|
from code_review_graph.eval.benchmarks import search_quality
|
|
|
|
observed = {}
|
|
|
|
def _search(_store, _query, *, limit, provider=None, model=None):
|
|
observed.update(provider=provider, model=model, limit=limit)
|
|
return []
|
|
|
|
monkeypatch.setattr("code_review_graph.search.hybrid_search", _search)
|
|
search_quality.run(
|
|
tmp_path,
|
|
object(),
|
|
{
|
|
"name": "mock",
|
|
"search_queries": [{"query": "natural language", "expected": "target"}],
|
|
"_embedding_provider": "google",
|
|
"_embedding_model": "text-embedding-test",
|
|
},
|
|
)
|
|
|
|
assert observed == {
|
|
"provider": "google",
|
|
"model": "text-embedding-test",
|
|
"limit": 20,
|
|
}
|
|
|
|
|
|
def test_multi_hop_uses_the_index_provider_and_model(monkeypatch, tmp_path):
|
|
from code_review_graph.eval.benchmarks import multi_hop_retrieval
|
|
|
|
observed = {}
|
|
|
|
def _search(_store, _query, *, limit, provider=None, model=None):
|
|
observed.update(provider=provider, model=model, limit=limit)
|
|
return []
|
|
|
|
monkeypatch.setattr("code_review_graph.search.hybrid_search", _search)
|
|
multi_hop_retrieval.run(
|
|
tmp_path,
|
|
object(),
|
|
{
|
|
"name": "mock",
|
|
"multi_hop_tasks": [
|
|
{
|
|
"id": "task",
|
|
"nl_query": "natural language",
|
|
"anchor_qualified_suffix": "::target",
|
|
"k": 7,
|
|
},
|
|
],
|
|
"_embedding_provider": "minimax",
|
|
"_embedding_model": "embedding-01",
|
|
},
|
|
)
|
|
|
|
assert observed == {
|
|
"provider": "minimax",
|
|
"model": "embedding-01",
|
|
"limit": 7,
|
|
}
|
|
|
|
|
|
def test_eval_closes_graph_store_when_embedding_bootstrap_fails(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
"""A provider failure must not leave the evaluation database connection open."""
|
|
from code_review_graph.eval import runner
|
|
from code_review_graph.graph import GraphStore
|
|
|
|
repo_path = _make_repo(tmp_path)
|
|
config = _mock_config()
|
|
monkeypatch.setattr(runner, "load_config", lambda _name: config)
|
|
monkeypatch.setattr(runner, "clone_or_update", lambda _config: repo_path)
|
|
monkeypatch.setenv("CRG_SERIAL_PARSE", "1")
|
|
monkeypatch.setenv("CRG_HOME", str(tmp_path / "state"))
|
|
|
|
def _boom(*_args, **_kwargs):
|
|
raise RuntimeError("provider failed")
|
|
|
|
monkeypatch.setattr(runner, "_build_embedding_index", _boom)
|
|
closed = []
|
|
original_close = GraphStore.close
|
|
|
|
def _track_close(self):
|
|
closed.append(self.db_path)
|
|
original_close(self)
|
|
|
|
monkeypatch.setattr(GraphStore, "close", _track_close)
|
|
|
|
with pytest.raises(RuntimeError, match="provider failed"):
|
|
runner.run_eval(
|
|
repos=["mock"],
|
|
benchmarks=["agent_baseline"],
|
|
output_dir=tmp_path / "results",
|
|
embed=True,
|
|
embedding_provider="local",
|
|
embedding_model="eval-test",
|
|
)
|
|
|
|
assert closed
|
|
|
|
|
|
# -- Semantic index guard (agent_baseline and friends) ---------------------
|
|
|
|
|
|
def test_agent_baseline_aggregate_reports_excluded_rows():
|
|
"""A run where the graph answered nothing must not read as 'no result'.
|
|
|
|
``ok_rows == 0`` with ``median is None`` is ambiguous on its own: it looks
|
|
the same whether zero questions were asked or every query came back empty.
|
|
The excluded-row counts disambiguate it.
|
|
"""
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
results = [
|
|
{"status": "no_graph_results", "baseline_to_graph_ratio": ""},
|
|
{"status": "no_graph_results", "baseline_to_graph_ratio": ""},
|
|
{"status": "no_baseline_match", "baseline_to_graph_ratio": ""},
|
|
]
|
|
agg = agent_baseline.aggregate(results)
|
|
|
|
assert agg["ok_rows"] == 0
|
|
assert agg["median_baseline_to_graph_ratio"] is None
|
|
assert agg["no_graph_results_rows"] == 2
|
|
assert agg["no_baseline_match_rows"] == 1
|
|
|
|
|
|
def test_agent_baseline_aggregate_counts_zero_on_a_healthy_run():
|
|
from code_review_graph.eval.benchmarks import agent_baseline
|
|
|
|
agg = agent_baseline.aggregate([
|
|
{"status": "ok", "baseline_to_graph_ratio": "10.0"},
|
|
{"status": "ok", "baseline_to_graph_ratio": "20.0"},
|
|
])
|
|
|
|
assert agg["ok_rows"] == 2
|
|
assert agg["no_graph_results_rows"] == 0
|
|
assert agg["no_baseline_match_rows"] == 0
|
|
assert agg["median_baseline_to_graph_ratio"] == 15.0
|
|
|
|
|
|
def test_warns_when_semantic_benchmark_runs_without_a_vector_index(caplog):
|
|
"""The silent-zero path must announce itself before the benchmark runs."""
|
|
import logging
|
|
|
|
from code_review_graph.eval.runner import _warn_if_semantic_index_missing
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
with caplog.at_level(logging.WARNING):
|
|
_warn_if_semantic_index_missing(store, ["agent_baseline"])
|
|
finally:
|
|
store.close()
|
|
|
|
assert "no vector index" in caplog.text
|
|
assert "--embed" in caplog.text
|
|
|
|
|
|
def test_no_warning_for_benchmarks_that_do_not_use_semantic_search(caplog):
|
|
import logging
|
|
|
|
from code_review_graph.eval.runner import _warn_if_semantic_index_missing
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = _make_repo(tmpdir)
|
|
store = _build_store(repo_path)
|
|
try:
|
|
with caplog.at_level(logging.WARNING):
|
|
_warn_if_semantic_index_missing(store, ["token_efficiency"])
|
|
finally:
|
|
store.close()
|
|
|
|
assert "no vector index" not in caplog.text
|
|
|
|
|
|
def test_no_warning_once_the_index_is_populated(caplog, monkeypatch):
|
|
import logging
|
|
|
|
from code_review_graph.eval import runner
|
|
|
|
monkeypatch.setattr(runner, "_embedding_count", lambda store: 42)
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
runner._warn_if_semantic_index_missing(object(), ["agent_baseline"])
|
|
|
|
assert "no vector index" not in caplog.text
|
|
|
|
|
|
def test_embedding_count_reraises_non_missing_table_errors():
|
|
"""A lock or a corrupt database must not be reported as 'no index'.
|
|
|
|
Reclassifying it would tell the user to re-run with --embed and send
|
|
them after the wrong problem.
|
|
"""
|
|
import sqlite3
|
|
|
|
from code_review_graph.eval.runner import _embedding_count
|
|
|
|
class _Boom:
|
|
class _Conn:
|
|
@staticmethod
|
|
def execute(*_args, **_kwargs):
|
|
raise sqlite3.OperationalError("database is locked")
|
|
|
|
_conn = _Conn()
|
|
|
|
with pytest.raises(sqlite3.OperationalError, match="locked"):
|
|
_embedding_count(_Boom())
|
|
|
|
|
|
def test_embedding_count_returns_none_for_a_missing_table():
|
|
import sqlite3
|
|
|
|
from code_review_graph.eval.runner import _embedding_count
|
|
|
|
class _NoTable:
|
|
class _Conn:
|
|
@staticmethod
|
|
def execute(*_args, **_kwargs):
|
|
raise sqlite3.OperationalError("no such table: embeddings")
|
|
|
|
_conn = _Conn()
|
|
|
|
assert _embedding_count(_NoTable()) is None
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed")
|
|
def test_clone_or_update_refuses_directory_inside_another_repo(tmp_path):
|
|
"""A target dir that is not its own repo must never be checked out.
|
|
|
|
``evaluate/test_repos/`` lives inside this project. If a target directory
|
|
exists but is not a git repository in its own right, ``git -C <dir>`` walks
|
|
up to the *enclosing* checkout, so ``git checkout <pinned sha>`` would
|
|
rewrite the developer's working tree instead of the test repo.
|
|
"""
|
|
from code_review_graph.eval.runner import clone_or_update
|
|
|
|
outer = tmp_path / "outer"
|
|
outer.mkdir()
|
|
git = ["git", "-c", "[email protected]", "-c", "user.name=t"]
|
|
subprocess.run(["git", "init", "-q"], cwd=outer, check=True)
|
|
(outer / "tracked.txt").write_text("first")
|
|
subprocess.run(["git", "add", "-A"], cwd=outer, check=True)
|
|
subprocess.run(git + ["commit", "-qm", "first"], cwd=outer, check=True)
|
|
first = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=outer, capture_output=True, text=True
|
|
).stdout.strip()
|
|
(outer / "tracked.txt").write_text("second")
|
|
subprocess.run(["git", "add", "-A"], cwd=outer, check=True)
|
|
subprocess.run(git + ["commit", "-qm", "second"], cwd=outer, check=True)
|
|
head_before = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=outer, capture_output=True, text=True
|
|
).stdout.strip()
|
|
|
|
repos_dir = outer / "evaluate" / "test_repos"
|
|
(repos_dir / "victim").mkdir(parents=True) # exists, but is not its own repo
|
|
|
|
# Pinning the *first* commit means a successful checkout would move the
|
|
# enclosing repo's HEAD -- exactly the data-loss case.
|
|
config = {"name": "victim", "url": "https://example.invalid/x.git", "commit": first}
|
|
with pytest.raises(RuntimeError, match="standalone git repository"):
|
|
clone_or_update(config, repos_dir)
|
|
|
|
head_after = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=outer, capture_output=True, text=True
|
|
).stdout.strip()
|
|
assert head_after == head_before, "enclosing repository was checked out"
|
|
assert (outer / "tracked.txt").read_text() == "second"
|