tests/test_sp02_q1_reaggregation.py

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
            )