from __future__ import annotations import csv import gzip import hashlib import json import tempfile import unittest from pathlib import Path from typing import Dict, Iterable, List, Mapping, Optional import duckdb from scripts import build_feature_mart as mart def token(character: str) -> str: return character * 64 def bucket(vehicle_token: str) -> int: return int(vehicle_token[:8], 16) % 100 def event( event_id: int, vehicle_token: str, event_ts: str, outcome: Optional[str], make: Optional[str] = "TOYOTA", model: Optional[str] = "CAMRY", model_year: Optional[int] = 2010, source_era: str = "slc", public_county: str = "salt_lake", label_source: Optional[str] = "overall_result", ) -> Dict[str, object]: return { "internal_event_id": event_id, "vehicle_token": vehicle_token, "vehicle_bucket": bucket(vehicle_token), "event_ts": event_ts, "source_era": source_era, "public_county": public_county, "canonical_outcome": outcome, "outcome_label_source": label_source if outcome is not None else None, "program_type": "obd", "test_type": "OBDII", "observed_make": make, "observed_model": model, "observed_model_year": model_year, } def write_export( directory: Path, name: str, rows: Iterable[Mapping[str, object]], start: str, end: str, kind: str = mart.DEVELOPMENT_KIND, ) -> Path: path = directory / name path.parent.mkdir(parents=True, exist_ok=True) materialized = list(rows) with gzip.open(path, "wt", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=mart.EXPECTED_COLUMNS) writer.writeheader() writer.writerows(materialized) digest = mart.file_sha256(path) manifest = { "export_kind": kind, "query_version": ( "inspection_batch_v2" if kind == mart.BOUNDED_KIND else "inspection_history_development_sample_v1" ), "query_sha256": "1" * 64, "generated_at_utc": "2026-07-15T12:00:00+00:00", "source_start_inclusive": start, "source_end_exclusive": end, "vehicle_token_key_version": "v1", "vehicle_token_key_fingerprint": "abcdef1234567890", "rows": len(materialized), "columns": list(mart.EXPECTED_COLUMNS), "compressed_file_sha256": digest, "compressed_file_bytes": path.stat().st_size, "classification": ( "private_pseudonymized_analytical_staging" if kind == mart.BOUNDED_KIND else "private_pseudonymized_development_only" ), } if kind == mart.DEVELOPMENT_KIND: manifest.update( { "population_estimate_allowed": False, "vehicle_sample_limit": 100, "sample_method": "synthetic_test_fixture", } ) manifest_path = path.with_name(path.name + ".manifest.json") manifest_path.write_text(json.dumps(manifest), encoding="utf-8") return path class FeatureMartTest(unittest.TestCase): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory() self.root = Path(self._temporary_directory.name) def tearDown(self) -> None: self._temporary_directory.cleanup() def build(self, input_path: Path) -> mart.BuildSummary: return mart.build_feature_mart( mart.BuildConfig( input_path=input_path, database_path=self.root / "warehouse.duckdb", output_path=self.root / "feature_mart.parquet", memory_limit="512MB", threads=1, allow_external_output=True, ) ) def row_dict( self, connection: duckdb.DuckDBPyConnection, sql: str, parameters=None ) -> Dict[str, object]: cursor = connection.execute(sql, parameters or []) names = [item[0] for item in cursor.description] row = cursor.fetchone() self.assertIsNotNone(row) return dict(zip(names, row)) def test_episode_boundary_quotes_deduplication_and_prior_only_features(self) -> None: vehicle = token("a") rows = [ event( 1, vehicle, "2015-01-01 00:00:00", "pass", make="ACME, MOTORS", model='MODEL "Q"', ), event( 2, vehicle, "2015-01-31 00:00:00", "fail", make="ACME, MOTORS", model='MODEL "Q"', ), # Exact analytical duplicate with a distinct source row ID. event( 3, vehicle, "2015-01-01 00:00:00", "pass", make="ACME, MOTORS", model='MODEL "Q"', ), # Thirty days and one second after the preceding accepted event. event( 4, vehicle, "2015-03-02 00:00:01", "reject", make="ACME, MOTORS", model='MODEL "Q"', ), event( 5, vehicle, "2016-03-02 00:00:01", "fail", make="TARGET, LEAK", model="CURRENT ROW", ), event( 6, vehicle, "2016-03-03 00:00:01", "pass", make="LATER LEAK", model="LATER ATTEMPT", ), ] source = write_export( self.root, "history.csv.gz", rows, "2010-01-01T00:00:00", "2020-01-01T00:00:00", ) summary = self.build(source) self.assertEqual(summary.clean_events, 5) self.assertEqual(summary.episodes, 3) self.assertFalse(summary.population_estimate_allowed) connection = duckdb.connect(str(summary.database_path), read_only=True) try: first_episode = self.row_dict( connection, """ SELECT attempt_count, first_outcome, final_outcome FROM inspection_episodes WHERE vehicle_token = ? AND episode_number = 1 """, [vehicle], ) self.assertEqual(first_episode["attempt_count"], 2) self.assertEqual(first_episode["first_outcome"], "pass") self.assertEqual(first_episode["final_outcome"], "fail") target = self.row_dict( connection, "SELECT * FROM feature_mart WHERE vehicle_token = ? " "AND episode_number = 3", [vehicle], ) self.assertEqual(target["target_nonpass"], 1) self.assertEqual( target["target_outcome_label_source"], "overall_result" ) self.assertTrue(target["eligible_returning_target"]) self.assertEqual(target["prior_episode_count"], 2) self.assertEqual(target["prior_total_attempt_count"], 3) self.assertEqual(target["prior_attempt_count"], 1) self.assertEqual(target["prior_first_outcome"], "reject") self.assertEqual(target["prior_final_outcome"], "reject") self.assertEqual(target["last_observed_make"], "ACME, MOTORS") self.assertEqual(target["last_observed_model"], 'MODEL "Q"') self.assertEqual(target["vehicle_age"], 6) self.assertEqual(target["temporal_partition"], "train") names = { row[0] for row in connection.execute( "DESCRIBE SELECT * FROM feature_mart" ).fetchall() } required = { "vehicle_token", "vehicle_bucket", "is_vin_audit", "episode_number", "episode_start", "first_outcome", "target_nonpass", "eligible_returning_target", "temporal_partition", "vehicle_age", "prior_episode_count", "prior_total_attempt_count", "prior_attempt_count", "days_since_prior_episode", "days_since_prior_adverse", "prior_nonpass_rate", "public_county", "source_era", "target_season", "prior_first_outcome", "prior_final_outcome", "last_observed_make", "last_observed_model", } self.assertTrue(required.issubset(names)) self.assertTrue( { "attempt_count", "final_outcome", "eventually_passed", "episode_end", "obd_result", }.isdisjoint(names) ) finally: connection.close() def test_conflicting_timestamp_is_quarantined_and_null_first_stays_null(self) -> None: conflicted = token("b") unlabeled = token("c") rows = [ event(10, conflicted, "2014-01-01", "pass"), event(11, conflicted, "2016-01-01", "pass"), event(12, conflicted, "2016-01-01", "fail"), event(13, conflicted, "2017-01-01", "pass"), event(20, unlabeled, "2015-01-01", "pass"), event(21, unlabeled, "2016-01-01", None), event(22, unlabeled, "2016-01-02", "pass"), ] source = write_export( self.root, "history.csv.gz", rows, "2010-01-01", "2020-01-01", ) summary = self.build(source) connection = duckdb.connect(str(summary.database_path), read_only=True) try: conflict_count = connection.execute( "SELECT count(*) FROM event_exclusions " "WHERE exclusion_reason = 'conflicting_same_timestamp'" ).fetchone()[0] self.assertEqual(conflict_count, 2) target = self.row_dict( connection, "SELECT first_outcome, target_nonpass, " "eligible_returning_target, eligibility_exclusion_reason " "FROM feature_mart WHERE vehicle_token = ? " "AND episode_start = TIMESTAMP '2016-01-01'", [unlabeled], ) self.assertIsNone(target["first_outcome"]) self.assertIsNone(target["target_nonpass"]) self.assertFalse(target["eligible_returning_target"]) self.assertEqual( target["eligibility_exclusion_reason"], "unlabeled_first_outcome" ) finally: connection.close() def test_bounded_files_are_combined_before_episode_construction(self) -> None: vehicle = token("d") first = write_export( self.root, "inspection_202601.csv.gz", [event(30, vehicle, "2026-01-31 12:00:00", "pass")], "2026-01-01", "2026-02-01", kind=mart.BOUNDED_KIND, ) write_export( self.root, "inspection_202602.csv.gz", [event(31, vehicle, "2026-02-01 12:00:00", "fail")], "2026-02-01", "2026-03-01", kind=mart.BOUNDED_KIND, ) summary = self.build(first.parent) self.assertTrue(summary.population_estimate_allowed) connection = duckdb.connect(str(summary.database_path), read_only=True) try: result = connection.execute( "SELECT count(*), max(attempt_count) FROM inspection_episodes" ).fetchone() self.assertEqual(result, (1, 2)) finally: connection.close() def test_vehicle_quality_eligibility_uses_only_prior_activity(self) -> None: vehicle = token("e") rows: List[Mapping[str, object]] = [ event(40, vehicle, "2014-01-01 00:00:00", "pass"), event(41, vehicle, "2016-01-01 00:00:00", "pass"), ] rows.extend( event(42 + index, vehicle, f"2016-03-10 0{index}:00:00", "fail") for index in range(5) ) rows.append(event(47, vehicle, "2017-04-15 00:00:00", "pass")) source = write_export( self.root, "history.csv.gz", rows, "2010-01-01", "2020-01-01", ) summary = self.build(source) connection = duckdb.connect(str(summary.database_path), read_only=True) try: current_busy_episode = self.row_dict( connection, "SELECT eligible_returning_target, prior_max_events_in_day " "FROM feature_mart WHERE vehicle_token = ? " "AND episode_start = TIMESTAMP '2016-03-10 00:00:00'", [vehicle], ) self.assertTrue(current_busy_episode["eligible_returning_target"]) self.assertEqual(current_busy_episode["prior_max_events_in_day"], 1) after_busy_episode = self.row_dict( connection, "SELECT eligible_returning_target, prior_max_events_in_day, " "eligibility_exclusion_reason FROM feature_mart " "WHERE vehicle_token = ? " "AND episode_start = TIMESTAMP '2017-04-15 00:00:00'", [vehicle], ) self.assertFalse(after_busy_episode["eligible_returning_target"]) self.assertEqual(after_busy_episode["prior_max_events_in_day"], 5) self.assertEqual( after_busy_episode["eligibility_exclusion_reason"], "prior_daily_activity_over_4", ) finally: connection.close() def test_digest_mismatch_fails_before_build(self) -> None: source = write_export( self.root, "history.csv.gz", [event(50, token("f"), "2016-01-01", "pass")], "2010-01-01", "2020-01-01", ) manifest_path = source.with_name(source.name + ".manifest.json") manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["compressed_file_sha256"] = "0" * 64 manifest_path.write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaisesRegex(mart.MartBuildError, "SHA-256 mismatch"): self.build(source) def test_outcome_label_source_is_validated_and_first_row_lineage_is_audit_only( self, ) -> None: utah_vehicle = token("7") rows = [ event( 70, utah_vehicle, "2015-01-01", "pass", source_era="utah", public_county="utah", label_source="utah_obd_proxy", ), event( 71, utah_vehicle, "2016-01-01", "fail", source_era="utah", public_county="utah", label_source="utah_obd_proxy", ), event( 72, token("8"), "2016-01-01", "pass", source_era="slc", public_county="salt_lake", label_source="utah_obd_proxy", ), event( 73, token("9"), "2016-01-01", "pass", label_source=None, ), event( 74, token("0"), "2016-01-01", "pass", label_source="unknown_source", ), ] source = write_export( self.root, "history.csv.gz", rows, "2010-01-01", "2020-01-01", ) summary = self.build(source) connection = duckdb.connect(str(summary.database_path), read_only=True) try: target = self.row_dict( connection, "SELECT target_outcome_label_source, source_era " "FROM feature_mart WHERE vehicle_token = ? " "AND episode_number = 2", [utah_vehicle], ) self.assertEqual( target["target_outcome_label_source"], "utah_obd_proxy" ) self.assertEqual(target["source_era"], "utah") reasons = dict( connection.execute( "SELECT exclusion_reason, count(*) FROM event_exclusions " "GROUP BY exclusion_reason" ).fetchall() ) self.assertEqual(reasons["utah_proxy_non_utah_source"], 1) self.assertEqual(reasons["missing_outcome_label_source"], 1) self.assertEqual(reasons["invalid_outcome_label_source"], 1) schema_names = { row[0] for row in connection.execute( "DESCRIBE SELECT * FROM feature_mart" ).fetchall() } self.assertIn("target_outcome_label_source", schema_names) self.assertNotIn("outcome_label_source", schema_names) self.assertNotIn("obd_result", schema_names) finally: connection.close() def test_bounded_gap_fails_closed(self) -> None: vehicle = token("1") write_export( self.root, "one.csv.gz", [event(60, vehicle, "2026-01-15", "pass")], "2026-01-01", "2026-02-01", kind=mart.BOUNDED_KIND, ) write_export( self.root, "two.csv.gz", [event(61, vehicle, "2026-03-15", "pass")], "2026-03-01", "2026-04-01", kind=mart.BOUNDED_KIND, ) with self.assertRaisesRegex(mart.MartBuildError, "interval gap"): self.build(self.root) incomplete = mart.build_feature_mart( mart.BuildConfig( input_path=self.root, database_path=self.root / "incomplete.duckdb", output_path=self.root / "incomplete.parquet", memory_limit="512MB", threads=1, allow_gaps=True, allow_external_output=True, ) ) self.assertFalse(incomplete.population_estimate_allowed) build_manifest = json.loads( incomplete.manifest_path.read_text(encoding="utf-8") ) self.assertFalse(build_manifest["population_estimate_allowed"]) if __name__ == "__main__": unittest.main()