initial code
This commit is contained in:
@@ -0,0 +1,299 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user