99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
|
"""Unit tests for PostgresEventStore using a mocked psycopg2 pool."""
|
||
|
|
from __future__ import annotations
|
||
|
|
import json
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
# Patch psycopg2 before importing the module under test
|
||
|
|
import sys
|
||
|
|
mock_psycopg2 = MagicMock()
|
||
|
|
mock_psycopg2.extras = MagicMock()
|
||
|
|
sys.modules.setdefault("psycopg2", mock_psycopg2)
|
||
|
|
sys.modules.setdefault("psycopg2.extras", mock_psycopg2.extras)
|
||
|
|
sys.modules.setdefault("psycopg2.pool", MagicMock())
|
||
|
|
|
||
|
|
from app.infrastructure.perception.base_event_store import BaseEventStore
|
||
|
|
|
||
|
|
|
||
|
|
SAMPLE_ROW = {
|
||
|
|
"id": "pg-001",
|
||
|
|
"source": "国标委",
|
||
|
|
"source_label": "国家标准化管理委员会",
|
||
|
|
"standard_code": "GB 18384-2025",
|
||
|
|
"title": "电动汽车安全要求",
|
||
|
|
"summary": "新增要求",
|
||
|
|
"full_text_url": "https://openstd.samr.gov.cn",
|
||
|
|
"status": "enacted",
|
||
|
|
"impact_level": "high",
|
||
|
|
"published_at": "2025-11-15",
|
||
|
|
"effective_at": "2026-07-01",
|
||
|
|
"category": "电动汽车安全",
|
||
|
|
"tags": ["电池安全"],
|
||
|
|
"obligations": None,
|
||
|
|
"deadlines": None,
|
||
|
|
"scope": None,
|
||
|
|
"penalties": None,
|
||
|
|
"content_hash": "abc123",
|
||
|
|
"previous_hash": None,
|
||
|
|
"change_summary": None,
|
||
|
|
"changed_sections": None,
|
||
|
|
"affected_docs": None,
|
||
|
|
"crawled_at": "2026-06-05T10:00:00+00:00",
|
||
|
|
"processed_at": None,
|
||
|
|
"raw_storage_key": None,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _make_store_with_pool(mock_pool):
|
||
|
|
with patch("psycopg2.pool.ThreadedConnectionPool", return_value=mock_pool):
|
||
|
|
with patch(
|
||
|
|
"app.infrastructure.perception.postgres_event_store.PostgresEventStore._ensure_schema"
|
||
|
|
):
|
||
|
|
from app.infrastructure.perception.postgres_event_store import PostgresEventStore
|
||
|
|
return PostgresEventStore()
|
||
|
|
|
||
|
|
|
||
|
|
def _cursor_returning(rows):
|
||
|
|
cursor = MagicMock()
|
||
|
|
cursor.__enter__ = lambda s: s
|
||
|
|
cursor.__exit__ = MagicMock(return_value=False)
|
||
|
|
cursor.fetchall.return_value = rows
|
||
|
|
cursor.fetchone.return_value = rows[0] if rows else None
|
||
|
|
return cursor
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_base_event_store():
|
||
|
|
mock_pool = MagicMock()
|
||
|
|
store = _make_store_with_pool(mock_pool)
|
||
|
|
assert isinstance(store, BaseEventStore)
|
||
|
|
|
||
|
|
|
||
|
|
def test_filter_returns_list():
|
||
|
|
mock_pool = MagicMock()
|
||
|
|
conn = MagicMock()
|
||
|
|
conn.__enter__ = lambda s: s
|
||
|
|
conn.__exit__ = MagicMock(return_value=False)
|
||
|
|
cursor = _cursor_returning([SAMPLE_ROW])
|
||
|
|
conn.cursor.return_value = cursor
|
||
|
|
mock_pool.getconn.return_value = conn
|
||
|
|
store = _make_store_with_pool(mock_pool)
|
||
|
|
result = store.filter(limit=10)
|
||
|
|
assert isinstance(result, list)
|
||
|
|
|
||
|
|
|
||
|
|
def test_stats_returns_correct_keys():
|
||
|
|
mock_pool = MagicMock()
|
||
|
|
conn = MagicMock()
|
||
|
|
conn.__enter__ = lambda s: s
|
||
|
|
conn.__exit__ = MagicMock(return_value=False)
|
||
|
|
cursor = MagicMock()
|
||
|
|
cursor.__enter__ = lambda s: s
|
||
|
|
cursor.__exit__ = MagicMock(return_value=False)
|
||
|
|
cursor.fetchone.return_value = {"count": 5}
|
||
|
|
conn.cursor.return_value = cursor
|
||
|
|
mock_pool.getconn.return_value = conn
|
||
|
|
store = _make_store_with_pool(mock_pool)
|
||
|
|
stats = store.stats()
|
||
|
|
for key in ("total", "high_impact", "medium_impact", "recent_90d"):
|
||
|
|
assert key in stats
|