File size: 5,902 Bytes
8b96826 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """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"]
|