tests/test_tools_traffic.py

"""Traffic tools with mocked TomTom responses."""

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import pytest

from app.tools import traffic

TZ = ZoneInfo("Asia/Kolkata")


def make_tomtom_route(travel_s=1800, no_traffic_s=1500, length_m=12000, with_points=False):
    route = {
        "summary": {
            "travelTimeInSeconds": travel_s,
            "noTrafficTravelTimeInSeconds": no_traffic_s,
            "lengthInMeters": length_m,
        }
    }
    if with_points:
        route["legs"] = [
            {"points": [{"latitude": 26.85, "longitude": 80.95}, {"latitude": 26.86, "longitude": 80.96}]}
        ]
    return route


@pytest.fixture
def mock_tomtom(monkeypatch, fake_response, fake_keys):
    captured = {"params": []}

    def fake_get(url, params=None, timeout=None):
        captured["params"].append(params)
        return fake_response({"routes": [make_tomtom_route(with_points="summaryOnly" not in (params or {}).get("routeRepresentation", ""))]})

    monkeypatch.setattr(traffic.requests, "get", fake_get)
    return captured


def test_get_route_summarizes_and_includes_points(mock_tomtom):
    routes = traffic.get_route(26.85, 80.95, 26.84, 80.92)
    assert len(routes) == 1
    route = routes[0]
    assert route["duration_min"] == 30.0
    assert route["delay_min"] == 5.0
    assert route["distance_km"] == 12.0
    assert route["points"] == [[26.85, 80.95], [26.86, 80.96]]


def test_compare_departures_builds_curve(mock_tomtom):
    arrive_by = (datetime.now(TZ) + timedelta(hours=3)).replace(second=0, microsecond=0)
    result = traffic.compare_departures(
        26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
        window_min=45, step_min=15,
    )
    curve = result["curve"]
    assert len(curve) >= 3
    for entry in curve:
        assert entry["travel_min"] == 30.0
        assert "depart" in entry and "eta" in entry
        assert isinstance(entry["on_time"], bool)


def test_compare_departures_recommends_latest_safe_departure(mock_tomtom):
    arrive_by = (datetime.now(TZ) + timedelta(hours=3)).replace(second=0, microsecond=0)
    result = traffic.compare_departures(
        26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
        window_min=60, step_min=15,
    )
    recommended = result["recommended"]
    assert recommended is not None
    assert recommended["on_time"] is True
    # All trips take 30 min and the sweep ends 15 min before the deadline, so
    # the latest candidate is late-but-on-time; recommendation must have margin >= 5 if any exists
    safe = [entry for entry in result["curve"] if entry["margin_min"] >= 5]
    if safe:
        assert recommended["depart"] == safe[-1]["depart"]


def test_sweep_uses_depart_at_param(mock_tomtom):
    arrive_by = (datetime.now(TZ) + timedelta(hours=2)).replace(second=0, microsecond=0)
    traffic.compare_departures(26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
                               window_min=30, step_min=15)
    depart_params = [p.get("departAt") for p in mock_tomtom["params"] if p.get("departAt")]
    assert len(depart_params) >= 2  # multiple future departures were simulated


def test_sweep_survives_partial_rate_limit(monkeypatch, fake_response, fake_keys):
    """One 429 slot is skipped; the rest of the curve still comes back."""
    import requests as requests_lib

    from app import netutil

    monkeypatch.setattr(netutil.time, "sleep", lambda s: None)
    calls = {"count": 0}

    def fake_get(url, params=None, timeout=None):
        calls["count"] += 1
        if calls["count"] == 1:  # first request permanently rate-limited
            response = fake_response({}, status_code=429, text="Too Many Requests")
            return response
        return fake_response({"routes": [make_tomtom_route()]})

    monkeypatch.setattr(netutil.requests, "get", fake_get)
    arrive_by = (datetime.now(TZ) + timedelta(hours=3)).replace(second=0, microsecond=0)
    result = traffic.compare_departures(
        26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
        window_min=45, step_min=15,
    )
    # The 429 candidate eventually succeeded on retry OR was skipped —
    # either way we must still get a usable curve and a recommendation.
    assert result["curve"]
    assert result["recommended"] is not None


def test_sweep_total_rate_limit_raises_friendly_error(monkeypatch, fake_response, fake_keys):
    from app import netutil

    monkeypatch.setattr(netutil.time, "sleep", lambda s: None)
    monkeypatch.setattr(
        netutil.requests, "get",
        lambda *a, **k: fake_response({}, status_code=429, text="Too Many Requests"),
    )
    arrive_by = (datetime.now(TZ) + timedelta(hours=3)).replace(second=0, microsecond=0)
    with pytest.raises(RuntimeError) as excinfo:
        traffic.compare_departures(26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
                                   window_min=30, step_min=15)
    message = str(excinfo.value)
    assert "rate limit" in message.lower()
    assert "key=" not in message  # no API key leakage


def test_compare_departures_handles_all_late(monkeypatch, fake_response, fake_keys):
    # Every trip takes 10 hours -> nothing is on time -> least-late pick
    def fake_get(url, params=None, timeout=None):
        return fake_response({"routes": [make_tomtom_route(travel_s=36000, no_traffic_s=36000)]})

    monkeypatch.setattr(traffic.requests, "get", fake_get)
    arrive_by = (datetime.now(TZ) + timedelta(hours=2)).replace(second=0, microsecond=0)
    result = traffic.compare_departures(26.85, 80.95, 26.84, 80.92, arrive_by.isoformat(),
                                        window_min=30, step_min=15)
    recommended = result["recommended"]
    assert recommended is not None
    assert recommended["on_time"] is False
    # least late = earliest departure
    assert recommended["depart"] == result["curve"][0]["depart"]