from __future__ import annotations import json from pathlib import Path import subprocess import sys ROOT = Path(__file__).resolve().parents[2] def positive_item(slug: str, *, f1: float = 0.6, coverage: bool = True) -> dict: matches = 60 false_positives = 30 false_negatives = 50 return { "sample_slug": slug, "sample_display_name": f"Mol {slug}", "municipality": "Mol", "operational_zone": slug, "recommended_split": "val", "model_asset_id": "mol-model", "tile_size": 512, "tile_overlap": 64, "threshold": 0.15, "project_id": f"project-{slug}", "area_id": f"area-{slug}", "analysis_run_id": f"run-{slug}", "quality_check_id": f"quality-{slug}", "detection_count": 90, "precision": 0.6666666667, "recall": 0.5454545455, "f1_score": f1, "mean_iou": 0.65, "matches": matches, "false_positives": false_positives, "false_negatives": false_negatives, "coverage_applied": coverage, "coverage_mode": "persisted_tile_manifest_union", "coverage_tile_count": 9, "candidate_raw_count": 90, "candidate_evaluated_count": 90, "candidate_excluded_outside_count": 0, "candidate_clipped_boundary_count": 3, "reference_raw_count": 112, "reference_evaluated_count": 110, "reference_excluded_outside_count": 2, "reference_clipped_boundary_count": 4, "reference_coverage_ratio": 110 / 112, "diagnostic_only": True, "strict_matches": matches, "envelope_matches": 72, "possible_box_to_footprint_mismatch_count": 12, "envelope_precision": 0.8, "envelope_recall": 0.65, "envelope_f1_score": 0.717, } def write_inputs(tmp_path: Path, *, rejected: bool = False) -> tuple[Path, Path, Path]: slugs = ["mol_achterbos", "mol_gompel", "mol_donk", "mol_postel"] items = [ positive_item(slug, f1=0.05 if rejected and slug == "mol_postel" else 0.6, coverage=not rejected) for slug in slugs ] positive_path = tmp_path / "positive.json" positive_path.write_text(json.dumps({"items": items}), encoding="utf-8") background_path = tmp_path / "background.json" background_path.write_text( json.dumps( { "items": [ { "sample_slug": "postel_bos", "background_category": "pure_empty_negative", "model_asset_id": "mol-model", "tile_size": 512, "tile_overlap": 64, "threshold": 0.15, "project_id": "project-background", "area_id": "area-background", "analysis_run_id": "run-background", "tile_count": 9, "detection_count": 2 if rejected else 0, } ] } ), encoding="utf-8", ) manifest_path = tmp_path / "manifest.json" manifest_path.write_text( json.dumps( { "samples": [ { "sample_slug": slug, "display_name": f"Mol {slug}", "municipality": "Mol", "operational_zone": slug, "recommended_split": "val", } for slug in slugs ] } ), encoding="utf-8", ) return positive_path, background_path, manifest_path def run_report(tmp_path: Path, *, rejected: bool = False) -> dict: positive, background, manifest = write_inputs(tmp_path, rejected=rejected) output_dir = tmp_path / "report" result = subprocess.run( [ sys.executable, str(ROOT / "scripts" / "build_mol_operational_benchmark_report.py"), "--positive-summary", str(positive), "--background-summary", str(background), "--manifest-path", str(manifest), "--output-dir", str(output_dir), "--base-url", "http://example.test", ], cwd=ROOT, check=True, capture_output=True, text=True, ) assert "Mol operational benchmark report passed" in result.stdout assert (output_dir / "mol_operational_benchmark_report.md").is_file() return json.loads((output_dir / "mol_operational_benchmark_report.json").read_text(encoding="utf-8")) def test_mol_benchmark_accepts_coverage_safe_multi_zone_evidence(tmp_path: Path) -> None: report = run_report(tmp_path) assert report["status"] == "accepted" assert report["recommendation"] == "retain_or_promote_candidate" decision = report["recommended_candidate"] assert decision["decision"] == "operationally_accepted" assert decision["positive_sample_count"] == 4 assert decision["background_sample_count"] == 1 assert decision["total_references_raw"] == 448 assert decision["total_references_evaluated"] == 440 assert decision["total_box_to_footprint_mismatch_count"] == 48 assert decision["total_background_detections"] == 0 assert decision["failed_gates"] == [] def test_mol_benchmark_rejects_missing_coverage_zone_collapse_and_background_pressure(tmp_path: Path) -> None: report = run_report(tmp_path, rejected=True) assert report["status"] == "review_required" assert report["recommended_candidate"] is None decision = report["candidate_decisions"][0] assert decision["decision"] == "review_required" assert "coverage_provenance" in decision["failed_gates"] assert "minimum_zone_f1" in decision["failed_gates"] assert "background_false_positive_pressure" in decision["failed_gates"] def test_mol_benchmark_is_wired_into_existing_operator_pipeline() -> None: matrix = (ROOT / "scripts" / "run_detection_quality_matrix.sh").read_text(encoding="utf-8") multi = (ROOT / "scripts" / "run_multi_sample_detection_quality_matrix.sh").read_text(encoding="utf-8") runner = (ROOT / "scripts" / "run_mol_operational_validation.sh").read_text(encoding="utf-8") readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8") dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8") assert 'findings.get("coverage")' in matrix assert 'findings.get("box_to_footprint_diagnostics")' in matrix assert '"reference_coverage_ratio"' in matrix assert '"possible_box_to_footprint_mismatch_count"' in matrix assert 'enriched["operational_zone"]' in multi assert "build_mol_operational_benchmark_report.py" in runner assert "MOL_MIN_REFERENCE_COVERAGE" in runner assert 'MOL_MIN_REFERENCE_COVERAGE="${MOL_MIN_REFERENCE_COVERAGE:-0.90}"' in runner assert 'default=0.90' in (ROOT / "scripts" / "build_mol_operational_benchmark_report.py").read_text(encoding="utf-8") assert "py_compile scripts/build_mol_operational_benchmark_report.py" in readiness assert "COPY scripts/build_mol_operational_benchmark_report.py" in dockerfile assert "fixture_mode" not in runner assert "manual-fixture-detector" not in runner