from __future__ import annotations import hashlib import hmac import os import tempfile import unittest import uuid from datetime import datetime, timedelta from pathlib import Path from types import SimpleNamespace from unittest import mock from scripts import export_history_sample as history_export from scripts import export_inspection_batch as exporter VALID_VIN = "1ABCD23EFGH456789" TEST_KEY = bytes(range(32)) class PrivateExportTests(unittest.TestCase): def test_normalize_vin_canonicalizes_case_and_whitespace(self) -> None: self.assertEqual( exporter.normalize_vin(f" {VALID_VIN.lower()}\n"), VALID_VIN, ) def test_normalize_vin_rejects_invalid_values_and_placeholders(self) -> None: invalid_vins = ( "", "1ABCD23EFGH45678", "1ABCD23EFGH4567890", "1ABCD23EFGI456789", "11111111111111111", "AAAAAAAAAAAAAAAAA", "12345678901234567", "98765432109876543", ) for vin in invalid_vins: with self.subTest(vin=vin): self.assertIsNone(exporter.normalize_vin(vin)) def test_hmac_is_deterministic_and_version_domain_separated(self) -> None: expected_v1 = hmac.new( TEST_KEY, b"utah-vehicle-health/v1/vin\0" + VALID_VIN.encode("ascii"), hashlib.sha256, ).hexdigest() token_v1 = exporter.vehicle_token(VALID_VIN, TEST_KEY, "v1") repeated_v1 = exporter.vehicle_token(VALID_VIN, TEST_KEY, "v1") token_v2 = exporter.vehicle_token(VALID_VIN, TEST_KEY, "v2") self.assertEqual(token_v1, expected_v1) self.assertEqual(repeated_v1, token_v1) self.assertNotEqual(token_v2, token_v1) self.assertEqual(len(token_v1), 64) def test_recognized_overall_result_precedes_utah_obd_proxy(self) -> None: self.assertEqual( exporter.canonicalize_outcome( "PASS", "FAIL", "utah", "other", "C" ), ("pass", "overall_result"), ) self.assertEqual( exporter.canonicalize_outcome( " f ", "PASS", " UTAH ", "tsi", "TSI" ), ("fail", "overall_result"), ) def test_utah_only_obd_fallback_uses_recognized_values(self) -> None: self.assertEqual( exporter.canonicalize_outcome( "", " P ", "utah", " ObD ", " obd " ), ("pass", "utah_obd_proxy"), ) self.assertEqual( exporter.canonicalize_outcome( "B", "REJECT", "UTAH", "obd", "OBD" ), ("reject", "utah_obd_proxy"), ) self.assertEqual( exporter.canonicalize_outcome( None, "FAIL", "weber", "obd", "OBD" ), (None, None), ) def test_utah_non_obd_program_or_test_rows_do_not_use_proxy(self) -> None: excluded_cases = ( ("other", "C"), ("tsi", "TSI"), ("obd", "C"), ("other", "OBD"), ("", ""), (None, None), ) for program_type, test_type in excluded_cases: with self.subTest(program_type=program_type, test_type=test_type): self.assertEqual( exporter.canonicalize_outcome( "B", "PASS", "utah", program_type, test_type ), (None, None), ) def test_unknown_blank_and_b_proxy_values_remain_unlabeled(self) -> None: cases = ( (None, None, "utah", "obd", "OBD"), ("", "", "utah", "obd", "OBD"), ("B", "B", "utah", "obd", "OBD"), ("UNKNOWN", "PASS", "slco", "obd", "OBD"), ) for overall, obd, source, program_type, test_type in cases: with self.subTest( overall=overall, obd=obd, source=source, program_type=program_type, test_type=test_type, ): self.assertEqual( exporter.canonicalize_outcome( overall, obd, source, program_type, test_type ), (None, None), ) def test_hmac_configuration_rejects_malformed_or_weak_keys(self) -> None: invalid_keys = ( "", "not-hex", "00" * 31, "00" * 33, "00" * 32, "ab" * 32, ) for encoded_key in invalid_keys: with self.subTest(encoded_key=encoded_key[:12]): with self._hmac_environment(encoded_key): with self.assertRaises(ValueError): exporter.get_hmac_configuration() def test_hmac_configuration_accepts_exact_random_key(self) -> None: with self._hmac_environment(TEST_KEY.hex(), version="key-2026.1"): key, version, fingerprint = exporter.get_hmac_configuration() self.assertEqual(key, TEST_KEY) self.assertEqual(version, "key-2026.1") self.assertRegex(fingerprint, r"^[0-9a-f]{16}$") def test_vehicle_bucket_is_stable_after_vin_normalization(self) -> None: row = self._source_row(VALID_VIN) normalized_row = exporter.private_row(row, TEST_KEY, "v1") lower_row = exporter.private_row( self._source_row(f" {VALID_VIN.lower()} "), TEST_KEY, "v1", ) self.assertIsNotNone(normalized_row) self.assertIsNotNone(lower_row) assert normalized_row is not None assert lower_row is not None token = normalized_row[1] bucket = normalized_row[2] self.assertEqual(lower_row[1], token) self.assertEqual(lower_row[2], bucket) self.assertEqual(bucket, int(str(token)[:8], 16) % 100) self.assertIn(bucket, range(100)) def test_validate_args_enforces_private_output_root(self) -> None: private_output = ( exporter.PRIVATE_DATA_ROOT / "inspection_batches" / f"unit-{uuid.uuid4().hex}.csv.gz" ) validated = exporter.validate_args(self._args(private_output)) self.assertEqual(validated, private_output.resolve()) with tempfile.TemporaryDirectory() as temporary_directory: external_output = Path(temporary_directory) / "private.csv.gz" with self.assertRaises(ValueError): exporter.validate_args(self._args(external_output)) allowed = exporter.validate_args( self._args(external_output, allow_external_output=True) ) self.assertEqual(allowed, external_output.resolve()) def test_declared_output_omits_raw_vin_and_raw_obd_result(self) -> None: self.assertNotIn("vin", exporter.OUTPUT_COLUMNS) self.assertNotIn("raw_vin", exporter.OUTPUT_COLUMNS) self.assertNotIn("obd_result", exporter.OUTPUT_COLUMNS) self.assertNotIn("overall_result", exporter.OUTPUT_COLUMNS) self.assertIn("vehicle_token", exporter.OUTPUT_COLUMNS) self.assertIn("outcome_label_source", exporter.OUTPUT_COLUMNS) transformed = exporter.private_row( self._source_row(VALID_VIN), TEST_KEY, "v1" ) self.assertIsNotNone(transformed) assert transformed is not None self.assertEqual(len(transformed), len(exporter.OUTPUT_COLUMNS)) self.assertEqual( transformed[exporter.OUTPUT_COLUMNS.index("outcome_label_source")], "overall_result", ) self.assertNotIn(VALID_VIN, transformed) def test_both_exporters_share_the_versioned_label_sql(self) -> None: history_sql = history_export.make_sql(0.5, 20260715) self.assertIn(exporter.OUTCOME_SELECT_SQL, exporter.EXTRACT_SQL) self.assertIn(exporter.OUTCOME_SELECT_SQL, history_sql) self.assertIn("s.obd_result", exporter.OUTCOME_SELECT_SQL) self.assertIn( "lower(btrim(s.program_type)) = 'obd'", exporter.OUTCOME_SELECT_SQL, ) self.assertIn( "upper(btrim(s.test_type)) = 'OBD'", exporter.OUTCOME_SELECT_SQL, ) self.assertNotIn("WHEN 'B'", exporter.OUTCOME_SELECT_SQL) self.assertEqual(exporter.QUERY_VERSION, "inspection_batch_v4") self.assertEqual( history_export.QUERY_VERSION, "inspection_history_development_sample_v3", ) def test_feasibility_sql_uses_labeled_utah_only_proxy(self) -> None: sql_path = exporter.PROJECT_ROOT / "sql/10_episode_cohort_feasibility.sql" sql = sql_path.read_text(encoding="utf-8") self.assertIn("lower(btrim(s.county)) = 'utah'", sql) self.assertIn("lower(btrim(s.program_type)) = 'obd'", sql) self.assertIn("upper(btrim(s.test_type)) = 'OBD'", sql) self.assertIn("upper(btrim(s.obd_result))", sql) self.assertIn("AS outcome_label_source", sql) self.assertIn("'utah_obd_proxy'", sql) self.assertNotIn("WHEN 'B' THEN", sql) @staticmethod def _source_row(vin: str) -> tuple[object, ...]: return ( 123, vin, datetime(2025, 1, 15, 12, 0), "slco", "salt_lake", "pass", "overall_result", "obd", "INITIAL", "EXAMPLE", "MODEL", 2020, ) @staticmethod def _args( output: Path, allow_external_output: bool = False ) -> SimpleNamespace: start = datetime(2025, 1, 1) return SimpleNamespace( start=start, end=start + timedelta(days=1), output=output, page_size=100, overwrite=False, allow_external_output=allow_external_output, ) @staticmethod def _hmac_environment( encoded_key: str, version: str = "v1" ) -> mock._patch_dict: missing_key_file = Path(tempfile.gettempdir()) / ( f"uvh-missing-key-{uuid.uuid4().hex}" ) return mock.patch.dict( os.environ, { "VIN_HASH_KEY": encoded_key, "VIN_HASH_KEY_FILE": str(missing_key_file), "VIN_HASH_KEY_VERSION": version, }, clear=False, ) if __name__ == "__main__": unittest.main()