| 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) |
| |
| |
| 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"] |
| ) |
|
|