fix(wndb): refresh closed project connections

tests/unit/test_wndb_connection.py

@@ -0,0 +1,78 @@
+import pytest
+
+from app.native.wndb import connection
+from app.native.wndb import database
+from app.native.wndb import project
+
+
+class _FakeCursor:
+    def __init__(self, rows):
+        self.rows = rows
+        self.executed = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, sql):
+        self.executed.append(sql)
+
+    def fetchall(self):
+        return self.rows
+
+
+class _FakeConnection:
+    def __init__(self, rows=None, *, closed=False):
+        self.rows = list(rows or [])
+        self.closed = closed
+        self.close_calls = 0
+
+    def cursor(self, row_factory=None):
+        if self.closed:
+            raise RuntimeError("the connection is closed")
+        return _FakeCursor(self.rows)
+
+    def close(self):
+        self.close_calls += 1
+        self.closed = True
+
+
+@pytest.fixture(autouse=True)
+def clear_native_connections():
+    connection.g_conn_dict.clear()
+    connection._project_locks.clear()
+    yield
+    connection.g_conn_dict.clear()
+    connection._project_locks.clear()
+
+
+def test_is_project_open_drops_closed_cached_connection():
+    connection.g_conn_dict["fengyang"] = _FakeConnection(closed=True)
+
+    assert project.is_project_open("fengyang") is False
+    assert "fengyang" not in connection.g_conn_dict
+
+
+def test_read_all_reopens_closed_cached_connection(monkeypatch):
+    stale = _FakeConnection(closed=True)
+    fresh = _FakeConnection(rows=[{"key": "DURATION", "value": "01:00:00"}])
+    connection.g_conn_dict["fengyang"] = stale
+
+    opened = []
+
+    def fake_connect(*, conninfo, autocommit):
+        opened.append((conninfo, autocommit))
+        return fresh
+
+    monkeypatch.setattr(connection.pg, "connect", fake_connect)
+    monkeypatch.setattr(
+        connection, "get_pgconn_string", lambda db_name: f"dbname={db_name}"
+    )
+
+    rows = database.read_all("fengyang", "select * from times")
+
+    assert rows == [{"key": "DURATION", "value": "01:00:00"}]
+    assert opened == [("dbname=fengyang", True)]
+    assert connection.g_conn_dict["fengyang"] is fresh