from __future__ import annotations import ast import argparse import copy import importlib.util import json from pathlib import Path import tempfile import pytest ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts/stages/reaggregate_sp02_q1.py" CONFIG = ( ROOT / "configs/evaluation/speech_kws/SP02_mlperf_tiny_streaming_wakeword_q1_reaggregation.json" ) RESULT_DIR = ( ROOT / "results/quality_speech_audit/SP02_mlcommons_streaming_host_reference" / "q1_mlperf_runner_readme_1p0s_v2" ) MATERIALIZED_INPUTS = ( ROOT / "research/downloads/speech_kws/mlcommons_tiny_master/benchmark/runner/README.md" ).is_file() and ( ROOT / "results/quality_speech_audit/SP02_mlcommons_streaming_host_reference/quality_summary.json" ).is_file() MATERIALIZED_RESULTS = (RESULT_DIR / "q1_quality_summary.json").is_file() def load_module(): spec = importlib.util.spec_from_file_location("reaggregate_sp02_q1", SCRIPT) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_readme_matching_uses_inclusive_one_second_post_end_window() -> None: module = load_module() truth = [[10.0, 11.0]] at_boundary = module.count_fp_fn([12.0], truth, 1.0, 1.0) after_boundary = module.count_fp_fn([12.000001], truth, 1.0, 1.0) assert (at_boundary["false_positives"], at_boundary["false_negatives"]) == (0, 0) assert (after_boundary["false_positives"], after_boundary["false_negatives"]) == (1, 1) def test_half_one_and_one_and_half_second_windows_are_distinct() -> None: module = load_module() truth = [[10.0, 11.0]] times = [11.75] half = module.count_fp_fn(times, truth, 0.5, 1.0) official = module.count_fp_fn(times, truth, 1.0, 1.0) one_half = module.count_fp_fn(times, truth, 1.5, 1.0) assert (half["false_positives"], half["false_negatives"]) == (1, 1) assert (official["false_positives"], official["false_negatives"]) == (0, 0) assert (one_half["false_positives"], one_half["false_negatives"]) == (0, 0) def test_false_positives_use_one_second_debounce() -> None: module = load_module() counts = module.count_fp_fn([1.0, 1.5, 2.0, 2.000001, 3.2], [], 1.0, 1.0) assert counts["false_positives"] == 3 assert counts["false_positive_timestamps_seconds"] == [1.0, 2.000001, 3.2] def test_config_marks_only_one_second_as_official_and_fp_fn_as_primary() -> None: config = json.loads(CONFIG.read_text()) assert config["evaluation"]["analysis_windows_seconds"] == [0.5, 1.0, 1.5] assert config["evaluation"]["official_criterion"]["post_end_window_seconds"] == 1.0 assert config["evaluation"]["official_criterion"]["result_role"] == "OFFICIAL_CRITERION" assert config["evaluation"]["primary_reported_metrics"] == [ "false_positives", "false_negatives", ] assert all(value is False for value in config["prohibited_operations"].values()) def test_reaggregator_cannot_import_or_invoke_model_runtime() -> None: tree = ast.parse(SCRIPT.read_text()) forbidden_imports = [] prohibited_calls = [] for node in ast.walk(tree): if isinstance(node, ast.Import): forbidden_imports.extend( alias.name for alias in node.names if alias.name.split(".")[0] in {"tensorflow", "keras", "torch", "onnxruntime", "ai_edge_litert"} ) elif isinstance(node, ast.ImportFrom) and node.module: if node.module.split(".")[0] in { "tensorflow", "keras", "torch", "onnxruntime", "ai_edge_litert", }: forbidden_imports.append(node.module) elif isinstance(node, ast.Call): if isinstance(node.func, ast.Attribute) and node.func.attr in { "invoke", "predict", "fit", "set_weights", "save_weights", }: prohibited_calls.append(node.func.attr) elif isinstance(node.func, ast.Name) and node.func.id in { "load_model", "Interpreter", "get_model", "clone_model", }: prohibited_calls.append(node.func.id) assert forbidden_imports == [] assert prohibited_calls == [] @pytest.mark.skipif(not MATERIALIZED_INPUTS, reason="compact repository omits SP02 upstream checkout and saved predictions") def test_resume_fingerprint_includes_config_script_and_algorithm_version() -> None: module = load_module() config = json.loads(CONFIG.read_text()) baseline = module.verify_inputs(ROOT, config, CONFIG, SCRIPT) with tempfile.TemporaryDirectory() as temporary: temporary_root = Path(temporary) changed_config = copy.deepcopy(config) changed_config["scope"]["note"] += " harmless fingerprint probe" changed_config_path = temporary_root / "config.json" changed_config_path.write_text(json.dumps(changed_config)) config_changed = module.verify_inputs(ROOT, changed_config, changed_config_path, SCRIPT) changed_script = temporary_root / "reaggregator.py" changed_script.write_bytes(SCRIPT.read_bytes() + b"\n# fingerprint probe\n") script_changed = module.verify_inputs(ROOT, config, CONFIG, changed_script) assert baseline["fingerprint_sha256"] != config_changed["fingerprint_sha256"] assert baseline["fingerprint_sha256"] != script_changed["fingerprint_sha256"] assert baseline["execution_definition"]["metric_algorithm_version"] == module.METRIC_ALGORITHM_VERSION @pytest.mark.skipif(not MATERIALIZED_INPUTS, reason="compact repository omits SP02 saved prediction metadata") def test_saved_metadata_rejects_threshold_tamper() -> None: module = load_module() config = json.loads(CONFIG.read_text()) config["prediction_interface"]["score_threshold"] = 0.999999 with pytest.raises(RuntimeError, match="SAVED_METADATA_MISMATCH:fp32.quality.score_threshold"): module.verify_saved_metadata(ROOT, config) def test_synthetic_official_failure_propagates_standard_failure_fields() -> None: module = load_module() outcome = module.terminal_quality_fields({ "fp32": {"acceptance_status": "PASS"}, "public_int8": {"acceptance_status": "FAIL"}, }) assert outcome == { "status": "FAIL", "failure_code": "FAIL_NUMERICAL_MISMATCH", "failure_detail": "QUALITY_THRESHOLD_EXCEEDED", "failed_variants": ["public_int8"], } def test_synthetic_main_failure_propagates_to_manifest_run_record_and_exit( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: module = load_module() prior = tmp_path / "prior.json" pre_audit = tmp_path / "pre_audit.json" canonical = tmp_path / "canonical.json" prediction = tmp_path / "prediction.csv" timestamp = tmp_path / "timestamp.csv" prior.write_text('{"legacy":true}\n') pre_audit.write_text('{"pre_audit":true}\n') canonical.write_bytes(pre_audit.read_bytes()) prediction.write_text("unused by patched test reader\n") timestamp.write_text( "prediction_index,timestamp_seconds,wakeword_score\n0,100.0,0.99\n" ) result_dir = tmp_path / "result" run_record = tmp_path / "logs" / "run_record.json" config_path = tmp_path / "config.json" expected_canonical = { "bytes": canonical.stat().st_size, "sha256": module.sha256_file(canonical), } variants = { name: { "format": model_format, "prediction_path": str(prediction), "prediction_bytes": prediction.stat().st_size, "prediction_sha256": module.sha256_file(prediction), "timestamp_path": str(timestamp), } for name, model_format in (("fp32", "keras_h5"), ("public_int8", "tflite")) } config = { "model_id": "SP02", "artifact_id": "synthetic-terminal-failure", "scope": {"formal_submission": False}, "protocol_source": {"synthetic": True}, "dataset": {"name": "synthetic unit-test truth", "duration_seconds": 1.0}, "saved_evidence": { "prior_summary": {"path": str(prior)}, "pre_audit_q1_summary": {"path": str(pre_audit)}, "variants": variants, }, "prediction_interface": { "rows": 1, "score_threshold": 0.95, "window_stride_seconds": 1.0, "expected_leading_alignment_seconds": 0.0, "timestamp_alignment": "synthetic persisted timestamp", }, "evaluation": { "official_criterion": { "post_end_window_seconds": 1.0, "false_positive_debounce_seconds": 1.0, "result_role": "OFFICIAL_CRITERION", "protocol_id": "SYNTHETIC_1P0S", }, "analysis_windows_seconds": [0.5, 1.0, 1.5], "auxiliary_result_role": "AUXILIARY_SENSITIVITY_ANALYSIS", "primary_reported_metrics": ["false_positives", "false_negatives"], "acceptance": { "false_positives_max": 0, "false_negatives_max": 0, "comparison": "<=", }, }, "outputs": { "result_dir": str(result_dir), "canonical_summary": str(canonical), "expected_superseded_canonical": expected_canonical, }, "prohibited_operations": {"model_inference": False}, } config_path.write_text(json.dumps(config)) timestamp_record = module.file_record(tmp_path, timestamp) inputs = { "fingerprint_sha256": "synthetic-fingerprint", "execution_definition": { "config": module.file_record(tmp_path, config_path), "reaggregator": module.file_record(tmp_path, SCRIPT), "metric_algorithm_version": module.METRIC_ALGORITHM_VERSION, }, "variants": { name: { "timestamp": timestamp_record, "prior_result": {"path": f"{name}_prior.json", "bytes": 1, "sha256": "0" * 64}, } for name in variants }, "saved_timestamp_alignment_metadata": { "variants": { name: {"aligned_detection_timestamps": 1} for name in variants } }, } monkeypatch.setattr( module, "parse_args", lambda: argparse.Namespace( project_root=tmp_path, config=config_path, output_dir=result_dir, canonical_output=canonical, run_record=run_record, resume=False, ), ) monkeypatch.setattr(module, "verify_inputs", lambda *args: inputs) monkeypatch.setattr(module, "read_truth", lambda *args: [[10.0, 11.0]]) monkeypatch.setattr( module, "read_saved_detections", lambda *args: ([{ "prediction_index": 0, "timestamp_seconds": 100.0, "wakeword_score": 0.99, }], 1, 0.0), ) assert module.main() == 1 summary = json.loads((result_dir / "q1_quality_summary.json").read_text()) manifest = json.loads((result_dir / "execution_manifest.json").read_text()) recorded = json.loads(run_record.read_text()) for document in (summary, manifest, recorded): assert document["status"] == "FAIL" assert document["failure_code"] == "FAIL_NUMERICAL_MISMATCH" assert document["failure_detail"] == "QUALITY_THRESHOLD_EXCEEDED" assert { name: result["status"] for name, result in summary["variant_results"].items() } == {"fp32": "FAIL", "public_int8": "FAIL"} assert canonical.read_bytes() == (result_dir / "q1_quality_summary.json").read_bytes() @pytest.mark.skipif(not MATERIALIZED_RESULTS, reason="compact repository omits materialized SP02 Q1 package") def test_materialized_q1_summary_uses_one_second_only_as_official() -> None: summary = json.loads((RESULT_DIR / "q1_quality_summary.json").read_text()) assert summary["stage"] == "Q1" assert summary["quality_status"] == "PASS" expected = { "fp32": {"0.5s": (15, 6), "1.0s": (5, 6), "1.5s": (5, 6)}, "public_int8": {"0.5s": (14, 6), "1.0s": (4, 6), "1.5s": (4, 6)}, } for variant, windows in expected.items(): quality = summary["quality"][variant] for name, (false_positives, false_negatives) in windows.items(): result = quality["all_window_results"][name] assert (result["false_positives"], result["false_negatives"]) == ( false_positives, false_negatives, ) assert result["official_result"] is (name == "1.0s") assert quality["false_positives"] == windows["1.0s"][0] assert quality["false_negatives"] == windows["1.0s"][1] assert quality["reuse"] == { "saved_prediction_rows_reused": 37470, "prediction_rows_inferred_this_run": 0, "model_inference_invocations": 0, } timestamp = quality["detection_timestamps"] pinned = json.loads(CONFIG.read_text())["saved_evidence"]["variants"][variant] assert timestamp["source"]["sha256"] == pinned["timestamp_sha256"] assert timestamp["reuse"] == "checksum-pinned persisted timestamp sidecar copied byte-for-byte" @pytest.mark.skipif(not MATERIALIZED_RESULTS, reason="compact repository omits materialized SP02 Q1 package") def test_materialized_manifest_hardens_resume_and_preserves_v1() -> None: config = json.loads(CONFIG.read_text()) manifest = json.loads((RESULT_DIR / "execution_manifest.json").read_text()) summary = json.loads((RESULT_DIR / "q1_quality_summary.json").read_text()) assert manifest["config"]["sha256"] == manifest["inputs"]["execution_definition"]["config"]["sha256"] assert manifest["reaggregator"] == manifest["inputs"]["execution_definition"]["reaggregator"] assert manifest["metric_algorithm_version"] == config["prediction_interface"]["metric_algorithm_version"] assert manifest["status"] == summary["status"] == "PASS" assert manifest["failure_code"] == summary["failure_code"] is None assert manifest["outputs"]["preserved_pre_audit_q1_summary"]["sha256"] == config[ "saved_evidence" ]["pre_audit_q1_summary"]["sha256"] @pytest.mark.skipif(not MATERIALIZED_RESULTS, reason="compact repository omits materialized SP02 Q1 package") def test_materialized_resume_rejects_changed_execution_definition() -> None: module = load_module() config = json.loads(CONFIG.read_text()) canonical = ROOT / config["outputs"]["canonical_summary"] with tempfile.TemporaryDirectory() as temporary: temporary_root = Path(temporary) changed_config = copy.deepcopy(config) changed_config["evaluation"]["official_criterion"][ "post_end_window_seconds" ] = 1.5 changed_config_path = temporary_root / "changed_config.json" changed_config_path.write_text(json.dumps(changed_config)) changed_inputs = module.verify_inputs( ROOT, changed_config, changed_config_path, SCRIPT ) with pytest.raises(RuntimeError, match="RESUME_INPUT_FINGERPRINT_MISMATCH"): module.validate_resume_outputs( ROOT, RESULT_DIR, canonical, changed_inputs )