from __future__ import annotations import csv import json from pathlib import Path import numpy as np import pytest from scripts import validate_ad01_compiled_numerical as validator def _write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, object]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def test_compare_arrays_uses_allclose_for_fp32_and_exact_for_quantized() -> None: reference = np.asarray([[1.0, 2.0]], dtype=np.float32) close = np.asarray([[1.0, 2.0 + 1e-6]], dtype=np.float32) assert validator.compare_arrays( reference, close, atol=1e-5, rtol=1e-5, exact=False )["status"] == "PASS" raw = np.asarray([[1, 2]], dtype=np.int8) changed = np.asarray([[1, 3]], dtype=np.int8) exact = validator.compare_arrays(raw, raw.copy(), atol=0.0, rtol=0.0, exact=True) mismatch = validator.compare_arrays(raw, changed, atol=0.0, rtol=0.0, exact=True) assert exact["status"] == "PASS" assert exact["mismatch_element_count"] == 0 assert mismatch["status"] == "FAIL" assert mismatch["mismatch_element_count"] == 1 assert mismatch["max_abs_error"] == 1.0 def test_auc_pauc_matches_known_perfect_and_worst_rankings() -> None: labels = [0, 0, 1, 1] perfect_auc, perfect_pauc = validator.auc_pauc(labels, [0.1, 0.2, 0.8, 0.9], 0.1) worst_auc, worst_pauc = validator.auc_pauc(labels, [0.9, 0.8, 0.2, 0.1], 0.1) assert perfect_auc == pytest.approx(1.0) assert perfect_pauc == pytest.approx(1.0) assert worst_auc == pytest.approx(0.0) # sklearn's standardized partial AUC maps random to 0.5 and worst-at-low-FPR # below 0.5 rather than returning the raw partial area. assert worst_pauc == pytest.approx((1.0 - 0.1) / (2.0 - 0.1)) def test_recompute_metrics_averages_four_machine_ids() -> None: rows: list[dict[str, object]] = [] for machine_id in validator.MACHINE_IDS: for label, score in ((0, 0.1), (0, 0.2), (1, 0.8), (1, 0.9)): rows.append( { "machine_id": machine_id, "label": label, **{variant: score for variant in validator.VARIANTS}, } ) metrics = validator.recompute_metrics(rows, max_fpr=0.1) assert len(metrics) == 20 averages = [row for row in metrics if row["machine_id"] == "Average"] assert len(averages) == 4 assert all(row["auc"] == pytest.approx(1.0) for row in averages) assert all(row["pauc"] == pytest.approx(1.0) for row in averages) def test_merged_quality_contract_and_metric_tamper_detection(tmp_path: Path) -> None: score_rows: list[dict[str, object]] = [] for machine_id in validator.MACHINE_IDS: for index, (label, score) in enumerate(((0, 0.1), (1, 0.9))): score_rows.append( { "filename": f"{'normal' if label == 0 else 'anomaly'}_{machine_id}_{index}.wav", "machine_id": machine_id, "label": label, "feature_vectors": 196, "fp32_onnxruntime_score": score, "fp32_compiled_score": score, "public_quantized_onnxruntime_score": score, "public_quantized_compiled_score": score, } ) score_path = tmp_path / "compiled_file_scores.csv" _write_csv(score_path, list(score_rows[0]), score_rows) loaded, paths = validator.load_quality_scores(tmp_path) assert paths == [score_path] assert len(loaded) == 8 metrics = validator.recompute_metrics(loaded, 0.1) metric_path = tmp_path / "compiled_quality_metrics.csv" _write_csv( metric_path, ["variant", "machine_id", "auc", "pauc", "max_fpr"], metrics, ) assert validator.validate_metric_csv(metric_path, metrics)["status"] == "PASS" tampered = list(metrics) tampered[0] = {**tampered[0], "auc": 0.25} _write_csv(metric_path, list(tampered[0]), tampered) with pytest.raises(validator.ValidationContractError, match="metric mismatch"): validator.validate_metric_csv(metric_path, metrics) def test_variant_directory_layout_is_joined_by_filename(tmp_path: Path) -> None: fields = [ "filename", "machine_id", "label", "feature_vectors", "onnxruntime_score", "compiled_score", ] fp32 = [{ "filename": "normal_id_01_a.wav", "machine_id": "id_01", "label": 0, "feature_vectors": 196, "onnxruntime_score": 1.0, "compiled_score": 1.1, }] quantized = [{ "filename": "normal_id_01_a.wav", "machine_id": "id_01", "label": 0, "feature_vectors": 196, "onnxruntime_score": 2.0, "compiled_score": 2.1, }] _write_csv(tmp_path / "fp32/file_scores.csv", fields, fp32) _write_csv(tmp_path / "public_quantized/file_scores.csv", fields, quantized) rows, paths = validator.load_quality_scores(tmp_path) assert len(paths) == 2 assert rows == [{ "filename": "normal_id_01_a.wav", "machine_id": "id_01", "label": 0, "feature_vectors": 196, "fp32_onnxruntime": 1.0, "fp32_compiled": 1.1, "public_quantized_onnxruntime": 2.0, "public_quantized_compiled": 2.1, }] def test_no_speed_policy_rejects_latency_claim_or_artifact(tmp_path: Path) -> None: config = {"policy": {"latency_measured": False}} (tmp_path / "summary.json").write_text( json.dumps({"latency_measured": False, "measurement_kind": "NUMERICAL"}) ) assert validator.validate_no_speed_policy(tmp_path, config)["status"] == "PASS" (tmp_path / "claim.json").write_text(json.dumps({"latency_measured": True})) report = validator.validate_no_speed_policy(tmp_path, config) assert report["status"] == "FAIL" assert any("speed policy is not false" in value for value in report["violations"]) def test_quality_merge_manifest_pins_variant_inputs_and_outputs(tmp_path: Path) -> None: result_dir = tmp_path / "result" result_dir.mkdir() merged_scores = result_dir / "compiled_file_scores.csv" merged_metrics = result_dir / "compiled_quality_metrics.csv" merged_score_rows = [{ "filename": "a.wav", "machine_id": "id_01", "label": 0, "feature_vectors": 196, "fp32_onnxruntime_score": 1.0, "fp32_compiled_score": 1.1, "public_quantized_onnxruntime_score": 2.0, "public_quantized_compiled_score": 2.1, }] _write_csv(merged_scores, list(merged_score_rows[0]), merged_score_rows) merged_metrics.write_text("variant\nfp32_compiled\n") artifact_hashes = { "fp32": { "onnx": {"sha256": "1" * 64}, "compiled_library": {"sha256": "2" * 64}, "source_tflite": {"sha256": "3" * 64}, }, "public_quantized": { "onnx": {"sha256": "4" * 64}, "compiled_library": {"sha256": "5" * 64}, "source_tflite": {"sha256": "6" * 64}, }, } inputs: dict[str, dict[str, object]] = {} for variant in ("fp32", "public_quantized"): directory = result_dir / variant directory.mkdir() variant_row = { "filename": "a.wav", "machine_id": "id_01", "label": 0, "feature_vectors": 196, "onnxruntime_score": 1.0 if variant == "fp32" else 2.0, "compiled_score": 1.1 if variant == "fp32" else 2.1, } _write_csv(directory / "file_scores.csv", list(variant_row), [variant_row]) (directory / "quality_metrics.csv").write_text("variant\ncompiled\n") summary = { "variant": variant, "status": "PARTIAL", "failure_code": None, "measurement_status": "MEASURED", "acceptance_status": "THRESHOLD_UNDEFINED", "fidelity_status": "PASS", "latency_measured": False, "evaluation_fingerprint": f"fingerprint-{variant}", "input_artifacts": { "canonical_onnx": artifact_hashes[variant]["onnx"], "compiled_shared_library": artifact_hashes[variant]["compiled_library"], "source_tflite": artifact_hashes[variant]["source_tflite"], }, } summary_path = directory / "quality_summary.json" summary_path.write_text(json.dumps(summary)) inputs[variant] = { "directory": str(directory), "summary_sha256": validator.sha256_file(summary_path), "file_scores_sha256": validator.sha256_file(directory / "file_scores.csv"), "quality_metrics_sha256": validator.sha256_file(directory / "quality_metrics.csv"), "evaluation_fingerprint": f"fingerprint-{variant}", } manifest = { "status": "PASS", "measurement_status": "MEASURED", "acceptance_status": "THRESHOLD_UNDEFINED", "fidelity_status": "PASS", "latency_measured": False, "file_score_rows": 2459, "quality_metric_rows": 20, "inputs": inputs, "outputs": { "compiled_file_scores": { "path": str(merged_scores), "bytes": merged_scores.stat().st_size, "sha256": validator.sha256_file(merged_scores), }, "compiled_quality_metrics": { "path": str(merged_metrics), "bytes": merged_metrics.stat().st_size, "sha256": validator.sha256_file(merged_metrics), }, }, } (result_dir / "compiled_merge_summary.json").write_text(json.dumps(manifest)) report = validator.validate_quality_manifest( result_dir, [merged_scores], merged_metrics, artifact_hashes ) assert report["status"] == "PASS" assert report["output_checksums_match"] is True merged_scores.write_text("filename\ntampered.wav\n") with pytest.raises(validator.ValidationContractError, match="checksum mismatch"): validator.validate_quality_manifest( result_dir, [merged_scores], merged_metrics, artifact_hashes ) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None: canonical = tmp_path / "canonical.csv" rows = [ { "filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav", "machine_id": machine_id, "label": index % 2, "feature_vectors": 196, **{variant: float(index) for variant in validator.VARIANTS}, } for index, machine_id in enumerate(validator.MACHINE_IDS, start=1) ] _write_csv( canonical, ["filename", "machine_id", "label"], [ {"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]} for row in rows ], ) config = { "quality": { "expected_test_files": 4, "expected_total_feature_vectors": 784, "expected_feature_vectors_per_file": 196, } } report = validator.validate_quality_identity(rows, canonical, config) assert report["canonical_filename_label_machine_id_match"] is True tampered = [dict(row) for row in rows] tampered[0]["label"] = 1 - int(tampered[0]["label"]) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) tampered = [dict(row) for row in rows] tampered[0]["machine_id"], tampered[1]["machine_id"] = ( tampered[1]["machine_id"], tampered[0]["machine_id"], ) with pytest.raises( validator.ValidationContractError, match="dataset identity differs from canonical Q1", ): validator.validate_quality_identity(tampered, canonical, config) def test_fixed_fixture_only_directory_cannot_pass_official_q1( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(validator, "CONFIG_RELATIVE", Path("contract.json")) artifact = tmp_path / "artifact.bin" artifact.write_bytes(b"fixed") artifact_contract = { "path": "artifact.bin", "sha256": validator.sha256_file(artifact), } contract = { "model_id": "AD01", "random_seed": 20260806, "quality": { "acceptance_threshold": None, "acceptance_policy": "THRESHOLD_UNDEFINED", "expected_test_files": 2459, "expected_feature_vectors_per_file": 196, "expected_total_feature_vectors": 481964, "max_fpr": 0.1, }, "policy": {"latency_measured": False}, "artifacts": { variant: { role: dict(artifact_contract) for role in ("onnx", "compiled_library", "source_tflite") } for variant in ("fp32", "public_quantized") }, } (tmp_path / "contract.json").write_text(json.dumps(contract)) result_dir = tmp_path / "results" result_dir.mkdir() report = validator.validate(tmp_path, result_dir) assert report["status"] == "FAIL" assert report["official_dcase_q1_recalculation"]["measurement_status"] == "FAIL" assert any( row["check"] == "official_dcase_quality_complete" and row["status"] == "FAIL" for row in report["checks"] )