refactor: make analytics _connect a context manager that closes

pixelmap/analytics.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
+import contextlib
 import os
 import sqlite3
 import threading
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Iterator
 
 import panel as pn
 import param
@@ -103,13 +104,16 @@ def _utc_now() -> str:
     return datetime.now(UTC).isoformat(timespec="seconds")
 
 
-def _connect(db_path: Path) -> sqlite3.Connection:
+@contextlib.contextmanager
+def _connect(db_path: Path) -> Iterator[sqlite3.Connection]:
+    """Open a connection that commits on success, rolls back on error, and always closes."""
     db_path.parent.mkdir(parents=True, exist_ok=True)
-    connection = sqlite3.connect(db_path, timeout=30)
-    connection.row_factory = sqlite3.Row
-    connection.execute("PRAGMA journal_mode=WAL;")
-    connection.execute("PRAGMA foreign_keys=ON;")
-    return connection
+    with contextlib.closing(sqlite3.connect(db_path, timeout=30)) as connection:
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA journal_mode=WAL;")
+        connection.execute("PRAGMA foreign_keys=ON;")
+        with connection:
+            yield connection
 
 
 def initialize_database(db_path: Path | None = None) -> Path:

tests/test_analytics.py

@@ -1,6 +1,10 @@
+import sqlite3
 from types import SimpleNamespace
 
+import pytest
+
 from pixelmap.analytics import (
+    _connect,
     build_request_context,
     build_visitor_key,
     get_total_visits,
@@ -99,3 +103,83 @@ def test_record_session_end_updates_last_seen(tmp_path):
     visitor_summary = get_visitor_summary("cookie:visitor-1", db_path=db_path)
     assert visitor_summary is not None
     assert visitor_summary["last_seen_at"] == "2026-03-19T18:20:00+00:00"
+
+
+_VISIT_REQUEST = SimpleNamespace(
+    headers={"User-Agent": "pytest-agent"},
+    remote_ip="198.51.100.12",
+    path="/",
+)
+_INSERT_SUMMARY = (
+    "INSERT INTO visitor_summary (visitor_key, ip_address, first_seen_at, last_seen_at, visit_count) "
+    "VALUES ('k', '198.51.100.12', 't', 't', 1)"
+)
+
+
+@pytest.fixture
+def tracked_connections(monkeypatch):
+    """Record every sqlite3 connection the analytics module opens."""
+    opened = []
+    real_connect = sqlite3.connect
+
+    def tracking_connect(*args, **kwargs):
+        connection = real_connect(*args, **kwargs)
+        opened.append(connection)
+        return connection
+
+    monkeypatch.setattr("pixelmap.analytics.sqlite3.connect", tracking_connect)
+    return opened
+
+
+def _assert_all_closed(connections):
+    assert connections
+    for connection in connections:
+        with pytest.raises(sqlite3.ProgrammingError):
+            connection.execute("SELECT 1")
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        pytest.param(lambda db: initialize_database(db), id="initialize_database"),
+        pytest.param(
+            lambda db: record_visit("visitor-1", build_request_context(_VISIT_REQUEST, "session-1"), db_path=db),
+            id="record_visit",
+        ),
+        pytest.param(
+            lambda db: record_session_end("visitor-1", "198.51.100.12", db_path=db),
+            id="record_session_end",
+        ),
+        pytest.param(lambda db: get_total_visits(db), id="get_total_visits"),
+        pytest.param(lambda db: get_unique_visitors(db), id="get_unique_visitors"),
+        pytest.param(lambda db: get_visitor_summary("cookie:visitor-1", db), id="get_visitor_summary"),
+    ],
+)
+def test_helpers_close_every_connection(tmp_path, tracked_connections, call):
+    # sqlite3's own connection context manager commits/rolls back but never
+    # closes, so each helper must close the connections it opens.
+    call(tmp_path / "visitor_analytics.sqlite3")
+
+    _assert_all_closed(tracked_connections)
+
+
+def test_connect_commits_on_success(tmp_path, tracked_connections):
+    db_path = initialize_database(tmp_path / "visitor_analytics.sqlite3")
+
+    with _connect(db_path) as connection:
+        connection.execute(_INSERT_SUMMARY)
+
+    assert get_visitor_summary("k", db_path) is not None
+    _assert_all_closed(tracked_connections)
+
+
+def test_connect_rolls_back_and_closes_on_error(tmp_path, tracked_connections):
+    db_path = initialize_database(tmp_path / "visitor_analytics.sqlite3")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        with _connect(db_path) as connection:
+            connection.execute(_INSERT_SUMMARY)
+            raise RuntimeError("boom")
+
+    assert get_visitor_summary("k", db_path) is None
+    _assert_all_closed(tracked_connections)