79 lines
2.0 KiB
Python
79 lines
2.0 KiB
Python
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
|