test: add tests for SQLite static data cache layer

backend/tests/test_gtfs_db.py

@@ -1,17 +1,15 @@
+"""Tests for the SQLite static data cache layer (services/gtfs_db.py)."""
+
+import json
 import tempfile
 
 import pytest
 
-from services.gtfs_db import (
-    clear_cache,
-    load_cached_static,
-    save_cached_static,
-)
+from services.gtfs_db import save_cached_static, load_cached_static, clear_cache
 
 
 @pytest.fixture
 def db_path():
-    """Yield a temporary SQLite database path and clean up after the test."""
     with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
         path = f.name
     yield path
@@ -20,120 +18,169 @@ def db_path():
     os.unlink(path)
 
 
+SAMPLE_ROUTES = {
+    "1": {"route_id": "1", "route_short_name": "101", "route_color": "#FF0000"},
+    "2": {"route_id": "2", "route_short_name": "202", "route_color": "#00FF00"},
+}
+
+SAMPLE_STOPS = {
+    "A": {
+        "stop_id": "A",
+        "stop_name": "Station A",
+        "stop_lat": 43.45,
+        "stop_lon": -80.49,
+    },
+    "B": {
+        "stop_id": "B",
+        "stop_name": "Station B",
+        "stop_lat": 43.46,
+        "stop_lon": -80.50,
+    },
+}
+
+SAMPLE_STOP_TIMES = {
+    "trip-1": [
+        {"stop_id": "A", "stop_sequence": 1, "arrival_time": "08:00:00", "departure_time": "08:05:00"},
+    ],
+}
+
+SAMPLE_TRIP_ROUTES = {
+    "trip-1": "1",
+}
+
+LAST_MODIFIED = "Mon, 01 Jun 2026 12:00:00 GMT"
+ETAG = '"abc123"'
+
+
 @pytest.mark.asyncio
-async def test_save_and_load_round_trip(db_path: str) -> None:
+async def test_save_and_load(db_path):
+    """Save data, then load it. Verify all fields match."""
     feed_url = "https://example.com/gtfs.zip"
-    routes = {"1": {"route_id": "1", "route_short_name": "1", "route_color": "#FF0000"}}
-    stops = {
-        "S1": {
-            "stop_id": "S1",
-            "stop_name": "Main St",
-            "stop_lat": 43.0,
-            "stop_lon": -80.0,
-        },
-    }
-    last_modified = "Mon, 01 Jun 2026 12:00:00 GMT"
 
     saved = await save_cached_static(
         feed_url,
-        routes,
-        stops,
-        last_modified=last_modified,
+        SAMPLE_ROUTES,
+        SAMPLE_STOPS,
+        stop_times=SAMPLE_STOP_TIMES,
+        trip_routes=SAMPLE_TRIP_ROUTES,
+        last_modified=LAST_MODIFIED,
+        etag=ETAG,
         db_path=db_path,
     )
     assert saved is True
 
-    loaded = await load_cached_static(feed_url, db_path=db_path)
-    assert loaded is not None
-    assert loaded["routes"] == routes
-    assert loaded["stops"] == stops
-    assert loaded["last_modified"] == last_modified
-    assert loaded["etag"] is None
+    result = await load_cached_static(feed_url, db_path=db_path)
+    assert result is not None
+
+    assert result["routes"] == SAMPLE_ROUTES
+    assert result["stops"] == SAMPLE_STOPS
+    assert result["stop_times"] == SAMPLE_STOP_TIMES
+    assert result["trip_routes"] == SAMPLE_TRIP_ROUTES
+    assert result["last_modified"] == LAST_MODIFIED
+    assert result["etag"] == ETAG
 
 
 @pytest.mark.asyncio
-async def test_cache_miss_returns_none(db_path: str) -> None:
-    loaded = await load_cached_static(
-        "https://example.com/unknown.zip", db_path=db_path
-    )
-    assert loaded is None
+async def test_load_missing_url(db_path):
+    """Load a feed URL that was never saved. Assert result is None."""
+    result = await load_cached_static("https://example.com/never-saved.zip", db_path=db_path)
+    assert result is None
 
 
 @pytest.mark.asyncio
-async def test_upsert_replaces_existing_entry(db_path: str) -> None:
+async def test_save_updates_existing(db_path):
+    """Save data with URL X, save again with same URL but different data, load
+    it back. Verify it returns the updated data, not the old data."""
     feed_url = "https://example.com/gtfs.zip"
-    routes_v1 = {"1": {"route_id": "1"}}
-    stops_v1: dict = {}
 
+    # Save initial data.
     await save_cached_static(
-        feed_url, routes_v1, stops_v1, last_modified="old", db_path=db_path
+        feed_url,
+        {"r1": {"route_id": "r1", "route_short_name": "OLD"}},
+        {"s1": {"stop_id": "s1", "stop_name": "Old Stop"}},
+        db_path=db_path,
     )
 
-    routes_v2 = {"2": {"route_id": "2"}}
+    # Save updated data with the same URL.
     await save_cached_static(
-        feed_url, routes_v2, stops_v1, last_modified="new", db_path=db_path
+        feed_url,
+        {"r2": {"route_id": "r2", "route_short_name": "NEW"}},
+        {"s2": {"stop_id": "s2", "stop_name": "New Stop"}},
+        db_path=db_path,
     )
 
-    loaded = await load_cached_static(feed_url, db_path=db_path)
-    assert loaded is not None
-    assert loaded["last_modified"] == "new"
-    assert loaded["routes"] == routes_v2
+    result = await load_cached_static(feed_url, db_path=db_path)
+    assert result is not None
+    assert result["routes"] == {"r2": {"route_id": "r2", "route_short_name": "NEW"}}
+    assert result["stops"] == {"s2": {"stop_id": "s2", "stop_name": "New Stop"}}
 
 
 @pytest.mark.asyncio
-async def test_clear_single_url(db_path: str) -> None:
-    await save_cached_static(
-        "https://example.com/a.zip",
-        {"a": {}},
-        {},
-        db_path=db_path,
-    )
-    await save_cached_static(
-        "https://example.com/b.zip",
-        {"b": {}},
-        {},
-        db_path=db_path,
-    )
+async def test_save_with_empty_dicts(db_path):
+    """Save with empty routes={} and stops={}. Load and verify they come back
+    as empty dicts."""
+    feed_url = "https://example.com/gtfs.zip"
 
-    await clear_cache("https://example.com/a.zip", db_path=db_path)
+    await save_cached_static(feed_url, {}, {}, db_path=db_path)
 
-    assert (
-        await load_cached_static("https://example.com/a.zip", db_path=db_path) is None
-    )
-    assert (
-        await load_cached_static("https://example.com/b.zip", db_path=db_path)
-        is not None
-    )
+    result = await load_cached_static(feed_url, db_path=db_path)
+    assert result is not None
+    assert result["routes"] == {}
+    assert result["stops"] == {}
 
 
 @pytest.mark.asyncio
-async def test_clear_all(db_path: str) -> None:
-    await save_cached_static(
-        "https://example.com/a.zip", {"a": {}}, {}, db_path=db_path
-    )
-    await save_cached_static(
-        "https://example.com/b.zip", {"b": {}}, {}, db_path=db_path
-    )
+async def test_save_without_optional_fields(db_path):
+    """Save with no stop_times, trip_routes, last_modified, or etag. Load and
+    verify stop_times and trip_routes are empty dicts {}, last_modified and
+    etag are None."""
+    feed_url = "https://example.com/gtfs.zip"
 
-    await clear_cache(db_path=db_path)
+    await save_cached_static(feed_url, SAMPLE_ROUTES, SAMPLE_STOPS, db_path=db_path)
 
-    assert (
-        await load_cached_static("https://example.com/a.zip", db_path=db_path) is None
-    )
-    assert (
-        await load_cached_static("https://example.com/b.zip", db_path=db_path) is None
-    )
+    result = await load_cached_static(feed_url, db_path=db_path)
+    assert result is not None
+    assert result["routes"] == SAMPLE_ROUTES
+    assert result["stops"] == SAMPLE_STOPS
+    assert result["stop_times"] == {}
+    assert result["trip_routes"] == {}
+    assert result["last_modified"] is None
+    assert result["etag"] is None
 
 
 @pytest.mark.asyncio
-async def test_etag_caching(db_path: str) -> None:
-    feed_url = "https://example.com/gtfs.zip"
-    routes = {"1": {"route_id": "1"}}
-    stops: dict = {}
-    etag = '"abc123"'
+async def test_clear_cache_all(db_path):
+    """Save two URLs, call clear_cache() with no feed_url argument. Load both
+    — both should return None."""
+    await save_cached_static("url1", {"r": {"id": "r"}}, {"s": {"id": "s"}}, db_path=db_path)
+    await save_cached_static("url2", {"r": {"id": "r"}}, {"s": {"id": "s"}}, db_path=db_path)
+
+    cleared = await clear_cache(db_path=db_path)
+    assert cleared is True
 
-    await save_cached_static(feed_url, routes, stops, etag=etag, db_path=db_path)
-    loaded = await load_cached_static(feed_url, db_path=db_path)
+    assert await load_cached_static("url1", db_path=db_path) is None
+    assert await load_cached_static("url2", db_path=db_path) is None
 
-    assert loaded is not None
-    assert loaded["etag"] == etag
+
+@pytest.mark.asyncio
+async def test_clear_cache_single(db_path):
+    """Save two URLs, call clear_cache(feed_url="url1"). "url1" should return
+    None, "url2" should still return data."""
+    await save_cached_static("url1", {"r": {"id": "r"}}, {"s": {"id": "s"}}, db_path=db_path)
+    await save_cached_static("url2", {"r": {"id": "r"}}, {"s": {"id": "s"}}, db_path=db_path)
+
+    cleared = await clear_cache(feed_url="url1", db_path=db_path)
+    assert cleared is True
+
+    assert await load_cached_static("url1", db_path=db_path) is None
+    assert await load_cached_static("url2", db_path=db_path) is not None
+
+
+@pytest.mark.asyncio
+async def test_load_nonexistent_db():
+    """Call load_cached_static with a path in a directory that doesn't exist.
+    The function should return None gracefully (catches exceptions internally)."""
+    result = await load_cached_static(
+        "url", db_path="/nonexistent/dir/cache.db"
+    )
+    assert result is None