tracksolid_timescale_grafana_prod/tests/unit/test_field_mapping.py

"""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