"""Unit tests locking in known field mapping fixes (FIX-E06, FIX-M16, BUG-03).""" import sys import os import pytest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) os.environ.setdefault("TRACKSOLID_APP_KEY", "test_key") os.environ.setdefault("TRACKSOLID_APP_SECRET", "test_secret") os.environ.setdefault("TRACKSOLID_USER_ID", "test_user") os.environ.setdefault("TRACKSOLID_PWD_MD5", "test_md5") os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost:5432/test") from ts_shared_rev import clean, clean_ts, clean_num from webhook_receiver_rev import _parse_trip_ts, unix_to_ts class TestFIXE06AlarmFieldMapping: """FIX-E06: Poll alarm endpoint uses alertTypeId/alarmTypeName/alertTime.""" def test_poll_uses_alert_type_id(self): """Alarm poll response must use alertTypeId, not alarmType.""" api_alarm = { "imei": "123456789012345", "alertTypeId": "4", # CORRECT poll field "alarmType": "WRONG_FIELD", # webhook field - should NOT be used for polls "alarmTypeName": "Speeding", "alertTime": "2024-04-12 07:30:00", } # FIX-E06: extract using alertTypeId (poll field name) alarm_type = clean(api_alarm.get("alertTypeId")) assert alarm_type == "4", "Must use alertTypeId not alarmType for poll responses" def test_poll_uses_alarm_type_name(self): """Alarm name must come from alarmTypeName, not alarmName.""" api_alarm = { "alertTypeId": "4", "alarmTypeName": "Speeding", # CORRECT poll field "alarmName": "WRONG_FIELD", # webhook field "alertTime": "2024-04-12 07:30:00", } alarm_name = clean(api_alarm.get("alarmTypeName")) assert alarm_name == "Speeding" def test_poll_uses_alert_time(self): """Alarm time must come from alertTime, not alarmTime.""" api_alarm = { "alertTypeId": "4", "alarmTypeName": "Speeding", "alertTime": "2024-04-12 07:30:00", # CORRECT poll field "alarmTime": "WRONG_FIELD", # webhook field } alarm_time = clean_ts(api_alarm.get("alertTime")) assert alarm_time == "2024-04-12 07:30:00" def test_wrong_field_names_return_none(self): """Using incorrect webhook field names on poll data returns None (the bug).""" api_alarm = {"alertTypeId": "4", "alarmTypeName": "Speeding", "alertTime": "2024-04-12 07:30:00"} # These are webhook fields — should NOT be present in poll responses assert clean(api_alarm.get("alarmType")) is None assert clean(api_alarm.get("alarmName")) is None assert clean_ts(api_alarm.get("alarmTime")) is None class TestFIXM16DistanceUnits: """FIX-M16: Trip distance arrives in METRES from API, must be stored as km.""" def test_metres_divided_by_1000(self): """15000 metres from API → 15.0 km stored.""" raw_dist_metres = 15000 dist_km = round(raw_dist_metres / 1000.0, 4) assert dist_km == pytest.approx(15.0) def test_small_distance(self): """500 metres → 0.5 km.""" assert round(500 / 1000.0, 4) == pytest.approx(0.5) def test_none_distance(self): """None distance stays None (no division by zero).""" raw_dist = clean_num(None) dist_km = round(raw_dist / 1000.0, 4) if raw_dist is not None else None assert dist_km is None def test_zero_distance(self): """0 metres → 0.0 km.""" raw_dist = clean_num(0) dist_km = round(raw_dist / 1000.0, 4) if raw_dist is not None else None assert dist_km == pytest.approx(0.0) def test_non_divided_would_be_wrong(self): """Verify that NOT dividing produces obviously wrong km values.""" raw_dist_metres = 15000 # Without fix: storing raw value as km wrong_km = raw_dist_metres # With fix: correct km correct_km = raw_dist_metres / 1000.0 assert wrong_km == 15000 # Would mean 15,000 km trip — clearly wrong assert correct_km == 15.0 class TestBUG03TripTimestamps: """BUG-03: Trip timestamps may be BCD format YYMMDDHHmmss or ISO string.""" def test_bcd_12_char_format(self): """220415103000 → 2022-04-15 10:30:00.""" result = _parse_trip_ts("220415103000") assert result == "2022-04-15 10:30:00" def test_bcd_14_char_format(self): """20220415103000 → 2022-04-15 10:30:00.""" result = _parse_trip_ts("20220415103000") assert result == "2022-04-15 10:30:00" def test_iso_string_passthrough(self): """ISO string passes through unchanged.""" result = _parse_trip_ts("2024-04-12 08:00:00") assert result == "2024-04-12 08:00:00" def test_none_returns_none(self): assert _parse_trip_ts(None) is None def test_garbage_returns_none(self): assert _parse_trip_ts("not-a-timestamp") is None def test_bcd_year_20xx(self): """24 prefix → 2024-xx-xx.""" result = _parse_trip_ts("240412080000") assert result is not None assert result.startswith("2024-04-12") class TestUnixToTs: """BUG-01: OBD event_time may be Unix epoch (seconds or milliseconds).""" def test_unix_seconds(self): result = unix_to_ts(1712908800) assert result is not None assert "2024" in result def test_unix_milliseconds(self): result = unix_to_ts(1712908800000) # ms — should be divided by 1000 assert result is not None assert "2024" in result def test_unix_seconds_matches_milliseconds(self): """Seconds and milliseconds of same moment produce same result.""" assert unix_to_ts(1712908800) == unix_to_ts(1712908800000) def test_none_returns_none(self): assert unix_to_ts(None) is None