300 lines
10 KiB
Python
300 lines
10 KiB
Python
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()
|