from __future__ import annotations import hashlib import json import tempfile import unittest from pathlib import Path from unittest import mock import duckdb from scripts import export_dashboard_data as dashboard_export class DashboardExportTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) self.mart = self.root / "private_feature_mart.parquet" self.mart_manifest = self.root / "private_feature_mart.parquet.manifest.json" self.model_manifest = self.root / "model_manifest.json" self.model_metrics = self.root / "model_metrics.json" self.public_root = (self.root / "dashboard/public/data").resolve() self._write_mart() self._write_mart_manifest() self._write_model_files() def tearDown(self) -> None: self.temporary.cleanup() def test_development_mart_is_refused_without_preview_flag(self) -> None: with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.ManifestError): self._export(development_preview=False) self.assertFalse(self.public_root.exists()) def test_preview_is_suppressed_deterministic_and_contains_no_locked_data(self) -> None: with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): paths = self._export(development_preview=True) expected_names = set(dashboard_export.PUBLIC_ASSET_NAMES) | { dashboard_export.SHA256_MANIFEST_NAME } self.assertEqual(set(paths), expected_names) first_bytes = { name: (self.public_root / name).read_bytes() for name in expected_names } joined = b"\n".join(first_bytes.values()).lower() for forbidden in ( b"lockedmake", b"secret_locked", b"vehicle_token", b"internal_event_id", b'"vin"', b'"plate"', b'"zip"', b'"station"', b"episode_start", b"target_outcome_label_source", b'"prediction"', ): self.assertNotIn(forbidden, joined) for name in expected_names: payload = json.loads((self.public_root / name).read_text(encoding="utf-8")) self.assertIs(payload["development_preview"], True) self.assertIs(payload["population_estimate_allowed"], False) scorecards = self._asset("cohort_scorecard.json")["rows"] self.assertEqual( scorecards, [ { "prior_make": "SAFE", "prior_model": "SUPPORTED", "support_rounded": 400, "nonpass_rate": 0.167, } ], ) overview = self._asset("overview_period_county.json")["rows"] self.assertEqual(len(overview), 3) self.assertEqual(overview[0]["support_rounded"], 800) self.assertNotIn("n", overview[0]) self.assertNotIn("nonpass_count", overview[0]) diagnostics = self._asset("model_diagnostics.json")["rows"] self.assertEqual( {row["partition"] for row in diagnostics}, {"train", "tune", "calibrate"}, ) self.assertEqual({row["model"] for row in diagnostics}, {"test_model"}) self.assertTrue(all("episodes" not in row for row in diagnostics)) data_manifest = self._asset("data_manifest.json") self.assertEqual(data_manifest["model_versions"], ["test_model_v1"]) self.assertRegex(data_manifest["release_id"], r"^[0-9a-f]{64}$") checksum_manifest = self._asset(dashboard_export.SHA256_MANIFEST_NAME) for entry in checksum_manifest["files"]: content = (self.public_root / entry["name"]).read_bytes() self.assertEqual(hashlib.sha256(content).hexdigest(), entry["sha256"]) with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): self._export(development_preview=True, overwrite=True) second_bytes = { name: (self.public_root / name).read_bytes() for name in expected_names } self.assertEqual(first_bytes, second_bytes) def test_release_id_includes_publication_policy_version(self) -> None: mart_digest = "a" * 64 provenance = [("model_v1", "b" * 64)] release_id = dashboard_export._release_id(mart_digest, provenance) with mock.patch.object( dashboard_export, "PUBLICATION_POLICY_VERSION", dashboard_export.PUBLICATION_POLICY_VERSION + "_changed", ): changed_release_id = dashboard_export._release_id( mart_digest, provenance ) self.assertNotEqual(release_id, changed_release_id) def test_scorecards_fold_only_new_family_aliases_before_suppression(self) -> None: connection = duckdb.connect(":memory:") try: connection.execute( """ CREATE TABLE safe_mart ( vehicle_token VARCHAR, eligible_returning_target BOOLEAN, target_nonpass INTEGER, last_observed_make VARCHAR, last_observed_model VARCHAR ) """ ) rows = [] def add_group(make: str, model: str, nonpass: int = 10) -> None: group_number = len(rows) for index in range(60): rows.append( ( "group-{}-vehicle-{}".format(group_number, index), True, 1 if index < nonpass else 0, make, model, ) ) # Every raw-string group is below the 100-vehicle threshold. Each # approved pair reaches 120 only after its explicit aliases merge. canonical_groups = { ("CHEVROLET", "SILVERADO 1500"): ( ("CHEVROLET", "SILVERADO 1500 LT"), ("CHEVR", "K15 SILVERADO"), ), ("DODGE", "RAM 1500"): ( ("DODGE", "RAM 1500 QUAD"), ("DODGE", "RAM PICKUP 1500"), ), ("HYUNDAI", "ELANTRA"): ( ("HYUNDAI", "ELANTRA GLS"), ("HYUND", "ELANTRA"), ), ("NISSAN", "ALTIMA"): ( ("NISSAN", "ALTIMA 2.5 S"), ("NISSA", "ALTIMA"), ), ("SUBARU", "OUTBACK"): ( ("SUBARU", "LEGACY OUTBACK 2.5I AWD"), ("SUBAR", "OUTBACK"), ), ("TOYOTA", "4RUNNER"): ( ("TOYOTA", "4RUNNER SR5"), ("TOYOT", "4RUNNER 4WD"), ), ("TOYOTA", "COROLLA"): ( ("TOYOTA", "COROLLA CE LE S"), ("TOYOT", "COROLLA"), ), ("TOYOTA", "TACOMA"): ( ("TOYOTA", "TACOMA V6"), ("TOYOT", "TACOMA 4WD"), ), } for variants in canonical_groups.values(): for make, model in variants: add_group(make, model) # These separately marketed lines share a prefix with a canonical # family but must remain distinct and suppressed at this support. for make, model in ( ("HONDA", "ACCORD CROSS TOUR EXL"), ("HONDA", "ACCORD CROSSTOUR"), ("HONDA", "CIVIC CRX SI"), ("HONDA", "CIVIC DEL SOL SI"), ("TOYOT", "CAMRY SOLARA"), ("TOYOTA", "COROLLA IM"), ("DODGE", "RAM 1500 VAN"), ("FORD", "F1500"), ("NISSAN", "ALTIMAX"), ("TOYOTA", "CAMRYX"), ): add_group(make, model, nonpass=30) connection.executemany( "INSERT INTO safe_mart VALUES (?, ?, ?, ?, ?)", rows ) scorecards = dashboard_export._scorecard_rows(connection) finally: connection.close() self.assertEqual( [(row["prior_make"], row["prior_model"]) for row in scorecards], sorted(canonical_groups), ) self.assertTrue( all(row["support_rounded"] == 100 for row in scorecards) ) self.assertTrue(all(row["nonpass_rate"] == 0.167 for row in scorecards)) def test_legacy_scorecard_cells_are_not_broadened_by_aliases(self) -> None: connection = duckdb.connect(":memory:") try: connection.execute( """ CREATE TABLE safe_mart ( vehicle_token VARCHAR, eligible_returning_target BOOLEAN, target_nonpass INTEGER, last_observed_make VARCHAR, last_observed_model VARCHAR ) """ ) rows = [] def add_group( make: str, model: str, n: int, nonpass: int, ) -> None: group_number = len(rows) for index in range(n): rows.append( ( "group-{}-vehicle-{}".format(group_number, index), True, 1 if index < nonpass else 0, make, model, ) ) legacy_cells = { ("FORD", "F150"): 12, ("HONDA", "ACCORD"): 18, ("HONDA", "CIVIC"): 24, ("TOYOTA", "CAMRY"): 30, } aliases = { ("FORD", "F150"): ( ("FORD", "F150 4WD"), ("ford", "f150"), ), ("HONDA", "ACCORD"): ( ("HONDA", "ACCORD LX"), ("honda", "accord"), ), ("HONDA", "CIVIC"): ( ("HONDA", "CIVIC EX"), ("honda", "civic"), ), ("TOYOTA", "CAMRY"): ( ("TOYOTA", "CAMRY LE"), ("TOYOT", "CAMRY"), ), } for (make, model), nonpass in legacy_cells.items(): add_group(make, model, 120, nonpass) # This raw key renders as the legacy key after safe-category # trimming. It must not become a second public identity. add_group(" " + make + " ", " " + model + " ", 120, 60) # Each alias is suppressed alone but would pass support if the # two were folded together, making accidental broadening clear. for alias_make, alias_model in aliases[(make, model)]: add_group(alias_make, alias_model, 60, 30) connection.executemany( "INSERT INTO safe_mart VALUES (?, ?, ?, ?, ?)", rows ) scorecards = dashboard_export._scorecard_rows(connection) finally: connection.close() self.assertEqual(len(scorecards), len(legacy_cells)) by_cell = { (row["prior_make"], row["prior_model"]): row for row in scorecards } self.assertEqual(set(by_cell), set(legacy_cells)) for cell, nonpass in legacy_cells.items(): with self.subTest(cell=cell): self.assertEqual(by_cell[cell]["support_rounded"], 100) self.assertEqual( by_cell[cell]["nonpass_rate"], round(nonpass / 120, 3) ) def test_unlocked_model_manifest_is_refused(self) -> None: manifest = self._read_json(self.model_manifest) manifest["locked_test_evaluated"] = True self._write_json(self.model_manifest, manifest) with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.ManifestError): self._export(development_preview=True) self.assertFalse(self.public_root.exists()) def test_locked_metric_row_is_refused_even_when_manifest_is_closed(self) -> None: metrics = self._read_json(self.model_metrics) locked = dict(metrics[0]) locked["partition"] = "locked_test" metrics.append(locked) self._write_json(self.model_metrics, metrics) self._refresh_metrics_digest() with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.ManifestError): self._export(development_preview=True) self.assertFalse(self.public_root.exists()) def test_metrics_digest_mismatch_is_refused(self) -> None: metrics = self._read_json(self.model_metrics) metrics[0]["average_precision"] = 0.99 self._write_json(self.model_metrics, metrics) with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.ManifestError): self._export(development_preview=True) self.assertFalse(self.public_root.exists()) def test_overwrite_refuses_unexpected_file(self) -> None: with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): self._export(development_preview=True) unexpected = self.public_root / "private-notes.txt" unexpected.write_text("must not ride along\n", encoding="utf-8") with self.assertRaises(dashboard_export.PublicSchemaError): self._export(development_preview=True, overwrite=True) self.assertTrue(unexpected.exists()) def test_overwrite_refuses_directory_in_public_contract(self) -> None: self.public_root.mkdir(parents=True) (self.public_root / "data_manifest.json").mkdir() with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.PublicSchemaError): self._export(development_preview=True, overwrite=True) def test_pre_2025_locked_partition_aliases_are_refused(self) -> None: for partition in dashboard_export.LOCKED_PARTITION_ALIASES: with self.subTest(partition=partition): self._move_locked_rows_before_boundary(partition) with mock.patch.object( dashboard_export, "PUBLIC_DATA_ROOT", self.public_root ): with self.assertRaises(dashboard_export.ManifestError): self._export(development_preview=True) self.assertFalse(self.public_root.exists()) def test_public_validator_rejects_sensitive_keys_and_exact_timestamps(self) -> None: base = { "schema_version": dashboard_export.SCHEMA_VERSION, "development_preview": True, "population_estimate_allowed": False, } for unsafe_field in ( "technician_id", "inspection_technician_code", "inspector_identifier", "database_password", "api_secret_value", ): with self.subTest(unsafe_field=unsafe_field): with self.assertRaises(dashboard_export.PublicSchemaError): dashboard_export._scan_public_value( {unsafe_field: "not-public"} ) unsafe_key = { **base, "rows": [ { "year": 2022, "quarter": 1, "public_county": "salt_lake", "support_rounded": 100, "nonpass_rate": 0.2, "vehicle_token": "not-public", } ], } with self.assertRaises(dashboard_export.PublicSchemaError): dashboard_export.validate_public_asset( "overview_period_county.json", unsafe_key ) exact_timestamp = { **base, "rows": [ { "year": 2022, "quarter": 1, "public_county": "2022-01-01T12:34:56", "support_rounded": 100, "nonpass_rate": 0.2, } ], } with self.assertRaises(dashboard_export.PublicSchemaError): dashboard_export.validate_public_asset( "overview_period_county.json", exact_timestamp ) def _export( self, development_preview: bool, overwrite: bool = False, ) -> dict: return dashboard_export.export_dashboard_data( mart=self.mart, mart_manifest=self.mart_manifest, bundles=[ dashboard_export.ModelBundle( manifest=self.model_manifest, metrics=self.model_metrics, ) ], output_dir=self.public_root, development_preview=development_preview, overwrite=overwrite, ) def _write_mart(self) -> None: connection = duckdb.connect(":memory:") try: connection.execute( """ CREATE TABLE mart ( vehicle_token VARCHAR, is_vin_audit BOOLEAN, episode_number BIGINT, episode_start TIMESTAMP, first_outcome VARCHAR, target_outcome_label_source VARCHAR, target_nonpass INTEGER, eligible_returning_target BOOLEAN, temporal_partition VARCHAR, public_county VARCHAR, source_era VARCHAR, vehicle_age INTEGER, last_observed_make VARCHAR, last_observed_model VARCHAR ) """ ) rows = [] rows.extend(self._group(120, 20, "SAFE", "SUPPORTED")) rows.extend(self._group(99, 19, "LOW", "SUPPORT")) rows.extend(self._group(120, 9, "LOW", "NONPASS")) rows.extend(self._group(120, 111, "LOW", "PASS")) rows.extend( self._group( 120, 60, "ENTITY", "LOWTOTAL", pass_vehicle_count=40, nonpass_vehicle_count=40, ) ) rows.extend( self._group( 120, 20, "ENTITY", "LOWNONPASS", nonpass_vehicle_count=9, ) ) rows.extend( self._group( 120, 100, "ENTITY", "LOWPASS", pass_vehicle_count=9, ) ) rows.extend( self._group( 120, 20, "SAFE", "SUPPORTED", temporal_partition="tune", ) ) rows.extend( self._group( 120, 20, "SAFE", "SUPPORTED", temporal_partition="calibrate", ) ) for index in range(120): rows.append( ( "secret_vehicle_token", False, 2, "2025-01-15 12:34:56", "secret_locked", "secret_locked_source", 999, True, "locked_test", "locked_county", "locked_source", 5, "LOCKEDMAKE", "LOCKEDMODEL", ) ) connection.executemany( "INSERT INTO mart VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", rows, ) connection.table("mart").write_parquet(str(self.mart)) finally: connection.close() @staticmethod def _group( n: int, nonpass: int, make: str, model: str, temporal_partition: str = "train", pass_vehicle_count: int = None, nonpass_vehicle_count: int = None, ) -> list: partition_dates = { "train": "2022-01-15 08:00:00", "tune": "2023-01-15 08:00:00", "calibrate": "2024-01-15 08:00:00", } pass_count = n - nonpass pass_vehicle_count = pass_count if pass_vehicle_count is None else pass_vehicle_count nonpass_vehicle_count = ( nonpass if nonpass_vehicle_count is None else nonpass_vehicle_count ) rows = [] for index in range(n): adverse = index < nonpass class_index = index if adverse else index - nonpass class_vehicle_count = ( nonpass_vehicle_count if adverse else pass_vehicle_count ) outcome = "nonpass" if adverse else "pass" vehicle_token = "{}_{}_{}_{}_{}".format( make, model, temporal_partition, outcome, class_index % class_vehicle_count, ) rows.append( ( vehicle_token, False, 2, partition_dates[temporal_partition], "fail" if adverse else "pass", "overall_result", 1 if adverse else 0, True, temporal_partition, "salt_lake", "slc", 5, make, model, ) ) return rows def _write_mart_manifest(self) -> None: connection = duckdb.connect(":memory:") try: columns = [ {"name": row[0], "duckdb_type": row[1]} for row in connection.execute( "DESCRIBE SELECT * FROM read_parquet(?)", [str(self.mart)] ).fetchall() ] finally: connection.close() self._write_json( self.mart_manifest, { "build_kind": "private_leakage_safe_inspection_feature_mart", "classification": "private_pseudonymized_analytical_mart", "source_data_kind": "vehicle_history_development_sample", "population_estimate_allowed": False, "episode_gap_days": 30, "columns": columns, "parquet_sha256": self._sha256(self.mart), }, ) def _write_model_files(self) -> None: connection = duckdb.connect(":memory:") try: private_support = { partition: (episodes, nonpass, vehicles) for partition, episodes, nonpass, vehicles in connection.execute( """ SELECT temporal_partition, count(*)::BIGINT, sum(target_nonpass)::BIGINT, count(DISTINCT vehicle_token)::BIGINT FROM read_parquet(?) WHERE temporal_partition IN ('train', 'tune', 'calibrate') AND eligible_returning_target AND target_nonpass IN (0, 1) AND NOT is_vin_audit GROUP BY 1 """, [str(self.mart)], ).fetchall() } finally: connection.close() rows = [] for partition, value in ( ("train", 0.25), ("tune", 0.24), ("calibrate", 0.23), ): episodes, nonpass, vehicles = private_support[partition] rows.append( { "model": "test_model", "partition": partition, "cohort": "non_audit", "episodes": episodes, "nonpass": nonpass, "vehicles": vehicles, "average_precision": value, "brier": 0.1, "log_loss": 0.3, "roc_auc": 0.7, "top_10_capture": 0.2, } ) for model, episodes, nonpass in ( ("low_support", 99, 20), ("low_nonpass", 120, 9), ("low_pass", 120, 111), ("low_vehicles", 120, 20), ): rows.append( { "model": model, "partition": "train", "cohort": "non_audit", "episodes": episodes, "nonpass": nonpass, "vehicles": 99 if model == "low_vehicles" else episodes, "average_precision": 0.2, "brier": 0.1, "log_loss": 0.3, "roc_auc": 0.7, "top_10_capture": 0.2, } ) self._write_json(self.model_metrics, rows) self._write_json( self.model_manifest, { "classification": "private_model_artifact_no_row_predictions", "input_sha256": self._sha256(self.mart), "metrics_sha256": self._sha256(self.model_metrics), "locked_test_evaluated": False, "model_version": "test_model_v1", }, ) def _refresh_metrics_digest(self) -> None: manifest = self._read_json(self.model_manifest) manifest["metrics_sha256"] = self._sha256(self.model_metrics) self._write_json(self.model_manifest, manifest) def _move_locked_rows_before_boundary(self, partition: str) -> None: rewritten = self.root / "rewritten.parquet" connection = duckdb.connect(":memory:") try: connection.execute( "CREATE TABLE rewritten AS SELECT * FROM read_parquet(?)", [str(self.mart)], ) connection.execute( """ UPDATE rewritten SET temporal_partition = ?, episode_start = TIMESTAMP '2024-12-31 12:34:56' WHERE last_observed_make = 'LOCKEDMAKE' """, [partition], ) connection.table("rewritten").write_parquet(str(rewritten)) finally: connection.close() rewritten.replace(self.mart) mart_manifest = self._read_json(self.mart_manifest) mart_manifest["parquet_sha256"] = self._sha256(self.mart) self._write_json(self.mart_manifest, mart_manifest) model_manifest = self._read_json(self.model_manifest) model_manifest["input_sha256"] = self._sha256(self.mart) self._write_json(self.model_manifest, model_manifest) def _asset(self, name: str) -> object: return self._read_json(self.public_root / name) @staticmethod def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() @staticmethod def _write_json(path: Path, value: object) -> None: path.write_text( json.dumps(value, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8", ) @staticmethod def _read_json(path: Path) -> object: return json.loads(path.read_text(encoding="utf-8")) if __name__ == "__main__": unittest.main()