2039 lines
82 KiB
Python
2039 lines
82 KiB
Python
#!/ usr / bin / env python3
|
|
"""Regression tests for the shared benchmark evaluation semantics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "benchmarks"))
|
|
|
|
run_project_eval = importlib.import_module("run_project_eval")
|
|
run_ab_evaluation = importlib.import_module("run_ab_evaluation")
|
|
runner_common = importlib.import_module("runner_common")
|
|
flaky_evidence = importlib.import_module("fold_eval.flaky_evidence")
|
|
select_impacted_tests = importlib.import_module("select_impacted_tests")
|
|
rust_toolchain_skew = importlib.import_module("check_rust_toolchain_skew")
|
|
|
|
FOLD_QUALITY_PATH = ROOT / "scripts" / "fold_quality.py"
|
|
_fold_quality_spec = importlib.util.spec_from_file_location(
|
|
"fold_quality", FOLD_QUALITY_PATH
|
|
)
|
|
if _fold_quality_spec is None or _fold_quality_spec.loader is None:
|
|
raise ImportError(f"failed to load {FOLD_QUALITY_PATH}")
|
|
fold_quality = importlib.util.module_from_spec(_fold_quality_spec)
|
|
_fold_quality_spec.loader.exec_module(fold_quality)
|
|
|
|
SOURCE_STRUCTURE_PATH = ROOT / "scripts" / "check_source_structure.py"
|
|
_source_structure_spec = importlib.util.spec_from_file_location(
|
|
"check_source_structure", SOURCE_STRUCTURE_PATH
|
|
)
|
|
if _source_structure_spec is None or _source_structure_spec.loader is None:
|
|
raise ImportError(f"failed to load {SOURCE_STRUCTURE_PATH}")
|
|
check_source_structure = importlib.util.module_from_spec(_source_structure_spec)
|
|
_source_structure_spec.loader.exec_module(check_source_structure)
|
|
|
|
LINT_BASELINE_PATH = ROOT / "scripts" / "compare_lint_baseline.py"
|
|
_lint_baseline_spec = importlib.util.spec_from_file_location(
|
|
"compare_lint_baseline", LINT_BASELINE_PATH
|
|
)
|
|
if _lint_baseline_spec is None or _lint_baseline_spec.loader is None:
|
|
raise ImportError(f"failed to load {LINT_BASELINE_PATH}")
|
|
compare_lint_baseline = importlib.util.module_from_spec(_lint_baseline_spec)
|
|
_lint_baseline_spec.loader.exec_module(compare_lint_baseline)
|
|
|
|
confirmation = importlib.import_module("fold_eval.confirmation")
|
|
differential = importlib.import_module("fold_eval.differential")
|
|
dispositions = importlib.import_module("fold_eval.dispositions")
|
|
metamorphic = importlib.import_module("fold_eval.metamorphic")
|
|
metrics = importlib.import_module("fold_eval.metrics")
|
|
performance = importlib.import_module("fold_eval.performance")
|
|
project = importlib.import_module("fold_eval.project")
|
|
sarif_helpers = importlib.import_module("fold_eval.sarif")
|
|
schema = importlib.import_module("fold_eval.schema")
|
|
scoring = importlib.import_module("fold_eval.scoring")
|
|
stability = importlib.import_module("fold_eval.stability")
|
|
token_efficiency = importlib.import_module("fold_eval.token_efficiency")
|
|
|
|
summarize_confirmation_results = confirmation.summarize_confirmation_results
|
|
validate_confirmation_progress = confirmation.validate_confirmation_progress
|
|
fixed_runtime_reported = differential.fixed_runtime_reported
|
|
vulnerable_runtime_detected = differential.vulnerable_runtime_detected
|
|
compare_dispositions = dispositions.compare_dispositions
|
|
configuration_fingerprint = dispositions.configuration_fingerprint
|
|
validate_disposition_baseline = dispositions.validate_disposition_baseline
|
|
compare_metamorphic_findings = metamorphic.compare_metamorphic_findings
|
|
aggregate_classification_metrics = metrics.aggregate_classification_metrics
|
|
bootstrap_interval = metrics.bootstrap_interval
|
|
classification_metrics = metrics.classification_metrics
|
|
paired_comparison = metrics.paired_comparison
|
|
compare_performance_lanes = performance.compare_performance_lanes
|
|
optimization_surface = performance.optimization_surface
|
|
manifest_configuration = project.manifest_configuration
|
|
manifest_fingerprint = project.manifest_fingerprint
|
|
validate_project_manifest = project.validate_project_manifest
|
|
normalize_uri = sarif_helpers.normalize_uri
|
|
summarize_sarif_findings = sarif_helpers.summarize_sarif_findings
|
|
artifact_record = schema.artifact_record
|
|
build_run_artifact = schema.build_run_artifact
|
|
validate_benchmark_case = schema.validate_benchmark_case
|
|
validate_run_artifact = schema.validate_run_artifact
|
|
path_matches = scoring.path_matches
|
|
score_grouped_findings = scoring.score_grouped_findings
|
|
score_labeled_cases = scoring.score_labeled_cases
|
|
negative_cases_from_universe = scoring.negative_cases_from_universe
|
|
stability_summary = stability.stability_summary
|
|
artifact_token_efficiency = token_efficiency.artifact_token_efficiency
|
|
estimate_text_tokens = token_efficiency.estimate_text_tokens
|
|
|
|
|
|
class ClassificationMetricsTest(unittest.TestCase):
|
|
def test_sanitizer_oracle_does_not_treat_application_exit_as_a_crash(self) -> None:
|
|
pattern = re.compile("AddressSanitizer: heap-use-after-free")
|
|
self.assertFalse(
|
|
fixed_runtime_reported(7, "normal application output", pattern)
|
|
)
|
|
self.assertTrue(
|
|
vulnerable_runtime_detected(
|
|
1, "AddressSanitizer: heap-use-after-free", pattern
|
|
)
|
|
)
|
|
|
|
def test_metrics_and_defined_status(self) -> None:
|
|
result = classification_metrics(3, 1, 2, 4)
|
|
self.assertEqual(result["precision"], 0.75)
|
|
self.assertEqual(result["recall"], 0.6)
|
|
self.assertAlmostEqual(result["f1"], 2.0 / 3.0)
|
|
self.assertEqual(result["specificity"], 0.8)
|
|
self.assertEqual(result["false_positive_rate"], 0.2)
|
|
self.assertEqual(result["accuracy"], 0.7)
|
|
self.assertEqual(result["balanced_accuracy"], 0.7)
|
|
self.assertAlmostEqual(
|
|
result["matthews_correlation_coefficient"], 10 / math.sqrt(600)
|
|
)
|
|
self.assertTrue(all(result["defined"].values()))
|
|
|
|
def test_does_not_invent_true_negatives(self) -> None:
|
|
result = classification_metrics(0, 0, 0)
|
|
self.assertNotIn("true_negative", result)
|
|
self.assertNotIn("specificity", result)
|
|
self.assertFalse(result["defined"]["precision"])
|
|
self.assertFalse(result["defined"]["recall"])
|
|
|
|
def test_all_negative_corpus_marks_positive_metrics_undefined(self) -> None:
|
|
result = classification_metrics(0, 0, 0, 3)
|
|
self.assertEqual(result["specificity"], 1.0)
|
|
self.assertEqual(result["accuracy"], 1.0)
|
|
self.assertFalse(result["defined"]["recall"])
|
|
self.assertFalse(result["defined"]["balanced_accuracy"])
|
|
self.assertFalse(result["defined"]["matthews_correlation_coefficient"])
|
|
|
|
def test_zero_f1_is_defined_when_cases_or_predictions_exist(self) -> None:
|
|
result = classification_metrics(0, 1, 1)
|
|
self.assertEqual(result["f1"], 0.0)
|
|
self.assertTrue(result["defined"]["f1"])
|
|
|
|
def test_rejects_invalid_counts(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
classification_metrics(-1, 0, 0)
|
|
with self.assertRaises(ValueError):
|
|
classification_metrics(True, 0, 0)
|
|
|
|
def test_bootstrap_is_seeded_and_bounded(self) -> None:
|
|
left = bootstrap_interval([1, 2, 3, 4, 5], resamples=200, seed=17)
|
|
right = bootstrap_interval([1, 2, 3, 4, 5], resamples=200, seed=17)
|
|
self.assertEqual(left, right)
|
|
self.assertEqual(left["estimate"], 3.0)
|
|
self.assertLessEqual(left["low"], left["estimate"])
|
|
self.assertGreaterEqual(left["high"], left["estimate"])
|
|
|
|
def test_micro_and_macro_metrics_preserve_weak_slices(self) -> None:
|
|
result = aggregate_classification_metrics(
|
|
{
|
|
"large-easy": {
|
|
"true_positive": 99,
|
|
"false_positive": 1,
|
|
"false_negative": 1,
|
|
},
|
|
"small-hard": {
|
|
"true_positive": 0,
|
|
"false_positive": 1,
|
|
"false_negative": 1,
|
|
},
|
|
}
|
|
)
|
|
self.assertAlmostEqual(result["micro"]["precision"], 99 / 101)
|
|
self.assertAlmostEqual(result["macro"]["precision"], 0.495)
|
|
self.assertEqual(result["macro_support"]["specificity"], 0)
|
|
|
|
def test_paired_comparison_reports_effect_and_exact_sign_test(self) -> None:
|
|
result = paired_comparison([10, 11, 12, 13, 14], [8, 9, 10, 11, 12])
|
|
self.assertEqual(result["wins"], 5)
|
|
self.assertEqual(result["losses"], 0)
|
|
self.assertEqual(result["median_delta"], -2.0)
|
|
self.assertEqual(result["common_language_effect"], 1.0)
|
|
self.assertEqual(result["two_sided_sign_test_p"], 0.0625)
|
|
|
|
def test_paired_comparison_rejects_unpaired_samples(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
paired_comparison([1, 2], [1])
|
|
|
|
|
|
class FindingScoringTest(unittest.TestCase):
|
|
def test_nested_paths_do_not_collapse_to_basename(self) -> None:
|
|
self.assertTrue(path_matches("src/a/item.cpp", "/repo/src/a/item.cpp"))
|
|
self.assertFalse(path_matches("src/a/item.cpp", "/repo/src/b/item.cpp"))
|
|
self.assertTrue(path_matches("item.cpp", "/repo/src/b/item.cpp"))
|
|
|
|
def test_kind_prefix_requires_a_family_separator(self) -> None:
|
|
gold = [{"kind": "dead", "files": ["a.cpp"]}]
|
|
accepted = score_grouped_findings(
|
|
[{"id": "1", "kind": "dead-branch", "file": "a.cpp"}], gold
|
|
)
|
|
rejected = score_grouped_findings(
|
|
[{"id": "1", "kind": "deadly", "file": "a.cpp"}], gold
|
|
)
|
|
self.assertEqual(accepted["true_positive"], 1)
|
|
self.assertEqual(rejected["true_positive"], 0)
|
|
|
|
def test_one_group_cannot_claim_two_gold_cases(self) -> None:
|
|
findings = [{"id": "same", "kind": "dead", "file": "src/a.cpp"}]
|
|
gold = [
|
|
{"kind": "dead", "files": ["src/a.cpp"]},
|
|
{"kind": "dead", "files": ["src/a.cpp"]},
|
|
]
|
|
result = score_grouped_findings(findings, gold)
|
|
self.assertEqual(result["true_positive"], 1)
|
|
self.assertEqual(result["false_negative"], 1)
|
|
self.assertEqual(result["false_positive"], 0)
|
|
|
|
def test_multi_location_group_matches_one_cluster(self) -> None:
|
|
findings = [
|
|
{"id": "clone-1", "kind": "clone", "file": "/repo/src/a.cpp"},
|
|
{"id": "clone-1", "kind": "clone", "file": "/repo/src/b.cpp"},
|
|
]
|
|
gold = [{"kind": "clone", "files": ["src/a.cpp", "src/b.cpp"]}]
|
|
result = score_grouped_findings(findings, gold)
|
|
self.assertEqual(result["true_positive"], 1)
|
|
self.assertEqual(result["finding_groups"], 1)
|
|
self.assertEqual(result["precision"], 1.0)
|
|
self.assertEqual(result["recall"], 1.0)
|
|
|
|
def test_explicit_negative_cases_produce_fp_and_tn(self) -> None:
|
|
findings = [
|
|
{"id": "tp", "kind": "null-deref", "file": "bug.cpp"},
|
|
{"id": "fp", "kind": "null-deref", "file": "bad-clean.cpp"},
|
|
{"id": "fp-duplicate", "kind": "null-deref", "file": "bad-clean.cpp"},
|
|
{"id": "outside", "kind": "other", "file": "outside.cpp"},
|
|
]
|
|
result = score_labeled_cases(
|
|
findings,
|
|
[
|
|
{"kind": "null-deref", "files": ["bug.cpp"]},
|
|
{"kind": "null-deref", "files": ["missed.cpp"]},
|
|
],
|
|
[
|
|
{"kind": "null-deref", "files": ["bad-clean.cpp"]},
|
|
{"kind": "null-deref", "files": ["good-clean.cpp"]},
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
(
|
|
result["true_positive"],
|
|
result["false_positive"],
|
|
result["false_negative"],
|
|
result["true_negative"],
|
|
),
|
|
(1, 1, 1, 1),
|
|
)
|
|
self.assertEqual(result["unscored_finding_ids"], ["outside"])
|
|
self.assertEqual(result["false_positive_cases"][0]["files"], ["bad-clean.cpp"])
|
|
self.assertEqual(
|
|
result["false_positive_cases"][0]["finding_ids"], ["fp", "fp-duplicate"]
|
|
)
|
|
|
|
def test_universe_expands_only_unlabeled_file_rule_pairs(self) -> None:
|
|
positives = [{"kind": "null-deref", "files": ["bug.cpp"]}]
|
|
negatives = negative_cases_from_universe(
|
|
positives,
|
|
{
|
|
"files": ["bug.cpp", "clean.cpp"],
|
|
"kinds": ["null-deref", "oob-access"],
|
|
},
|
|
)
|
|
self.assertEqual(len(negatives), 3)
|
|
self.assertNotIn(
|
|
("bug.cpp", "null-deref"),
|
|
{(entry["files"][0], entry["kind"]) for entry in negatives},
|
|
)
|
|
|
|
|
|
class SarifAndStabilityTest(unittest.TestCase):
|
|
def test_file_uri_normalization(self) -> None:
|
|
self.assertEqual(
|
|
normalize_uri(f"file://{ROOT / 'src/a.cpp'}", ROOT, ROOT), "src/a.cpp"
|
|
)
|
|
|
|
def test_uri_normalization_resolves_project_root_symlinks(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
real_root = Path(directory) / "real"
|
|
project = real_root / "project"
|
|
source = project / "src" / "a.cpp"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("int main() {}\n", encoding="utf-8")
|
|
linked_root = Path(directory) / "linked"
|
|
linked_root.symlink_to(real_root, target_is_directory=True)
|
|
self.assertEqual(
|
|
normalize_uri(
|
|
source.as_posix(), linked_root / "project", linked_root / "project"
|
|
),
|
|
"src/a.cpp",
|
|
)
|
|
|
|
def test_sarif_summary_flattens_all_runs(self) -> None:
|
|
report = {
|
|
"runs": [
|
|
{"results": [{"ruleId": "first"}]},
|
|
{"results": [{"ruleId": "second"}]},
|
|
]
|
|
}
|
|
summary = summarize_sarif_findings(report, ROOT, ROOT)
|
|
self.assertEqual(summary["count"], 2)
|
|
self.assertEqual(summary["by_rule"], {"first": 1, "second": 1})
|
|
|
|
def test_compare_findings_cli_flattens_all_runs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
report = root / "report.sarif.json"
|
|
spec = root / "spec.json"
|
|
first = {
|
|
"ruleId": "ok",
|
|
"properties": {"backend": "source"},
|
|
"locations": [
|
|
{
|
|
"physicalLocation": {
|
|
"artifactLocation": {"uri": "src/a.cpp"},
|
|
"region": {"startLine": 1},
|
|
}
|
|
}
|
|
],
|
|
}
|
|
second = {
|
|
"ruleId": "forbidden",
|
|
"properties": {"backend": "source"},
|
|
"locations": [
|
|
{
|
|
"physicalLocation": {
|
|
"artifactLocation": {"uri": "src/b.cpp"},
|
|
"region": {"startLine": 2},
|
|
}
|
|
}
|
|
],
|
|
}
|
|
report.write_text(
|
|
json.dumps({"runs": [{"results": [first]}, {"results": [second]}]}),
|
|
encoding="utf-8",
|
|
)
|
|
spec.write_text(
|
|
json.dumps(
|
|
{
|
|
"expect": {
|
|
"coverage_requirements": {"forbidden": ["negative"]},
|
|
"must_not_report": [
|
|
{"ruleId": "forbidden", "coverage": "negative"}
|
|
],
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "scripts" / "benchmarks" / "compare_findings.py"),
|
|
"--report",
|
|
str(report),
|
|
"--spec",
|
|
str(spec),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(proc.returncode, 1)
|
|
summary = json.loads(proc.stdout)
|
|
self.assertEqual(summary["total"], 2)
|
|
self.assertEqual(len(summary["failures"]), 1)
|
|
|
|
def test_sarif_summary_is_canonical_and_reports_missing_evidence(self) -> None:
|
|
report = {
|
|
"runs": [
|
|
{
|
|
"results": [
|
|
{
|
|
"ruleId": "dead",
|
|
"message": {"text": "unreachable"},
|
|
"properties": {"backend": "source", "confidence": "proved"},
|
|
"locations": [
|
|
{
|
|
"physicalLocation": {
|
|
"artifactLocation": {"uri": "src/a.cpp"},
|
|
"region": {"startLine": 7},
|
|
},
|
|
"logicalLocations": [
|
|
{"fullyQualifiedName": "project::f"}
|
|
],
|
|
}
|
|
],
|
|
},
|
|
{"ruleId": "locationless", "message": {"text": "crash"}},
|
|
]
|
|
}
|
|
]
|
|
}
|
|
summary = summarize_sarif_findings(report, ROOT, ROOT)
|
|
self.assertEqual(summary["count"], 2)
|
|
self.assertEqual(summary["by_rule"], {"dead": 1, "locationless": 1})
|
|
self.assertEqual(summary["missing_location"], 1)
|
|
self.assertEqual(summary["missing_backend"], 1)
|
|
self.assertEqual(len(summary["identity_keys"]), 2)
|
|
self.assertEqual(len(summary["fingerprint"]), 64)
|
|
proved = next(
|
|
item for item in summary["identity_keys"] if '"rule_id":"dead"' in item
|
|
)
|
|
self.assertNotIn('"summary_backed"', proved)
|
|
self.assertNotIn('"cross_tu_corroboration"', proved)
|
|
|
|
def test_sarif_identity_ignores_witness_message(self) -> None:
|
|
def report(message: str) -> dict:
|
|
return {
|
|
"runs": [
|
|
{
|
|
"results": [
|
|
{
|
|
"ruleId": "fold.int-overflow",
|
|
"message": {"text": message},
|
|
"properties": {
|
|
"id": "overflow:f:1",
|
|
"backend": "source",
|
|
},
|
|
"locations": [
|
|
{
|
|
"physicalLocation": {
|
|
"artifactLocation": {"uri": "src/a.cpp"},
|
|
"region": {"startLine": 7},
|
|
}
|
|
}
|
|
],
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
|
|
first = summarize_sarif_findings(report("when x=1"), ROOT, ROOT)
|
|
second = summarize_sarif_findings(report("when x=2"), ROOT, ROOT)
|
|
self.assertEqual(first["identity_keys"], second["identity_keys"])
|
|
self.assertEqual(first["fingerprint"], second["fingerprint"])
|
|
|
|
def test_sarif_retains_stable_finding_ids_for_review(self) -> None:
|
|
report = {
|
|
"runs": [
|
|
{
|
|
"results": [
|
|
{
|
|
"ruleId": "fold.dead-branch",
|
|
"properties": {"id": "dead:f:1", "backend": "source"},
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
summary = summarize_sarif_findings(report, ROOT, ROOT)
|
|
self.assertEqual(summary["records"][0]["finding_id"], "dead:f:1")
|
|
|
|
def test_partial_drift_is_visible(self) -> None:
|
|
result = stability_summary([{"a", "b"}, {"b", "c"}])
|
|
self.assertFalse(result["exact"])
|
|
self.assertEqual(result["intersection_count"], 1)
|
|
self.assertEqual(result["union_count"], 3)
|
|
self.assertEqual(result["churn_count"], 2)
|
|
self.assertAlmostEqual(result["pairwise_jaccard"]["median"], 0.333)
|
|
|
|
|
|
class DispositionDeltaTest(unittest.TestCase):
|
|
def baseline(self) -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"target_name": "example",
|
|
"reviewed_at": "2026-07-13T12:00:00Z",
|
|
"reviewer": "security-review",
|
|
"target": {
|
|
"revision": "abc",
|
|
"tu_set_fingerprint": "tu-set",
|
|
"configuration_fingerprint": configuration_fingerprint(
|
|
{"detectors": ["clone"]}
|
|
),
|
|
},
|
|
"findings": [
|
|
{
|
|
"id": "persisting",
|
|
"rule_id": "fold.clone",
|
|
"status": "accepted-risk",
|
|
"rationale": "Intentional schema adapter family.",
|
|
},
|
|
{
|
|
"id": "resolved",
|
|
"rule_id": "fold.high-complexity",
|
|
"status": "actionable",
|
|
"rationale": "Split this function.",
|
|
},
|
|
],
|
|
}
|
|
|
|
def test_reports_new_persisting_resolved_and_changed_file_density(self) -> None:
|
|
result = compare_dispositions(
|
|
[
|
|
{
|
|
"finding_id": "persisting",
|
|
"rule_id": "fold.clone",
|
|
"uri": "src/a.cpp",
|
|
},
|
|
{
|
|
"finding_id": "new",
|
|
"rule_id": "fold.null-deref",
|
|
"uri": "src/b.cpp",
|
|
},
|
|
],
|
|
self.baseline(),
|
|
changed_files=["src/b.cpp"],
|
|
scope_kloc=2.0,
|
|
)
|
|
self.assertEqual(result["counts"]["new"], 1)
|
|
self.assertEqual(result["counts"]["persisting"], 1)
|
|
self.assertEqual(result["counts"]["resolved"], 1)
|
|
self.assertEqual(result["counts"]["new_in_changed_files"], 1)
|
|
self.assertEqual(result["resolved_by_disposition"], {"actionable": 1})
|
|
self.assertEqual(result["new_by_rule"], {"fold.null-deref": 1})
|
|
self.assertEqual(result["new_alert_density_per_kloc"], 0.5)
|
|
|
|
def test_rejects_unreviewed_or_duplicate_ids(self) -> None:
|
|
baseline = self.baseline()
|
|
baseline["findings"][1]["id"] = "persisting"
|
|
self.assertTrue(validate_disposition_baseline(baseline))
|
|
with self.assertRaises(ValueError):
|
|
compare_dispositions(
|
|
[{"finding_id": "same"}, {"finding_id": "same"}], self.baseline()
|
|
)
|
|
|
|
def test_stable_id_cannot_silently_change_rule_identity(self) -> None:
|
|
result = compare_dispositions(
|
|
[
|
|
{
|
|
"finding_id": "persisting",
|
|
"rule_id": "fold.dead-branch",
|
|
"uri": "src/a.cpp",
|
|
}
|
|
],
|
|
self.baseline(),
|
|
)
|
|
self.assertEqual(len(result["identity_mismatches"]), 1)
|
|
|
|
|
|
class ScanOperationalEvidenceTest(unittest.TestCase):
|
|
def test_classifies_dependency_timeout_crash_and_frontend_failures(self):
|
|
evidence = run_project_eval.scan_operational_evidence(
|
|
"fatal error: generated.pb.h file not found\n"
|
|
"worker pid 1 killed by signal 14 while processing TU: slow.cpp\n"
|
|
"worker pid 2 killed by signal 11 while processing TU: crash.cpp\n"
|
|
"worker pid 3 exited with status 2 while processing TU: broken.cpp\n"
|
|
)
|
|
self.assertEqual(evidence["missing_dependencies"]["count"], 1)
|
|
self.assertEqual(evidence["tu_timeouts"]["count"], 1)
|
|
self.assertEqual(evidence["tu_crashes"]["count"], 1)
|
|
self.assertEqual(evidence["frontend_failures"]["count"], 1)
|
|
|
|
def test_solver_timeout_is_not_mislabeled_as_tu_timeout(self):
|
|
evidence = run_project_eval.scan_operational_evidence(
|
|
"[dead_branch] solver unknown: timeout\n"
|
|
)
|
|
self.assertTrue(all(row["count"] == 0 for row in evidence.values()))
|
|
|
|
|
|
class PerformanceAttributionTest(unittest.TestCase):
|
|
def test_separates_named_source_phases_from_unattributed_time(self) -> None:
|
|
def metric(value: int) -> dict[str, int]:
|
|
return {"median": value, "max": value}
|
|
|
|
profile = {
|
|
"scan": {
|
|
"analyze_ms": metric(1000),
|
|
"function_fingerprint_ms": metric(100),
|
|
"call_graph_ms": metric(20),
|
|
"project_index_load_ms": metric(3),
|
|
"project_graph_collect_ms": metric(7),
|
|
"project_graph_merge_ms": metric(5),
|
|
"project_graph_fallback_ms": metric(0),
|
|
"project_index_contribution_update_ms": metric(2),
|
|
"project_index_snapshot_update_ms": metric(4),
|
|
"security_candidate_ms": metric(1),
|
|
"cross_tu_property_ms": metric(80),
|
|
"symbolic_summary_ms": metric(50),
|
|
"peak_rss_kb": metric(2048),
|
|
},
|
|
"detectors": {
|
|
"dead": {
|
|
"wall_ms": metric(300),
|
|
"z3_queries": metric(2),
|
|
"z3_timeouts": metric(0),
|
|
}
|
|
},
|
|
}
|
|
|
|
result = optimization_surface(profile)
|
|
self.assertEqual(result["shared_frontend_and_orchestration_ms"], 700.0)
|
|
self.assertEqual(result["known_source_phase_ms"], 251.0)
|
|
self.assertEqual(result["unattributed_frontend_and_orchestration_ms"], 449.0)
|
|
self.assertEqual(result["peak_rss_mb"], 2.0)
|
|
self.assertEqual(
|
|
result["ranked_source_phases"][0]["phase"], "function_fingerprint"
|
|
)
|
|
self.assertEqual(
|
|
result["ranked_graph_subphases"][0],
|
|
{"phase": "project_graph_collect", "wall_ms": 7.0},
|
|
)
|
|
|
|
def test_performance_comparison_pairs_matching_repetitions(self) -> None:
|
|
baseline = {
|
|
"performance": {
|
|
"profiled_serial": {
|
|
"wall_samples_seconds": [10.0, 12.0, 11.0],
|
|
"wall_seconds": {"median": 11.0},
|
|
}
|
|
}
|
|
}
|
|
current = {
|
|
"performance": {
|
|
"profiled_serial": {
|
|
"wall_samples_seconds": [9.0, 10.0, 10.0],
|
|
"wall_seconds": {"median": 10.0},
|
|
}
|
|
}
|
|
}
|
|
comparison, failures = compare_performance_lanes(
|
|
current, baseline, max_wall_percent=5.0
|
|
)
|
|
self.assertEqual(
|
|
comparison["profiled_serial"]["method"], "paired-by-repetition"
|
|
)
|
|
self.assertEqual(comparison["profiled_serial"]["paired"]["wins"], 3)
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_old_performance_baseline_is_labelled_median_only(self) -> None:
|
|
baseline = {
|
|
"performance": {"profiled_serial": {"wall_seconds": {"median": 10.0}}}
|
|
}
|
|
current = {
|
|
"performance": {"profiled_serial": {"wall_seconds": {"median": 12.5}}}
|
|
}
|
|
comparison, failures = compare_performance_lanes(
|
|
current, baseline, max_wall_percent=20.0
|
|
)
|
|
self.assertEqual(comparison["profiled_serial"]["method"], "median-only")
|
|
self.assertEqual(len(failures), 1)
|
|
|
|
def test_gold_threshold_gate_reports_metric_regressions(self) -> None:
|
|
failures = run_project_eval.gold_threshold_failures(
|
|
{"precision": 0.8, "recall": 0.9, "f1": 0.84},
|
|
{"precision": 0.85, "f1": 0.85},
|
|
)
|
|
self.assertEqual(
|
|
failures,
|
|
[
|
|
"gold precision 0.800 below threshold 0.850",
|
|
"gold f1 0.840 below threshold 0.850",
|
|
],
|
|
)
|
|
|
|
def test_per_rule_gold_threshold_gate_reports_regressions(self) -> None:
|
|
failures = run_project_eval.per_rule_gold_threshold_failures(
|
|
{"div-by-zero": {"precision": 0.8, "recall": 1.0, "f1": 0.88}},
|
|
{"div-by-zero": {"precision": 0.85, "f1": 0.9}},
|
|
)
|
|
self.assertEqual(
|
|
failures,
|
|
[
|
|
"rule div-by-zero precision 0.800 below threshold 0.850",
|
|
"rule div-by-zero f1 0.880 below threshold 0.900",
|
|
],
|
|
)
|
|
|
|
def test_portable_time_parser_does_not_invent_rss(self) -> None:
|
|
cpu, rss = runner_common._parse_posix_time(
|
|
"program output\nreal 0.12\nuser 0.03\nsys 0.02\n"
|
|
)
|
|
self.assertAlmostEqual(cpu, 0.05)
|
|
self.assertIsNone(rss)
|
|
|
|
def test_token_efficiency_is_reproducible_and_uses_report_bytes(self) -> None:
|
|
self.assertEqual(estimate_text_tokens("abcde"), 2)
|
|
result = artifact_token_efficiency(
|
|
[
|
|
{"output_bytes": 4, "report_bytes": 12, "finding_count": 3},
|
|
{"output_bytes": 8, "report_bytes": 20, "finding_count": 4},
|
|
]
|
|
)
|
|
self.assertEqual(result["estimator"], "ceil(utf8_bytes/4)")
|
|
self.assertEqual(result["driver_output"]["estimated_tokens"]["median"], 1.5)
|
|
self.assertEqual(result["sarif_report"]["estimated_tokens"]["median"], 4.0)
|
|
self.assertAlmostEqual(
|
|
result["sarif_report"]["findings_per_1k_estimated_tokens"]["median"],
|
|
900.0,
|
|
)
|
|
|
|
def test_project_lane_does_not_reuse_stale_report_or_profile(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output_dir = Path(tmp)
|
|
(output_dir / "source-1.sarif.json").write_text(
|
|
'{"version":"2.1.0","runs":[]}', encoding="utf-8"
|
|
)
|
|
(output_dir / "source-1.profile.json").write_text(
|
|
'{"stale":true}', encoding="utf-8"
|
|
)
|
|
measured = runner_common.MeasuredProcess(0, "no artifacts", 0.01)
|
|
with mock.patch.object(
|
|
run_project_eval.common, "run_measured", return_value=measured
|
|
):
|
|
runs, failures = run_project_eval.run_lane(
|
|
name="source",
|
|
fold=Path(sys.executable),
|
|
output_dir=output_dir,
|
|
repeats=1,
|
|
command_for=lambda report, profile, index: ["fake-fold"],
|
|
uri_base=ROOT,
|
|
project_root=ROOT,
|
|
timeout=1.0,
|
|
experiment_prefix="stale-artifact-test",
|
|
target={},
|
|
budgets={},
|
|
profiled=True,
|
|
)
|
|
|
|
self.assertNotIn("findings", runs[0])
|
|
self.assertTrue(
|
|
any("produced invalid artifacts" in failure for failure in failures)
|
|
)
|
|
self.assertFalse((output_dir / "source-1.sarif.json").exists())
|
|
self.assertFalse((output_dir / "source-1.profile.json").exists())
|
|
|
|
def test_project_lane_fails_on_missing_generated_dependency(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output_dir = Path(tmp)
|
|
|
|
def command_for(report: Path, profile: Path, _index: int) -> list[str]:
|
|
report.write_text(
|
|
'{"version":"2.1.0","runs":[{"results":[]}]}',
|
|
encoding="utf-8",
|
|
)
|
|
profile.write_text('{"scan":{},"detectors":[]}', encoding="utf-8")
|
|
return ["fake-fold"]
|
|
|
|
measured = runner_common.MeasuredProcess(
|
|
0,
|
|
"proto_serializer.cpp: fatal error: generated.pb.h file not found",
|
|
0.01,
|
|
)
|
|
with mock.patch.object(
|
|
run_project_eval.common, "run_measured", return_value=measured
|
|
):
|
|
runs, failures = run_project_eval.run_lane(
|
|
name="source",
|
|
fold=Path(sys.executable),
|
|
output_dir=output_dir,
|
|
repeats=1,
|
|
command_for=command_for,
|
|
uri_base=ROOT,
|
|
project_root=ROOT,
|
|
timeout=1.0,
|
|
experiment_prefix="missing-generated-header-test",
|
|
target={"selected_tus": 1},
|
|
budgets={},
|
|
profiled=True,
|
|
)
|
|
|
|
self.assertEqual(
|
|
runs[0]["operational_evidence"]["missing_dependencies"]["count"], 1
|
|
)
|
|
self.assertEqual(
|
|
runs[0]["coverage_reconciliation"],
|
|
{
|
|
"selected_tus": 1,
|
|
"failed_tus": 1,
|
|
"completed_tus_lower_bound": 0,
|
|
"affected_tus": ["proto_serializer.cpp"],
|
|
},
|
|
)
|
|
self.assertTrue(any("missing dependencies: 1" in row for row in failures))
|
|
|
|
def test_project_lane_fails_on_profile_coverage_loss(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output_dir = Path(tmp)
|
|
|
|
def command_for(report: Path, profile: Path, _index: int) -> list[str]:
|
|
report.write_text(
|
|
'{"version":"2.1.0","runs":[{"results":[]}]}',
|
|
encoding="utf-8",
|
|
)
|
|
profile.write_text(
|
|
json.dumps(
|
|
{
|
|
"scan": {
|
|
"source_tus_selected": 2,
|
|
"source_tus_attempted": 2,
|
|
"source_tus_completed": 1,
|
|
"source_tus_timed_out": 0,
|
|
"source_tus_frontend_failed": 1,
|
|
},
|
|
"detectors": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return ["fake-fold"]
|
|
|
|
measured = runner_common.MeasuredProcess(0, "", 0.01)
|
|
with mock.patch.object(
|
|
run_project_eval.common, "run_measured", return_value=measured
|
|
):
|
|
runs, failures = run_project_eval.run_lane(
|
|
name="source",
|
|
fold=Path(sys.executable),
|
|
output_dir=output_dir,
|
|
repeats=1,
|
|
command_for=command_for,
|
|
uri_base=ROOT,
|
|
project_root=ROOT,
|
|
timeout=1.0,
|
|
experiment_prefix="profile-coverage-test",
|
|
target={"selected_tus": 2},
|
|
budgets={},
|
|
profiled=True,
|
|
)
|
|
|
|
self.assertEqual(runs[0]["coverage_reconciliation"]["completed_tus"], 1)
|
|
self.assertTrue(
|
|
any("incomplete profile coverage" in failure for failure in failures)
|
|
)
|
|
|
|
def test_missing_executable_is_recorded_as_a_launch_failure(self) -> None:
|
|
result = runner_common.run_measured(
|
|
["/definitely/missing/fold"], ROOT, timeout_seconds=1.0
|
|
)
|
|
self.assertEqual(result.returncode, 127)
|
|
self.assertFalse(result.timed_out)
|
|
self.assertIn("failed to launch", result.output)
|
|
|
|
|
|
class PairedAbHarnessTest(unittest.TestCase):
|
|
def test_matrix_distinguishes_persistent_and_process_warmth(self):
|
|
cells = run_ab_evaluation.matrix_cells(4)
|
|
self.assertEqual(
|
|
[cell.key for cell in cells],
|
|
["serial-cold", "serial-warm", "fanout-cold", "fanout-warm"],
|
|
)
|
|
self.assertTrue(cells[0].incremental)
|
|
self.assertTrue(cells[1].incremental)
|
|
self.assertFalse(cells[2].incremental)
|
|
self.assertEqual(cells[3].cache_mechanism, "excluded-preflight-process")
|
|
with self.assertRaises(ValueError):
|
|
run_ab_evaluation.matrix_cells(1)
|
|
|
|
def test_paired_summary_preserves_raw_pairs_and_finding_parity(self):
|
|
baseline = [
|
|
{
|
|
"pair_repetition": 1,
|
|
"wall_seconds": 2.0,
|
|
"findings": {"fingerprint": "same", "count": 3},
|
|
},
|
|
{
|
|
"pair_repetition": 2,
|
|
"wall_seconds": 4.0,
|
|
"findings": {"fingerprint": "same", "count": 3},
|
|
},
|
|
]
|
|
current = [
|
|
{
|
|
"pair_repetition": 1,
|
|
"wall_seconds": 1.0,
|
|
"findings": {"fingerprint": "same", "count": 3},
|
|
},
|
|
{
|
|
"pair_repetition": 2,
|
|
"wall_seconds": 3.0,
|
|
"findings": {"fingerprint": "same", "count": 3},
|
|
},
|
|
]
|
|
summary = run_ab_evaluation.paired_cell_summary(baseline, current)
|
|
self.assertTrue(summary["finding_equivalent"])
|
|
self.assertEqual(summary["paired_wall_delta_seconds"], [-1.0, -1.0])
|
|
self.assertEqual(summary["baseline_wall_median_seconds"], 3.0)
|
|
self.assertEqual(summary["current_wall_median_seconds"], 2.0)
|
|
|
|
current[1]["findings"]["fingerprint"] = "changed"
|
|
self.assertFalse(
|
|
run_ab_evaluation.paired_cell_summary(baseline, current)[
|
|
"finding_equivalent"
|
|
]
|
|
)
|
|
|
|
def test_output_directory_is_immutable(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
root = Path(raw)
|
|
output = root / "new"
|
|
run_ab_evaluation.ensure_empty_output(output)
|
|
(output / "artifact").write_text("retained", encoding="utf-8")
|
|
with self.assertRaises(ValueError):
|
|
run_ab_evaluation.ensure_empty_output(output)
|
|
|
|
|
|
class BenchmarkSchemaTest(unittest.TestCase):
|
|
def test_all_checked_in_manifests_use_canonical_schema(self) -> None:
|
|
# Runtime evaluation output (tests/benchmarks/runs/) and the derived
|
|
# performance baselines (tests/benchmarks/projects/baselines/) are
|
|
# not benchmark cases, manifests, or dispositions; the optimization-
|
|
# loop harness writes run artifacts and baseline summaries there.
|
|
# Skip them so the schema test covers exactly the checked-in
|
|
# documents it claims to cover.
|
|
skip_dirs = {"runs", "baselines"}
|
|
manifests = sorted(
|
|
path
|
|
for path in (ROOT / "tests" / "benchmarks").glob("**/*.json")
|
|
if not (skip_dirs & set(path.relative_to(ROOT).parts))
|
|
)
|
|
self.assertTrue(manifests)
|
|
for path in manifests:
|
|
with self.subTest(path=path):
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
if "target_name" in document and "findings" in document:
|
|
self.assertEqual(validate_disposition_baseline(document), [])
|
|
continue
|
|
if "lane" in document and "project" in document:
|
|
self.assertEqual(validate_project_manifest(document), [])
|
|
continue
|
|
result = validate_benchmark_case(
|
|
document, allow_legacy=False, source_path=str(path)
|
|
)
|
|
self.assertEqual(result["errors"], [])
|
|
|
|
def test_legacy_case_is_normalized_with_warning(self) -> None:
|
|
result = validate_benchmark_case(
|
|
{"name": "guarded division", "build_source": "guard.c", "expect": {}}
|
|
)
|
|
self.assertTrue(result["valid"])
|
|
self.assertTrue(result["warnings"])
|
|
self.assertEqual(result["case"]["case_id"], "guarded-division")
|
|
self.assertEqual(result["case"]["backend"], "binary")
|
|
|
|
def test_dynamic_case_requires_replayable_runtime_oracle(self) -> None:
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"case_id": "div-runtime",
|
|
"case_type": "dynamic-confirmation",
|
|
"origin": "synthetic",
|
|
"backend": "source",
|
|
"oracle": {},
|
|
"runtime_oracle": {
|
|
"hazard": "div-by-zero",
|
|
"input_channel": "harness",
|
|
"checker": "ubsan",
|
|
"replay_command": "./harness crash-input",
|
|
},
|
|
}
|
|
self.assertTrue(validate_benchmark_case(manifest)["valid"])
|
|
del manifest["runtime_oracle"]["replay_command"]
|
|
self.assertFalse(validate_benchmark_case(manifest)["valid"])
|
|
|
|
def test_differential_case_names_both_oracles(self) -> None:
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"case_id": "sast-ubsan",
|
|
"case_type": "differential",
|
|
"origin": "synthetic",
|
|
"backend": "source",
|
|
"oracle": {},
|
|
"differential_oracle": {
|
|
"agreement": "same vulnerable/fixed classification",
|
|
"comparators": ["fold-source", "ubsan"],
|
|
},
|
|
}
|
|
self.assertTrue(validate_benchmark_case(manifest)["valid"])
|
|
manifest["differential_oracle"]["comparators"] = ["fold-source"]
|
|
self.assertFalse(validate_benchmark_case(manifest)["valid"])
|
|
|
|
def test_organic_pairs_require_pinned_provenance_and_local_hashes(self) -> None:
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"case_id": "organic-pair",
|
|
"case_type": "vulnerable-fixed",
|
|
"origin": "organic",
|
|
"backend": "source",
|
|
"expect": {},
|
|
"lineage": {
|
|
"vulnerable_revision": "per-pair",
|
|
"fixed_revision": "per-pair",
|
|
},
|
|
"corpus": {
|
|
"kind": "adapted-upstream-vulnerable-fixed",
|
|
"entries": [
|
|
{
|
|
"name": "example-cve",
|
|
"cve": "CVE-2000-0001",
|
|
"project": "example",
|
|
"repository": "https://example.test/repo",
|
|
"reference": "https://example.test/fix",
|
|
"vulnerable_revision": "a" * 40,
|
|
"fixed_revision": "b" * 40,
|
|
"upstream_path": "src/example.c",
|
|
"license": "MIT",
|
|
"adaptation": "Dependency-minimized semantic slice.",
|
|
"vulnerable_source": "vulnerable.cpp",
|
|
"vulnerable_sha256": "a" * 64,
|
|
"fixed_source": "fixed.cpp",
|
|
"fixed_sha256": "b" * 64,
|
|
"expected": {
|
|
"vulnerable_detected": True,
|
|
"fixed_reported": False,
|
|
},
|
|
}
|
|
],
|
|
},
|
|
}
|
|
self.assertTrue(validate_benchmark_case(manifest)["valid"])
|
|
del manifest["corpus"]["entries"][0]["adaptation"]
|
|
self.assertFalse(validate_benchmark_case(manifest)["valid"])
|
|
|
|
def test_run_artifact_records_hashes_command_and_machine(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
report = Path(directory) / "report.json"
|
|
report.write_text("{}\n", encoding="utf-8")
|
|
document = build_run_artifact(
|
|
experiment_id="exp-1",
|
|
case_id="case-1",
|
|
tool={"name": "fold", "version": "0.1.0+abc", "revision": "abc"},
|
|
target={"revision": "def", "input_fingerprint": "123"},
|
|
command=["fold", "scan"],
|
|
started_at="2026-07-13T00:00:00+00:00",
|
|
ended_at="2026-07-13T00:00:01+00:00",
|
|
exit_status=0,
|
|
budgets={"wall_seconds": 10},
|
|
artifacts=[artifact_record("report", report, "application/sarif+json")],
|
|
)
|
|
self.assertEqual(validate_run_artifact(document), [])
|
|
self.assertEqual(document["tool"]["version"], "0.1.0+abc")
|
|
self.assertEqual(len(document["artifacts"][0]["sha256"]), 64)
|
|
self.assertTrue(document["execution"]["machine"]["system"])
|
|
|
|
|
|
class RustToolchainSkewTest(unittest.TestCase):
|
|
def test_release_parser_enforces_numeric_semver(self) -> None:
|
|
self.assertEqual(
|
|
rust_toolchain_skew.version_tuple("1.99.0-nightly"), (1, 99, 0)
|
|
)
|
|
self.assertIsNone(rust_toolchain_skew.version_tuple("nightly"))
|
|
self.assertIsNone(rust_toolchain_skew.version_tuple("1.99-nightly"))
|
|
|
|
def test_component_names_normalize_rustup_target_suffixes(self) -> None:
|
|
self.assertEqual(
|
|
rust_toolchain_skew.component_name("rustc-dev-aarch64-apple-darwin"),
|
|
"rustc-dev",
|
|
)
|
|
self.assertEqual(
|
|
rust_toolchain_skew.component_name("llvm-tools-x86_64-unknown-linux-gnu"),
|
|
"llvm-tools-preview",
|
|
)
|
|
|
|
def test_analyzer_exclusions_are_actionable(self) -> None:
|
|
old_nightly = rust_toolchain_skew.Toolchain(
|
|
"nightly-old", "1.97.0-nightly", (1, 97, 0), True, ()
|
|
)
|
|
stable = rust_toolchain_skew.Toolchain(
|
|
"stable", "1.99.0", (1, 99, 0), False, ()
|
|
)
|
|
self.assertIn("minimum", rust_toolchain_skew.analyzer_exclusion(old_nightly))
|
|
self.assertIn("rustc_private", rust_toolchain_skew.analyzer_exclusion(stable))
|
|
|
|
|
|
class MetamorphicAndConfirmationTest(unittest.TestCase):
|
|
def test_alpha_rename_and_line_shift_map_to_same_finding(self) -> None:
|
|
baseline = [
|
|
{
|
|
"rule_id": "fold.div-by-zero",
|
|
"backend": "source",
|
|
"confidence": "proved",
|
|
"uri": "before.cpp",
|
|
"line": 4,
|
|
"function": "divide",
|
|
}
|
|
]
|
|
transformed = [
|
|
{
|
|
"rule_id": "fold.div-by-zero",
|
|
"backend": "source",
|
|
"confidence": "proved",
|
|
"uri": "after.cpp",
|
|
"line": 6,
|
|
"function": "quotient",
|
|
}
|
|
]
|
|
relation = {
|
|
"uri_map": {"before.cpp": "after.cpp"},
|
|
"function_map": {"divide": "quotient"},
|
|
"line_delta": {"after.cpp": 2},
|
|
}
|
|
self.assertTrue(
|
|
compare_metamorphic_findings(baseline, transformed, relation)["exact"]
|
|
)
|
|
|
|
def test_metamorphic_comparison_preserves_duplicate_counts(self) -> None:
|
|
finding = {"rule_id": "fold.clone", "backend": "source"}
|
|
result = compare_metamorphic_findings(
|
|
[finding], [finding, finding], {"fields": ["rule_id", "backend"]}
|
|
)
|
|
self.assertFalse(result["exact"])
|
|
self.assertEqual(len(result["unexpected"]), 1)
|
|
|
|
def test_confirmation_progress_is_monotonic_and_summarized(self) -> None:
|
|
self.assertTrue(
|
|
validate_confirmation_progress(
|
|
{"reached": False, "triggered": True, "detected": True}
|
|
)
|
|
)
|
|
results = [
|
|
{
|
|
"finding_id": "f1",
|
|
"status": "confirmed",
|
|
"progress": {
|
|
"reached": True,
|
|
"triggered": True,
|
|
"detected": True,
|
|
"replayed": True,
|
|
"time_to_detected_ms": 12,
|
|
},
|
|
}
|
|
]
|
|
summary = summarize_confirmation_results(results)
|
|
self.assertEqual(summary["stage_counts"]["replayed"], 1)
|
|
self.assertEqual(summary["time_to_stage_ms"]["detected"]["median"], 12.0)
|
|
self.assertEqual(summary["validation_errors"], [])
|
|
|
|
|
|
class ProjectManifestTest(unittest.TestCase):
|
|
def valid_source(self) -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"project": "yams",
|
|
"lane": "source",
|
|
"corpus": {
|
|
"path": "../yams",
|
|
"compile_db": "../yams/build/release/compile_commands.json",
|
|
},
|
|
"analysis": {
|
|
"scopes": ["quality"],
|
|
"path_regex": "/src/metadata/",
|
|
"max_tus": 12,
|
|
"processes": 4,
|
|
},
|
|
"evaluation": {
|
|
"repeats": 3,
|
|
"dispositions": "tests/benchmarks/yams/metadata_dispositions.json",
|
|
"wall_regression_pct": 10.0,
|
|
"z3_timeout_budget": 0,
|
|
},
|
|
}
|
|
|
|
def test_valid_source_manifest(self) -> None:
|
|
self.assertEqual(validate_project_manifest(self.valid_source()), [])
|
|
|
|
def test_fold_policy_is_the_local_quality_default(self) -> None:
|
|
manifest_path = ROOT / "tests" / "benchmarks" / "projects" / "fold_source.json"
|
|
disposition_path = (
|
|
ROOT / "tests" / "benchmarks" / "fold" / "source_dispositions.json"
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
dispositions_document = json.loads(disposition_path.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(fold_quality.DEFAULT_MANIFEST, manifest_path)
|
|
self.assertEqual(validate_project_manifest(manifest), [])
|
|
self.assertEqual(manifest["project"], "fold")
|
|
self.assertEqual(manifest["corpus"]["missing_corpus"], "fail")
|
|
self.assertNotIn("scopes", manifest["analysis"])
|
|
self.assertEqual(
|
|
manifest["analysis"]["detectors"],
|
|
[
|
|
"clone",
|
|
"unused-function",
|
|
"uninit-read",
|
|
"hardcoded-credential",
|
|
"use-after-move",
|
|
"sql-injection",
|
|
],
|
|
)
|
|
self.assertEqual(manifest["analysis"]["max_tus"], 0)
|
|
self.assertEqual(manifest["evaluation"]["z3_timeout_budget"], 0)
|
|
self.assertEqual(manifest["evaluation"]["stability"]["max_churn"], 0)
|
|
self.assertEqual(
|
|
manifest["evaluation"]["stability"]["min_pairwise_jaccard"], 1.0
|
|
)
|
|
self.assertEqual(validate_disposition_baseline(dispositions_document), [])
|
|
self.assertEqual(dispositions_document["target_name"], "fold")
|
|
statuses = [finding["status"] for finding in dispositions_document["findings"]]
|
|
self.assertEqual(statuses.count("accepted-risk"), 7)
|
|
self.assertEqual(statuses.count("false-positive"), 2)
|
|
expected_configuration = {
|
|
"scopes": sorted(manifest["analysis"].get("scopes", [])),
|
|
"detectors": ",".join(sorted(manifest["analysis"]["detectors"])),
|
|
"scan_settings": run_project_eval.source_scan_settings(
|
|
manifest["analysis"]
|
|
),
|
|
}
|
|
self.assertEqual(
|
|
dispositions_document["target"]["configuration_fingerprint"],
|
|
configuration_fingerprint(expected_configuration),
|
|
)
|
|
new_finding = {
|
|
"finding_id": "new-unreviewed-id",
|
|
"rule_id": "fold.clone",
|
|
"uri": "src/report/sarif.cpp",
|
|
"line": 1,
|
|
}
|
|
delta = compare_dispositions([new_finding], dispositions_document)
|
|
self.assertEqual(delta["counts"]["new"], 1)
|
|
self.assertEqual(delta["new"], ["new-unreviewed-id"])
|
|
|
|
def test_committed_chromium_manifest_is_bounded_and_valid(self) -> None:
|
|
manifest_path = (
|
|
ROOT / "tests" / "benchmarks" / "projects" / "chromium_source.json"
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(validate_project_manifest(manifest), [])
|
|
self.assertEqual(manifest["project"], "chromium")
|
|
self.assertEqual(manifest["corpus"]["missing_corpus"], "skip")
|
|
self.assertEqual(manifest["analysis"]["max_tus"], 8)
|
|
self.assertEqual(manifest["evaluation"]["repeats"], 1)
|
|
self.assertIn("services/network", manifest["analysis"]["path_regex"])
|
|
|
|
def test_lane_specific_corpus_requirements(self) -> None:
|
|
rust = self.valid_source()
|
|
rust["lane"] = "rust"
|
|
rust["corpus"] = {"path": "../crypto/sage-protocol/packages/sage"}
|
|
self.assertEqual(validate_project_manifest(rust), [])
|
|
|
|
binary = self.valid_source()
|
|
binary["lane"] = "binary"
|
|
binary["corpus"] = {"binaries": ["../yams/build/release/yamsd"]}
|
|
self.assertEqual(validate_project_manifest(binary), [])
|
|
|
|
missing_db = self.valid_source()
|
|
del missing_db["corpus"]["compile_db"]
|
|
self.assertTrue(validate_project_manifest(missing_db))
|
|
|
|
empty_binaries = self.valid_source()
|
|
empty_binaries["lane"] = "binary"
|
|
empty_binaries["corpus"] = {"binaries": []}
|
|
self.assertTrue(validate_project_manifest(empty_binaries))
|
|
|
|
def test_rejects_bad_lane_version_and_policy(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["lane"] = "fortran"
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
manifest = self.valid_source()
|
|
manifest["schema_version"] = 99
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
manifest = self.valid_source()
|
|
manifest["corpus"]["missing_corpus"] = "explode"
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
def test_rejects_conflicting_detector_declarations(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["analysis"]["detectors"] = ["clone"]
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
def test_rejects_invalid_minimum_function_contract(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["evaluation"]["min_functions"] = -1
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
def test_rejects_invalid_gold_thresholds(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["evaluation"]["gold_thresholds"] = {"f1": 1.2}
|
|
self.assertTrue(validate_project_manifest(manifest))
|
|
|
|
def test_rejects_gold_thresholds_without_gold(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["evaluation"]["gold_thresholds"] = {"f1": 0.85}
|
|
self.assertIn(
|
|
"evaluation.gold_thresholds requires evaluation.gold",
|
|
validate_project_manifest(manifest),
|
|
)
|
|
|
|
def test_accepts_gold_thresholds_in_unit_interval(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["evaluation"]["gold"] = "gold.json"
|
|
manifest["evaluation"]["gold_thresholds"] = {
|
|
"precision": 0.9,
|
|
"recall": 0.8,
|
|
"f1": 0.85,
|
|
"specificity": 0.95,
|
|
"balanced_accuracy": 0.9,
|
|
"matthews_correlation_coefficient": 0.8,
|
|
}
|
|
self.assertEqual(validate_project_manifest(manifest), [])
|
|
|
|
def test_fingerprint_ignores_evaluation_policy(self) -> None:
|
|
first = self.valid_source()
|
|
second = self.valid_source()
|
|
second["evaluation"] = {"repeats": 9, "baseline": "other.json"}
|
|
self.assertEqual(manifest_fingerprint(first), manifest_fingerprint(second))
|
|
|
|
def test_fingerprint_tracks_analysis_configuration(self) -> None:
|
|
first = self.valid_source()
|
|
second = self.valid_source()
|
|
second["analysis"]["max_tus"] = 24
|
|
self.assertNotEqual(manifest_fingerprint(first), manifest_fingerprint(second))
|
|
config = manifest_configuration(first)
|
|
self.assertEqual(config["scopes"], ["quality"])
|
|
self.assertEqual(config["lane"], "source")
|
|
|
|
def test_runner_rejects_gold_thresholds_without_gold(self) -> None:
|
|
manifest = self.valid_source()
|
|
manifest["evaluation"]["gold_thresholds"] = {"f1": 0.85}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
manifest_path = Path(directory) / "manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "scripts" / "benchmarks" / "run_project_eval.py"),
|
|
"--manifest",
|
|
str(manifest_path),
|
|
"--fold",
|
|
sys.executable,
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(proc.returncode, 2)
|
|
self.assertIn(
|
|
"evaluation.gold_thresholds requires evaluation.gold",
|
|
proc.stderr,
|
|
)
|
|
|
|
|
|
class ArchitectureContractTest(unittest.TestCase):
|
|
def test_required_content_checks_are_rejected_as_brittle(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
with self.assertRaisesRegex(ValueError, "required_content"):
|
|
check_source_structure.validate_architecture(
|
|
root,
|
|
{
|
|
"architecture": {
|
|
"required_content": [
|
|
{"path": "src/example.cpp", "contains": "helper_name"}
|
|
]
|
|
}
|
|
},
|
|
)
|
|
|
|
def test_helper_renames_do_not_change_semantic_boundary_check(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source_dir = root / "src"
|
|
include_dir = root / "include" / "fold"
|
|
source_dir.mkdir()
|
|
include_dir.mkdir(parents=True)
|
|
source = source_dir / "example.cpp"
|
|
config = {
|
|
"architecture": {
|
|
"forbidden_patterns": [
|
|
{
|
|
"name": "no direct shell",
|
|
"regex": r"\bsystem\s*\(",
|
|
"paths": ["src/**"],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
source.write_text("void old_helper() {}\n", encoding="utf-8")
|
|
self.assertEqual(
|
|
check_source_structure.validate_architecture(root, config), []
|
|
)
|
|
source.write_text("void renamed_helper() {}\n", encoding="utf-8")
|
|
self.assertEqual(
|
|
check_source_structure.validate_architecture(root, config), []
|
|
)
|
|
source.write_text('void renamed_helper() { system("true"); }\n')
|
|
errors = check_source_structure.validate_architecture(root, config)
|
|
|
|
self.assertEqual(errors, ["no direct shell: src/example.cpp:1"])
|
|
|
|
|
|
class LintBaselineTest(unittest.TestCase):
|
|
def baseline(self) -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"tool": "clang-tidy",
|
|
"reviewed_at": "2026-07-29T00:00:00Z",
|
|
"reviewer": "Fold maintainers",
|
|
"findings": [
|
|
{
|
|
"path": "src/example.cpp",
|
|
"severity": "warning",
|
|
"rule": "bugprone-example",
|
|
"message": "existing warning",
|
|
"status": "accepted-risk",
|
|
"rationale": "Reviewed existing debt; gate any additional identities.",
|
|
}
|
|
],
|
|
}
|
|
|
|
def test_line_moves_do_not_turn_reviewed_warning_into_new_warning(self) -> None:
|
|
root = ROOT
|
|
current = compare_lint_baseline.parse_diagnostics(
|
|
f"{root}/src/example.cpp:99:7: warning: existing warning [bugprone-example]",
|
|
root,
|
|
)
|
|
result = compare_lint_baseline.compare_diagnostics(current, self.baseline())
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["counts"]["persisting"], 1)
|
|
self.assertEqual(result["counts"]["new"], 0)
|
|
|
|
def test_additional_occurrence_of_reviewed_warning_is_new(self) -> None:
|
|
current = compare_lint_baseline.parse_diagnostics(
|
|
"\n".join(
|
|
[
|
|
"src/example.cpp:10:2: warning: existing warning [bugprone-example]",
|
|
"src/example.cpp:20:2: warning: existing warning [bugprone-example]",
|
|
]
|
|
),
|
|
ROOT,
|
|
)
|
|
result = compare_lint_baseline.compare_diagnostics(current, self.baseline())
|
|
self.assertEqual(result["status"], "failed")
|
|
self.assertEqual(result["counts"]["persisting"], 1)
|
|
self.assertEqual(result["counts"]["new"], 1)
|
|
self.assertEqual(result["new"][0]["line"], 20)
|
|
|
|
def test_compile_database_relative_paths_normalize_to_project_paths(self) -> None:
|
|
current = compare_lint_baseline.parse_diagnostics(
|
|
"../../include/fold/api.hpp:4:2: warning: diagnostic [bugprone-example]",
|
|
ROOT,
|
|
ROOT / "build" / "lint",
|
|
)
|
|
self.assertEqual(current[0]["path"], "include/fold/api.hpp")
|
|
|
|
def test_external_toolchain_diagnostics_are_not_project_findings(self) -> None:
|
|
current = compare_lint_baseline.parse_diagnostics(
|
|
"/usr/include/example.h:2:1: error: toolchain issue [clang-diagnostic-error]",
|
|
ROOT,
|
|
)
|
|
self.assertEqual(current, [])
|
|
|
|
def test_new_warning_fails_while_full_current_set_remains_counted(self) -> None:
|
|
current = compare_lint_baseline.parse_diagnostics(
|
|
"\n".join(
|
|
[
|
|
"src/example.cpp:10:2: warning: existing warning [bugprone-example]",
|
|
"src/new.cpp:3:1: error: new problem [clang-diagnostic-error]",
|
|
]
|
|
),
|
|
ROOT,
|
|
)
|
|
result = compare_lint_baseline.compare_diagnostics(current, self.baseline())
|
|
self.assertEqual(result["status"], "failed")
|
|
self.assertEqual(result["counts"]["current"], 2)
|
|
self.assertEqual(result["counts"]["new"], 1)
|
|
self.assertEqual(result["new"][0]["path"], "src/new.cpp")
|
|
|
|
def test_baseline_requires_accepted_risk_rationale(self) -> None:
|
|
baseline = self.baseline()
|
|
del baseline["findings"][0]["rationale"]
|
|
errors = compare_lint_baseline.validate_baseline(baseline, "clang-tidy")
|
|
self.assertTrue(any("rationale" in error for error in errors))
|
|
|
|
|
|
class CoverageStatusTest(unittest.TestCase):
|
|
def write_tool(self, directory: Path, name: str, body: str) -> Path:
|
|
path = directory / name
|
|
path.write_text(f"#!/bin/sh\n{body}\n", encoding="utf-8")
|
|
path.chmod(0o755)
|
|
return path
|
|
|
|
def test_test_and_coverage_failures_are_aggregated(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
temp = Path(directory)
|
|
tools = temp / "bin"
|
|
tools.mkdir()
|
|
llvm_config = self.write_tool(tools, "llvm-config", "exit 0")
|
|
self.write_tool(tools, "llvm-cov", "exit 0")
|
|
self.write_tool(
|
|
tools,
|
|
"uname",
|
|
'if [ "${1:-}" = "-m" ]; then echo x86_64; else echo Linux; fi',
|
|
)
|
|
self.write_tool(tools, "conan", "exit 0")
|
|
self.write_tool(tools, "ninja", "exit 0")
|
|
self.write_tool(
|
|
tools,
|
|
"meson",
|
|
'if [ "${1:-}" = "test" ]; then exit "${FAKE_TEST_EXIT:-0}"; fi\nexit 0',
|
|
)
|
|
self.write_tool(
|
|
tools,
|
|
"uvx",
|
|
'echo invoked > "$FAKE_UVX_MARKER"\nexit "${FAKE_COVERAGE_EXIT:-0}"',
|
|
)
|
|
|
|
cases = [
|
|
(0, 0, 0),
|
|
(1, 0, 1),
|
|
(0, 1, 1),
|
|
(1, 1, 1),
|
|
]
|
|
for index, (test_exit, coverage_exit, expected) in enumerate(cases):
|
|
with self.subTest(
|
|
test_exit=test_exit,
|
|
coverage_exit=coverage_exit,
|
|
):
|
|
marker = temp / f"uvx-{index}.marker"
|
|
env = os.environ.copy()
|
|
env.update(
|
|
{
|
|
"PATH": f"{tools}{os.pathsep}{env['PATH']}",
|
|
"LLVM_CONFIG": str(llvm_config),
|
|
"COVERAGE_BUILD_DIR": str(temp / f"coverage-{index}"),
|
|
"FAKE_TEST_EXIT": str(test_exit),
|
|
"FAKE_COVERAGE_EXIT": str(coverage_exit),
|
|
"FAKE_UVX_MARKER": str(marker),
|
|
}
|
|
)
|
|
proc = subprocess.run(
|
|
["bash", str(ROOT / "scripts" / "coverage.sh")],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(
|
|
proc.returncode, expected, proc.stdout + proc.stderr
|
|
)
|
|
self.assertTrue(
|
|
marker.is_file(), "coverage report step did not run"
|
|
)
|
|
|
|
|
|
class FlakyTestEvidenceTest(unittest.TestCase):
|
|
def test_meson_log_parser_emits_sanitized_evidence(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
log = Path(directory) / "testlog.json"
|
|
log.write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "fold:scan_e2e",
|
|
"result": "OK",
|
|
"is_fail": False,
|
|
"duration": 0.125,
|
|
"returncode": 0,
|
|
"stdout": "large output",
|
|
"env": {"SECRET_TOKEN": "must-not-be-copied"},
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
evidence = flaky_evidence.parse_meson_testlog(log, 1, 0)
|
|
|
|
record = evidence["tests"][0]
|
|
self.assertEqual(record["outcome"], "pass")
|
|
self.assertNotIn("env", record)
|
|
self.assertNotIn("stdout", record)
|
|
self.assertNotIn("must-not-be-copied", json.dumps(evidence))
|
|
|
|
def test_pass_then_fail_is_flaky_with_attempt_evidence(self) -> None:
|
|
summary = flaky_evidence.summarize_attempts(
|
|
[
|
|
{
|
|
"attempt": 1,
|
|
"exit_code": 0,
|
|
"tests": [
|
|
{
|
|
"name": "fold:scan_e2e",
|
|
"outcome": "pass",
|
|
"duration_seconds": 0.1,
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"attempt": 2,
|
|
"exit_code": 1,
|
|
"tests": [
|
|
{
|
|
"name": "fold:scan_e2e",
|
|
"outcome": "fail",
|
|
"duration_seconds": 0.2,
|
|
}
|
|
],
|
|
},
|
|
],
|
|
expected_tests=["fold:scan_e2e"],
|
|
)
|
|
|
|
self.assertEqual(summary["counts"]["flaky"], 1)
|
|
self.assertEqual(summary["tests"][0]["classification"], "flaky")
|
|
self.assertEqual(summary["tests"][0]["outcomes"], ["pass", "fail"])
|
|
self.assertFalse(summary["all_attempts_passed"])
|
|
|
|
def test_repeated_failure_is_not_mislabeled_flaky(self) -> None:
|
|
summary = flaky_evidence.summarize_attempts(
|
|
[
|
|
{
|
|
"attempt": attempt,
|
|
"exit_code": 1,
|
|
"tests": [
|
|
{
|
|
"name": "fold:binary_backend",
|
|
"outcome": "fail",
|
|
"duration_seconds": 0.1,
|
|
}
|
|
],
|
|
}
|
|
for attempt in (1, 2, 3)
|
|
],
|
|
expected_tests=["fold:binary_backend"],
|
|
)
|
|
|
|
self.assertEqual(summary["tests"][0]["classification"], "stable-fail")
|
|
self.assertEqual(summary["counts"]["stable-fail"], 1)
|
|
self.assertFalse(summary["all_attempts_passed"])
|
|
|
|
def test_missing_attempt_result_is_inconsistent_harness_evidence(self) -> None:
|
|
summary = flaky_evidence.summarize_attempts(
|
|
[
|
|
{
|
|
"attempt": 1,
|
|
"exit_code": 0,
|
|
"tests": [
|
|
{
|
|
"name": "fold:source_structure",
|
|
"outcome": "pass",
|
|
"duration_seconds": 0.1,
|
|
}
|
|
],
|
|
},
|
|
{"attempt": 2, "exit_code": 2, "tests": []},
|
|
],
|
|
expected_tests=["fold:source_structure"],
|
|
)
|
|
|
|
self.assertEqual(summary["tests"][0]["classification"], "inconsistent")
|
|
self.assertEqual(summary["tests"][0]["outcomes"], ["pass", "missing"])
|
|
self.assertFalse(summary["all_attempts_passed"])
|
|
|
|
|
|
class ImpactedTestSelectionTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.manifest = ROOT / "tests" / "quality" / "impacted_tests.json"
|
|
|
|
def test_cli_report_change_selects_smoke_and_scan_regressions(self) -> None:
|
|
selection = select_impacted_tests.select_tests(
|
|
["src/report/sarif.cpp"], self.manifest
|
|
)
|
|
|
|
self.assertFalse(selection["full_suite"])
|
|
self.assertEqual(
|
|
[item["name"] for item in selection["selected_tests"]],
|
|
[
|
|
"fold:benchmark_evaluation",
|
|
"fold:scan_e2e",
|
|
"fold:source_structure",
|
|
"fold:subprocess_fanout",
|
|
],
|
|
)
|
|
self.assertTrue(all(item["reasons"] for item in selection["selected_tests"]))
|
|
|
|
def test_binary_change_selects_binary_regressions(self) -> None:
|
|
selection = select_impacted_tests.select_tests(
|
|
["src/backend/r2_backend.cpp"], self.manifest
|
|
)
|
|
|
|
names = {item["name"] for item in selection["selected_tests"]}
|
|
self.assertFalse(selection["full_suite"])
|
|
self.assertIn("fold:binary_backend", names)
|
|
self.assertIn("fold:binary_rop", names)
|
|
self.assertNotIn("fold:scan_e2e", names)
|
|
|
|
def test_unmapped_production_change_falls_back_to_full_suite(self) -> None:
|
|
selection = select_impacted_tests.select_tests(
|
|
["src/unmapped/new_subsystem.cpp"], self.manifest
|
|
)
|
|
|
|
self.assertTrue(selection["full_suite"])
|
|
self.assertIn("src/unmapped/new_subsystem.cpp", selection["unmapped_paths"])
|
|
self.assertTrue(selection["fallback_reasons"])
|
|
|
|
def test_explicit_full_suite_override_preserves_reason(self) -> None:
|
|
selection = select_impacted_tests.select_tests(
|
|
["docs/architecture/quality.md"], self.manifest
|
|
)
|
|
|
|
fold_quality.apply_full_test_override(selection, True)
|
|
|
|
self.assertTrue(selection["full_suite"])
|
|
self.assertIn(
|
|
"complete suite requested explicitly", selection["fallback_reasons"]
|
|
)
|
|
|
|
def test_docs_only_change_keeps_required_smoke_tests(self) -> None:
|
|
selection = select_impacted_tests.select_tests(
|
|
["docs/architecture/quality.md"], self.manifest
|
|
)
|
|
|
|
self.assertFalse(selection["full_suite"])
|
|
self.assertEqual(
|
|
[item["name"] for item in selection["selected_tests"]],
|
|
["fold:benchmark_evaluation", "fold:source_structure"],
|
|
)
|
|
|
|
|
|
class QualitySummaryTest(unittest.TestCase):
|
|
def test_verbose_step_writes_full_log_but_returns_bounded_evidence(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
output = Path(directory)
|
|
result = fold_quality.run_step(
|
|
"verbose",
|
|
[sys.executable, "-c", "print('x' * 12000)"],
|
|
artifact_dir=output,
|
|
)
|
|
artifact = result["artifacts"][0]
|
|
log_path = output / artifact["path"]
|
|
|
|
self.assertNotIn("output", result)
|
|
self.assertTrue(log_path.is_file())
|
|
self.assertGreater(len(log_path.read_text(encoding="utf-8")), 12000)
|
|
self.assertLessEqual(len("\n".join(result["evidence"])), 1000)
|
|
self.assertGreater(artifact["estimated_tokens"], 0)
|
|
self.assertEqual(
|
|
artifact["estimated_tokens"],
|
|
estimate_text_tokens(log_path.read_text(encoding="utf-8")),
|
|
)
|
|
|
|
def test_policy_delta_excludes_persisting_and_bounds_new_resolved_details(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
output = Path(directory)
|
|
policy = output / "policy"
|
|
policy.mkdir()
|
|
summary_path = policy / "summary.json"
|
|
summary_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "failed",
|
|
"change_time": {
|
|
"counts": {"new": 25, "persisting": 2, "resolved": 21},
|
|
"new": [f"new-{index}" for index in range(25)],
|
|
"persisting": ["old-1", "old-2"],
|
|
"resolved": [f"resolved-{index}" for index in range(21)],
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(policy / "summary.md").write_text("full report\n", encoding="utf-8")
|
|
|
|
compact = fold_quality.summarize_policy_report(summary_path, output)
|
|
|
|
delta = compact["finding_delta"]
|
|
self.assertNotIn("persisting", delta)
|
|
self.assertEqual(len(delta["new"]), 20)
|
|
self.assertEqual(len(delta["resolved"]), 20)
|
|
self.assertEqual(delta["truncated"], {"new": 5, "resolved": 1})
|
|
self.assertEqual(
|
|
[artifact["path"] for artifact in compact["artifacts"]],
|
|
["policy/summary.json", "policy/summary.md"],
|
|
)
|
|
|
|
def test_no_finding_policy_delta_is_stably_empty(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
output = Path(directory)
|
|
policy = output / "policy"
|
|
policy.mkdir()
|
|
summary_path = policy / "summary.json"
|
|
summary_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "ok",
|
|
"change_time": {
|
|
"counts": {"new": 0, "persisting": 0, "resolved": 0},
|
|
"new": [],
|
|
"persisting": [],
|
|
"resolved": [],
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
first = fold_quality.summarize_policy_report(summary_path, output)
|
|
second = fold_quality.summarize_policy_report(summary_path, output)
|
|
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(first["finding_delta"]["new"], [])
|
|
self.assertEqual(first["finding_delta"]["resolved"], [])
|
|
|
|
|
|
class QualityBranchSelectionTest(unittest.TestCase):
|
|
def git(self, root: Path, *args: str) -> str:
|
|
proc = subprocess.run(
|
|
["git", *args], cwd=root, text=True, capture_output=True, check=True
|
|
)
|
|
return proc.stdout.strip()
|
|
|
|
def initialize_repo(self, root: Path) -> None:
|
|
self.git(root, "init", "-q")
|
|
self.git(root, "config", "user.email", "fold@example.test")
|
|
self.git(root, "config", "user.name", "Fold Test")
|
|
(root / "src").mkdir()
|
|
(root / "src" / "main.cpp").write_text("int main() { return 0; }\n")
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "base")
|
|
|
|
def test_merge_base_diff_selects_committed_pr_sources(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
base = self.git(root, "rev-parse", "HEAD")
|
|
self.git(root, "checkout", "-qb", "feature")
|
|
(root / "src" / "main.cpp").write_text("int main() { return 1; }\n")
|
|
(root / "README.md").write_text("documentation only\n")
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "feature")
|
|
|
|
changed = fold_quality.changed_source_files(
|
|
root=root, base_ref=base, head_ref="HEAD"
|
|
)
|
|
|
|
self.assertEqual(changed, ["src/main.cpp"])
|
|
|
|
def test_test_impact_diff_includes_deleted_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
removed = root / "src" / "removed.cpp"
|
|
removed.write_text("int removed() { return 0; }\n")
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "add removable source")
|
|
removed.unlink()
|
|
|
|
changed = fold_quality.changed_test_paths(root=root)
|
|
|
|
self.assertEqual(changed, ["src/removed.cpp"])
|
|
|
|
def test_local_diff_includes_modified_and_untracked_sources(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
(root / "src" / "main.cpp").write_text("int main() { return 2; }\n")
|
|
(root / "src" / "new.cpp").write_text("int added() { return 1; }\n")
|
|
(root / "notes.txt").write_text("not source\n")
|
|
|
|
changed = fold_quality.changed_source_files(root=root)
|
|
|
|
self.assertEqual(changed, ["src/main.cpp", "src/new.cpp"])
|
|
|
|
def test_changed_paths_preserve_git_filenames_verbatim(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
unusual = "src/weird\nname.cpp"
|
|
(root / unusual).write_text("int unusual() { return 0; }\n")
|
|
|
|
changed = fold_quality.changed_paths(root=root)
|
|
|
|
self.assertIn(unusual, changed)
|
|
|
|
def write_compile_db(self, root: Path, dependencies: dict[str, list[str]]) -> Path:
|
|
build = root / "build"
|
|
build.mkdir(exist_ok=True)
|
|
entries = []
|
|
for source, headers in dependencies.items():
|
|
stem = Path(source).stem
|
|
depfile = build / f"{stem}.o.d"
|
|
depfile.write_text(
|
|
f"{stem}.o: {source} {' '.join(headers)}\n", encoding="utf-8"
|
|
)
|
|
entries.append(
|
|
{
|
|
"directory": str(root),
|
|
"command": f"clang++ -MF {depfile} -c {source}",
|
|
"file": source,
|
|
"output": str(build / f"{stem}.o"),
|
|
}
|
|
)
|
|
compile_db = build / "compile_commands.json"
|
|
compile_db.write_text(json.dumps(entries), encoding="utf-8")
|
|
return compile_db
|
|
|
|
def test_header_only_change_selects_impacted_translation_units(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
(root / "include").mkdir()
|
|
header = root / "include" / "api.hpp"
|
|
header.write_text("int api();\n", encoding="utf-8")
|
|
(root / "src" / "main.cpp").write_text(
|
|
'#include "api.hpp"\nint main() { return api(); }\n', encoding="utf-8"
|
|
)
|
|
(root / "src" / "other.cpp").write_text(
|
|
"int other() { return 0; }\n", encoding="utf-8"
|
|
)
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "add header users")
|
|
compile_db = self.write_compile_db(
|
|
root,
|
|
{
|
|
"src/main.cpp": ["include/api.hpp"],
|
|
"src/other.cpp": [],
|
|
},
|
|
)
|
|
header.write_text("int api(int value = 0);\n", encoding="utf-8")
|
|
|
|
changed = fold_quality.changed_source_files(
|
|
root=root, compile_db=compile_db
|
|
)
|
|
|
|
self.assertEqual(changed, ["src/main.cpp"])
|
|
|
|
def test_ninja_dependencies_select_header_users_when_depfiles_are_consumed(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
(root / "include").mkdir()
|
|
header = root / "include" / "api.hpp"
|
|
header.write_text("int api();\n", encoding="utf-8")
|
|
(root / "src" / "other.cpp").write_text(
|
|
"int other() { return 0; }\n", encoding="utf-8"
|
|
)
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "add dependency fixture")
|
|
compile_db = self.write_compile_db(
|
|
root,
|
|
{
|
|
"src/main.cpp": ["include/api.hpp"],
|
|
"src/other.cpp": [],
|
|
},
|
|
)
|
|
for depfile in (root / "build").glob("*.o.d"):
|
|
depfile.unlink()
|
|
header.write_text("int api(int value = 0);\n", encoding="utf-8")
|
|
dependency_map = {
|
|
(root / "build" / "main.o").resolve(): {header.resolve()},
|
|
(root / "build" / "other.o").resolve(): {
|
|
(root / "src" / "other.cpp").resolve()
|
|
},
|
|
}
|
|
|
|
with mock.patch.object(
|
|
fold_quality,
|
|
"load_ninja_dependencies",
|
|
return_value=dependency_map,
|
|
):
|
|
changed = fold_quality.changed_source_files(
|
|
root=root, compile_db=compile_db
|
|
)
|
|
|
|
self.assertEqual(changed, ["src/main.cpp"])
|
|
|
|
def test_generated_and_vendor_headers_do_not_expand_scan(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
self.initialize_repo(root)
|
|
(root / "build").mkdir()
|
|
(root / "third_party").mkdir()
|
|
generated = root / "build" / "generated.hpp"
|
|
vendor = root / "third_party" / "vendor.hpp"
|
|
generated.write_text("#define GENERATED 1\n", encoding="utf-8")
|
|
vendor.write_text("#define VENDOR 1\n", encoding="utf-8")
|
|
self.git(root, "add", ".")
|
|
self.git(root, "commit", "-qm", "add external headers")
|
|
compile_db = self.write_compile_db(
|
|
root,
|
|
{"src/main.cpp": ["build/generated.hpp", "third_party/vendor.hpp"]},
|
|
)
|
|
generated.write_text("#define GENERATED 2\n", encoding="utf-8")
|
|
vendor.write_text("#define VENDOR 2\n", encoding="utf-8")
|
|
|
|
changed = fold_quality.changed_source_files(
|
|
root=root, compile_db=compile_db
|
|
)
|
|
|
|
self.assertEqual(changed, [])
|
|
|
|
def test_ci_without_branch_metadata_or_compile_db_fails_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
fold = root / "fold"
|
|
fold.write_text("placeholder\n")
|
|
out_dir = root / "quality"
|
|
env = os.environ.copy()
|
|
env["CI"] = "true"
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(FOLD_QUALITY_PATH),
|
|
"--fold",
|
|
str(fold),
|
|
"--out-dir",
|
|
str(out_dir),
|
|
"--skip-format",
|
|
"--skip-tests",
|
|
"--skip-policy",
|
|
],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
summary = json.loads((out_dir / "summary.json").read_text())
|
|
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertEqual(summary["status"], "failed")
|
|
self.assertFalse(summary["steps"][0]["passed"])
|
|
self.assertIn("base/head refs", " ".join(summary["steps"][0]["errors"]))
|
|
self.assertIn("compile database", " ".join(summary["steps"][0]["errors"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|