| 1 | """End-to-end tests of the parse -> normalize -> dedup path and the analytics engine. |
| 2 | These run with no database: the core ALPR logic is pure-function by design.""" |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import os |
| 6 | from datetime import datetime, timezone |
| 7 | |
| 8 | from app.services.normalize import ( |
| 9 | build_alias_map, dedup_hash, normalize_confidence, normalize_direction, |
| 10 | normalize_plate, normalize_read, normalize_timestamp, |
| 11 | ) |
| 12 | from app.services.parsers import get_parser |
| 13 | from app.services.watchlist import WatchEntry, match_entry |
| 14 | |
| 15 | SAMPLE = os.path.join(os.path.dirname(__file__), "..", "sample_data") |
| 16 | |
| 17 | |
| 18 | def test_normalize_plate(): |
| 19 | assert normalize_plate("abc-1234") == "ABC1234" |
| 20 | assert normalize_plate(" a b c 1 ") == "ABC1" |
| 21 | assert normalize_plate(None) == "" |
| 22 | |
| 23 | |
| 24 | def test_normalize_confidence_percent_and_fraction(): |
| 25 | assert normalize_confidence("98") == 0.98 |
| 26 | assert normalize_confidence(0.95) == 0.95 |
| 27 | assert normalize_confidence("150") == 1.0 |
| 28 | assert normalize_confidence("bad") is None |
| 29 | |
| 30 | |
| 31 | def test_normalize_direction_words(): |
| 32 | assert normalize_direction("north") == "N" |
| 33 | assert normalize_direction("NB") == "N" |
| 34 | assert normalize_direction("SE") == "SE" |
| 35 | assert normalize_direction("nonsense") is None |
| 36 | |
| 37 | |
| 38 | def test_normalize_timestamp_tz(): |
| 39 | dt = normalize_timestamp("2026-06-01T08:02:11Z") |
| 40 | assert dt.tzinfo is not None |
| 41 | assert dt == datetime(2026, 6, 1, 8, 2, 11, tzinfo=timezone.utc) |
| 42 | naive = normalize_timestamp("2026-06-01 08:02:11") |
| 43 | assert naive.hour == 8 |
| 44 | |
| 45 | |
| 46 | def test_csv_parse_and_normalize(): |
| 47 | parser = get_parser("flock_sample.csv") |
| 48 | with open(os.path.join(SAMPLE, "flock_sample.csv"), "rb") as fh: |
| 49 | result = parser.parse(fh.read()) |
| 50 | assert len(result.rows) == 16 |
| 51 | alias = build_alias_map(result.headers) |
| 52 | assert alias["plate"] == "Plate" |
| 53 | assert alias["read_ts"] == "Timestamp" |
| 54 | n = normalize_read(result.rows[0], alias) |
| 55 | assert n.plate_norm == "ABC1234" |
| 56 | assert n.external_camera_id == "CAM-DOWNTOWN-01" |
| 57 | assert n.confidence == 0.98 |
| 58 | assert n.state == "TN" |
| 59 | |
| 60 | |
| 61 | def test_json_parse_nested_camera(): |
| 62 | parser = get_parser("flock_sample.json") |
| 63 | with open(os.path.join(SAMPLE, "flock_sample.json"), "rb") as fh: |
| 64 | result = parser.parse(fh.read()) |
| 65 | assert len(result.rows) == 3 |
| 66 | alias = build_alias_map(result.headers) |
| 67 | n = normalize_read(result.rows[0], alias) |
| 68 | assert n.plate_norm == "GHI5566" |
| 69 | assert n.external_camera_id == "CAM-DOWNTOWN-01" |
| 70 | assert n.confidence == 0.97 |
| 71 | |
| 72 | |
| 73 | def test_pdf_parse_extracts_reads(): |
| 74 | pdf = os.path.join(SAMPLE, "flock_report.pdf") |
| 75 | if not os.path.exists(pdf): |
| 76 | import pytest |
| 77 | pytest.skip("sample PDF not generated") |
| 78 | parser = get_parser("flock_report.pdf") |
| 79 | with open(pdf, "rb") as fh: |
| 80 | result = parser.parse(fh.read()) |
| 81 | assert not result.errors |
| 82 | assert len(result.rows) == 4 |
| 83 | alias = build_alias_map(result.headers) |
| 84 | n = normalize_read(result.rows[0], alias) |
| 85 | assert n.plate_norm == "DQR8842" |
| 86 | assert n.external_camera_id == "CAM-DOWNTOWN-01" |
| 87 | assert n.lat == 35.1495 |
| 88 | |
| 89 | |
| 90 | def test_dedup_hash_idempotent_and_bucketed(): |
| 91 | ts1 = datetime(2026, 6, 1, 8, 9, 47, tzinfo=timezone.utc) |
| 92 | ts2 = datetime(2026, 6, 1, 8, 9, 47, 500000, tzinfo=timezone.utc) |
| 93 | h1 = dedup_hash("org", "ABC1234", "cam1", ts1, window_seconds=1) |
| 94 | h2 = dedup_hash("org", "ABC1234", "cam1", ts2, window_seconds=1) |
| 95 | assert h1 == h2 |
| 96 | h3 = dedup_hash("org", "ABC1234", "cam2", ts1, window_seconds=1) |
| 97 | assert h1 != h3 |
| 98 | |
| 99 | |
| 100 | def test_unusable_rows_dropped(): |
| 101 | from app.services.normalize import RawRead |
| 102 | alias = {"plate": "plate", "read_ts": "ts"} |
| 103 | assert normalize_read(RawRead(fields={"plate": "", "ts": "2026-01-01"}), alias) is None |
| 104 | assert normalize_read(RawRead(fields={"plate": "ABC", "ts": ""}), alias) is None |
| 105 | |
| 106 | |
| 107 | def test_watchlist_matching(): |
| 108 | e_exact = WatchEntry(id="1", plate_norm="XYZ7788") |
| 109 | e_partial = WatchEntry(id="2", partial_plate="XYZ") |
| 110 | e_desc = WatchEntry(id="3", vehicle_desc={"color": "Black", "make": "Honda"}) |
| 111 | assert match_entry(e_exact, "XYZ7788") == "exact_plate" |
| 112 | assert match_entry(e_exact, "ABC1234") is None |
| 113 | assert match_entry(e_partial, "XYZ7788") == "partial_plate" |
| 114 | assert match_entry(e_desc, "XYZ7788", {"color": "Black", "make": "Honda"}) == "vehicle_description" |
| 115 | assert match_entry(e_desc, "XYZ7788", {"color": "White", "make": "Honda"}) is None |