jiang
fa188af0b1
Reorganize WNDB by responsibility and remove legacy scheme endpoints.\n\nRoute analysis and time-series access through project pools, preserve transactional realtime replacement, and refresh GIS materialized views after writes.\n\nAdd database architecture documentation, live pooling coverage, API contract updates, and executable container verification.\n\nBREAKING CHANGE: legacy scheme APIs and flat app.native.wndb module imports are removed.
55 lines
1.3 KiB
Python
55 lines
1.3 KiB
Python
import asyncio
|
|
from uuid import uuid4
|
|
|
|
from app.infra.db.postgresql.analysis import AnalysisRepository
|
|
|
|
|
|
class _FakeCursor:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def execute(self, query, params):
|
|
self.calls.append((str(query), params))
|
|
|
|
async def fetchall(self):
|
|
return []
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self):
|
|
self.cursor_instance = _FakeCursor()
|
|
|
|
def cursor(self):
|
|
return self.cursor_instance
|
|
|
|
|
|
def test_list_results_uses_typed_comparison_when_result_type_is_present():
|
|
conn = _FakeConnection()
|
|
run_id = uuid4()
|
|
|
|
asyncio.run(
|
|
AnalysisRepository.list_results(conn, run_id, "leakage_identification")
|
|
)
|
|
|
|
query, params = conn.cursor_instance.calls[0]
|
|
assert "result_type = %s" in query
|
|
assert "%s IS NULL" not in query
|
|
assert params == (run_id, "leakage_identification")
|
|
|
|
|
|
def test_list_results_omits_result_type_filter_when_not_requested():
|
|
conn = _FakeConnection()
|
|
run_id = uuid4()
|
|
|
|
asyncio.run(AnalysisRepository.list_results(conn, run_id))
|
|
|
|
query, params = conn.cursor_instance.calls[0]
|
|
assert "result_type = %s" not in query
|
|
assert params == (run_id,)
|