73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
"""Tests for the notification API routes.
|
|
|
|
Follows the same direct-call convention as backend/tests/mcp/test_mcp_status.py:
|
|
patch the bootstrap singleton getter where the route module imports it, then
|
|
call the async route function directly with asyncio.run rather than standing
|
|
up a FastAPI TestClient — this repo has no existing TestClient harness, and
|
|
these route handlers are thin enough that exercising them directly tests the
|
|
same logic without inventing a new testing convention for two pass-through
|
|
endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from unittest.mock import patch
|
|
|
|
from app.domain.auth.models import UserClaims, UserRole
|
|
from app.infrastructure.perception.mock_notification_store import MockNotificationStore
|
|
|
|
|
|
def _user(user_id: str) -> UserClaims:
|
|
return UserClaims(user_id=user_id, username=user_id, role=UserRole.READONLY)
|
|
|
|
|
|
def _list_notifications(store, user_id: str, limit: int = 20) -> dict:
|
|
from app.api.routes.perception import list_notifications
|
|
|
|
with patch("app.api.routes.perception.get_notification_store", return_value=store):
|
|
return asyncio.run(list_notifications(limit=limit, current_user=_user(user_id)))
|
|
|
|
|
|
def _mark_read(store, user_id: str) -> dict:
|
|
from app.api.routes.perception import mark_notifications_read
|
|
|
|
with patch("app.api.routes.perception.get_notification_store", return_value=store):
|
|
return asyncio.run(mark_notifications_read(current_user=_user(user_id)))
|
|
|
|
|
|
def test_a_fresh_user_sees_the_notification_as_unread():
|
|
"""GET .../notifications reports the caller's own unread_count."""
|
|
store = MockNotificationStore()
|
|
store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None)
|
|
|
|
result = _list_notifications(store, "user-a")
|
|
|
|
assert result["unread_count"] == 1
|
|
assert len(result["items"]) == 1
|
|
assert result["items"][0]["read"] is False
|
|
|
|
|
|
def test_mark_read_zeroes_a_subsequent_get():
|
|
"""POST .../read must clear unread_count for the next GET by the same user."""
|
|
store = MockNotificationStore()
|
|
store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None)
|
|
|
|
marked = _mark_read(store, "user-a")
|
|
result = _list_notifications(store, "user-a")
|
|
|
|
assert marked == {"marked": 1}
|
|
assert result["unread_count"] == 0
|
|
assert result["items"][0]["read"] is True
|
|
|
|
|
|
def test_mark_read_for_one_user_does_not_affect_another():
|
|
"""Read state is per-user — the whole point of the read-receipt design."""
|
|
store = MockNotificationStore()
|
|
store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None)
|
|
|
|
_mark_read(store, "user-a")
|
|
result_b = _list_notifications(store, "user-b")
|
|
|
|
assert result_b["unread_count"] == 1
|