from __future__ import annotations import asyncio from datetime import datetime, timezone import sqlite3 import pytest from src.tracking_service import ( TrackingService, get_current_search_id, set_current_search_id, ) def test_init_creates_tables(tmp_path) -> None: db_path = tmp_path / "tracking.db" tracker = TrackingService(db_path) with sqlite3.connect(str(db_path)) as conn: tables = { row[0] for row in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ).fetchall() } tracker.close() assert "search_logs" in tables assert "api_call_logs" in tables def test_create_and_complete_search(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="s1", environment="beta", source="api", search_type="period", commune_1="Paris", commune_1_slug="paris-75", latitude=48.85, longitude=2.35, start_date="2026-03-01", end_date="2026-03-02", ) tracker.complete_search(search_id="s1", status="success", duration_ms=123) detail = tracker.get_search_detail("s1") tracker.close() assert detail is not None assert detail["search"]["status"] == "success" assert detail["search"]["duration_ms"] == 123 def test_log_api_call(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="s2", environment="beta", source="seo", search_type="period", ) tracker.log_api_call( search_id="s2", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary="lat=48.85,lon=2.35,2026-03-01/2026-03-02", cache_key="weather:48.85:2.35:2026-03-01:2026-03-02", cache_status="miss", status_code=200, duration_ms=87, success=True, error_message=None, ) detail = tracker.get_search_detail("s2") tracker.close() assert detail is not None assert len(detail["api_calls"]) == 1 assert detail["api_calls"][0]["cache_status"] == "miss" def test_search_error_does_not_raise(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.close() tracker.create_search( search_id="closed", environment="dev", source="api", search_type="period", ) def test_api_call_error_does_not_raise(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.close() tracker.log_api_call( search_id="closed", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key=None, cache_status="hit", status_code=None, duration_ms=0, success=True, error_message=None, ) def test_list_searches_pagination(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") for idx in range(15): tracker.create_search( search_id=f"s{idx}", environment="beta", source="api", search_type="period", commune_1_slug=f"c{idx}", ) page = tracker.list_searches(limit=10, offset=0) tracker.close() assert len(page["searches"]) == 10 assert page["total"] == 15 def test_list_searches_filter_status(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="ok1", environment="beta", source="api", search_type="period", ) tracker.create_search( search_id="err1", environment="beta", source="api", search_type="period", ) tracker.complete_search(search_id="ok1", status="success", duration_ms=5) tracker.complete_search(search_id="err1", status="error", duration_ms=5) filtered = tracker.list_searches(status="error") tracker.close() assert len(filtered["searches"]) == 1 assert filtered["searches"][0]["id"] == "err1" def test_list_searches_filter_commune(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="gap", environment="beta", source="seo", search_type="period", commune_1="Gap", commune_1_slug="gap-05", ) tracker.create_search( search_id="lyon", environment="beta", source="seo", search_type="period", commune_1="Lyon", commune_1_slug="lyon-69", ) filtered = tracker.list_searches(commune="gap") tracker.close() assert len(filtered["searches"]) == 1 assert filtered["searches"][0]["commune_1_slug"] == "gap-05" def test_get_search_detail(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="detail", environment="beta", source="seo", search_type="period", ) tracker.log_api_call( search_id="detail", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="normals:44.56:6.08", cache_status="hit", status_code=None, duration_ms=0, success=True, error_message=None, ) detail = tracker.get_search_detail("detail") tracker.close() assert detail is not None assert detail["search"]["id"] == "detail" assert len(detail["api_calls"]) == 1 def test_get_dashboard(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") today = datetime.now(timezone.utc).date().isoformat() tracker.create_search( search_id="d1", environment="beta", source="api", search_type="period", commune_1="Paris", commune_1_slug="paris-75", ) tracker.complete_search(search_id="d1", status="success", duration_ms=80) tracker.log_api_call( search_id="d1", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="w:1", cache_status="hit", status_code=None, duration_ms=0, success=True, error_message=None, ) tracker.create_search( search_id="d2", environment="beta", source="api", search_type="period", commune_1="Lyon", commune_1_slug="lyon-69", ) tracker.complete_search( search_id="d2", status="error", duration_ms=120, error_message="boom", ) tracker.log_api_call( search_id="d2", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="w:2", cache_status="miss", status_code=502, duration_ms=12, success=False, error_message="boom", ) dashboard = tracker.get_dashboard(today) tracker.close() assert dashboard["date"] == today assert dashboard["searches_today"] == 2 assert dashboard["api_calls_today"]["total"] == 1 assert dashboard["api_calls_today"]["by_service"]["weather"] == 1 assert dashboard["cache_hit_ratio"] == 0.5 assert dashboard["error_rate"] == 0.5 def test_count_api_calls_excludes_cache_hits(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="count-1", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="count-1", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="k1", cache_status="hit", status_code=None, duration_ms=0, success=True, error_message=None, ) tracker.log_api_call( search_id="count-1", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="k2", cache_status="miss", status_code=200, duration_ms=10, success=True, error_message=None, ) assert tracker._count_api_calls("count-1") == 1 tracker.close() def test_log_api_call_with_service(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="service-1", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="service-1", service="normals", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key=None, cache_status="miss", status_code=200, duration_ms=20, success=True, error_message=None, ) detail = tracker.get_search_detail("service-1") tracker.close() assert detail is not None assert detail["api_calls"][0]["service"] == "normals" def test_dashboard_api_calls_by_service(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") today = datetime.now(timezone.utc).date().isoformat() tracker.create_search( search_id="svc-1", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="svc-1", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="a", cache_status="miss", status_code=200, duration_ms=5, success=True, error_message=None, ) tracker.log_api_call( search_id="svc-1", service="normals", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key=None, cache_status="miss", status_code=200, duration_ms=7, success=True, error_message=None, ) tracker.log_api_call( search_id="svc-1", service="communes", provider="geo-api-gouv", endpoint="https://geo.api.gouv.fr/communes", params_summary=None, cache_key="b", cache_status="miss", status_code=200, duration_ms=9, success=True, error_message=None, ) dashboard = tracker.get_dashboard(today) tracker.close() assert dashboard["api_calls_today"]["total"] == 3 assert dashboard["api_calls_today"]["by_service"]["weather"] == 1 assert dashboard["api_calls_today"]["by_service"]["normals"] == 1 assert dashboard["api_calls_today"]["by_service"]["communes"] == 1 def test_dashboard_cache_hit_ratio_new_formula(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") today = datetime.now(timezone.utc).date().isoformat() tracker.create_search( search_id="r1", environment="beta", source="api", search_type="period", ) tracker.complete_search( search_id="r1", status="success", duration_ms=10, total_api_calls=0, ) tracker.create_search( search_id="r2", environment="beta", source="api", search_type="period", ) tracker.complete_search( search_id="r2", status="success", duration_ms=10, total_api_calls=2, ) tracker.create_search( search_id="r3", environment="beta", source="api", search_type="period", ) tracker.complete_search( search_id="r3", status="error", duration_ms=10, total_api_calls=0, error_message="boom", ) dashboard = tracker.get_dashboard(today) tracker.close() assert dashboard["cache_hit_ratio"] == 0.33 def test_migration_adds_service_column(tmp_path) -> None: db_path = tmp_path / "tracking.db" with sqlite3.connect(str(db_path)) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS search_logs ( id TEXT PRIMARY KEY, created_at TEXT NOT NULL, environment TEXT NOT NULL, source TEXT NOT NULL, search_type TEXT NOT NULL, commune_1 TEXT, commune_1_slug TEXT, commune_2 TEXT, commune_2_slug TEXT, latitude REAL, longitude REAL, start_date TEXT, end_date TEXT, status TEXT NOT NULL DEFAULT 'pending', total_api_calls INTEGER NOT NULL DEFAULT 0, duration_ms INTEGER, error_message TEXT ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS api_call_logs ( id TEXT PRIMARY KEY, search_id TEXT NOT NULL, created_at TEXT NOT NULL, provider TEXT NOT NULL, endpoint TEXT NOT NULL, params_summary TEXT, cache_key TEXT, cache_status TEXT NOT NULL, status_code INTEGER, duration_ms INTEGER NOT NULL, success INTEGER NOT NULL DEFAULT 1, error_message TEXT, FOREIGN KEY (search_id) REFERENCES search_logs(id) ) """ ) tracker = TrackingService(db_path) with sqlite3.connect(str(db_path)) as conn: columns = [ row[1] for row in conn.execute("PRAGMA table_info('api_call_logs')").fetchall() ] tracker.close() assert "service" in columns def test_log_api_call_with_null_search_id(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.log_api_call( search_id=None, service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary="lat=48.85,lon=2.35,2026-03-01/2026-03-02", cache_key="weather:48.85:2.35:2026-03-01:2026-03-02", cache_status="miss", status_code=200, duration_ms=42, success=True, error_message=None, ) row = tracker.conn.execute( "SELECT search_id FROM api_call_logs ORDER BY created_at DESC LIMIT 1" ).fetchone() tracker.close() assert row is not None assert row["search_id"] is None def test_count_api_calls_excludes_null_search_id(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") tracker.create_search( search_id="abc", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="abc", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="k-abc", cache_status="miss", status_code=200, duration_ms=11, success=True, error_message=None, ) tracker.log_api_call( search_id=None, service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="k-null", cache_status="miss", status_code=200, duration_ms=12, success=True, error_message=None, ) assert tracker._count_api_calls("abc") == 1 tracker.close() def test_dashboard_counts_standalone_calls(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") today = datetime.now(timezone.utc).date().isoformat() tracker.create_search( search_id="db-standalone", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="db-standalone", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="k1", cache_status="miss", status_code=200, duration_ms=10, success=True, error_message=None, ) tracker.log_api_call( search_id=None, service="normals", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key=None, cache_status="miss", status_code=200, duration_ms=10, success=True, error_message=None, ) dashboard = tracker.get_dashboard(today) tracker.close() assert dashboard["api_calls_today"]["total"] == 2 def test_dashboard_attributed_vs_standalone(tmp_path) -> None: tracker = TrackingService(tmp_path / "tracking.db") today = datetime.now(timezone.utc).date().isoformat() tracker.create_search( search_id="db-split", environment="beta", source="api", search_type="period", ) tracker.log_api_call( search_id="db-split", service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key="a", cache_status="miss", status_code=200, duration_ms=8, success=True, error_message=None, ) tracker.log_api_call( search_id=None, service="communes", provider="geo-api-gouv", endpoint="https://geo.api.gouv.fr/communes", params_summary=None, cache_key="b", cache_status="miss", status_code=200, duration_ms=8, success=True, error_message=None, ) dashboard = tracker.get_dashboard(today) tracker.close() assert dashboard["api_calls_today"]["attributed"] == 1 assert dashboard["api_calls_today"]["standalone"] == 1 def test_migration_search_id_nullable(tmp_path) -> None: db_path = tmp_path / "tracking.db" with sqlite3.connect(str(db_path)) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS search_logs ( id TEXT PRIMARY KEY, created_at TEXT NOT NULL, environment TEXT NOT NULL, source TEXT NOT NULL, search_type TEXT NOT NULL, commune_1 TEXT, commune_1_slug TEXT, commune_2 TEXT, commune_2_slug TEXT, latitude REAL, longitude REAL, start_date TEXT, end_date TEXT, status TEXT NOT NULL DEFAULT 'pending', total_api_calls INTEGER NOT NULL DEFAULT 0, duration_ms INTEGER, error_message TEXT ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS api_call_logs ( id TEXT PRIMARY KEY, search_id TEXT NOT NULL, created_at TEXT NOT NULL, service TEXT NOT NULL DEFAULT 'weather', provider TEXT NOT NULL, endpoint TEXT NOT NULL, params_summary TEXT, cache_key TEXT, cache_status TEXT NOT NULL, status_code INTEGER, duration_ms INTEGER NOT NULL, success INTEGER NOT NULL DEFAULT 1, error_message TEXT, FOREIGN KEY (search_id) REFERENCES search_logs(id) ) """ ) tracker = TrackingService(db_path) tracker.log_api_call( search_id=None, service="weather", provider="open-meteo", endpoint="https://archive-api.open-meteo.com/v1/archive", params_summary=None, cache_key=None, cache_status="miss", status_code=200, duration_ms=5, success=True, error_message=None, ) with sqlite3.connect(str(db_path)) as conn: row = conn.execute( "SELECT search_id FROM api_call_logs ORDER BY created_at DESC LIMIT 1" ).fetchone() search_col = next( col for col in conn.execute("PRAGMA table_info('api_call_logs')") if col[1] == "search_id" ) tracker.close() assert row is not None assert row[0] is None assert search_col[3] == 0 @pytest.mark.asyncio async def test_contextvars_isolation() -> None: async def worker(search_id: str) -> str | None: set_current_search_id(search_id) await asyncio.sleep(0) return get_current_search_id() first, second = await asyncio.gather(worker("a"), worker("b")) set_current_search_id(None) assert first == "a" assert second == "b"