通知

tests/test_news.py

@@ -0,0 +1,64 @@
+"""通知（お知らせ）读取链路测试：redis_keys 契约 + /api/news 路由排序/截断/降级。"""
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+from app.api.routes.news import list_news
+from app.core.config import Settings
+from surugaya_common import redis_keys
+
+
+class FakeRedis:
+    """最小化异步 Redis 替身，仅实现 hgetall（值已 decode 为 str）。"""
+
+    def __init__(self, mapping: dict[str, str]):
+        self._mapping = mapping
+
+    async def hgetall(self, key: str) -> dict[str, str]:
+        return dict(self._mapping)
+
+
+def _container(redis_obj, app_env: str = "prod") -> SimpleNamespace:
+    settings = Settings(_env_file=None, app_env=app_env)
+    session_store = SimpleNamespace(_redis=redis_obj)
+    return SimpleNamespace(settings=settings, session_store=session_store)
+
+
+def _news_value(data_id: str, **extra) -> str:
+    return json.dumps({"data_id": data_id, "message": f"msg-{data_id}", **extra}, ensure_ascii=False)
+
+
+class TestNewsHashKey:
+    def test_literals(self):
+        assert redis_keys.NEWS_HASH == "jp_surugaya_news"
+        assert redis_keys.news_hash() == "jp_surugaya_news"
+        assert redis_keys.news_hash("0") == "jp_surugaya_news"
+        assert redis_keys.news_hash("1") == "jp_surugaya_news_dev"
+
+
+async def test_list_news_sorted_desc_by_data_id():
+    redis_obj = FakeRedis(
+        {"100": _news_value("100"), "300": _news_value("300", trade_code="S300"), "200": _news_value("200")}
+    )
+    resp = await list_news(limit=None, container=_container(redis_obj))
+    assert resp.success and resp.code == 0
+    assert [i.data_id for i in resp.data] == ["300", "200", "100"]
+    assert resp.data[0].trade_code == "S300"
+
+
+async def test_list_news_limit():
+    redis_obj = FakeRedis({str(i): _news_value(str(i)) for i in range(1, 11)})
+    resp = await list_news(limit=3, container=_container(redis_obj))
+    assert [i.data_id for i in resp.data] == ["10", "9", "8"]
+
+
+async def test_list_news_skips_bad_json():
+    redis_obj = FakeRedis({"1": _news_value("1"), "2": "not-json"})
+    resp = await list_news(limit=None, container=_container(redis_obj))
+    assert [i.data_id for i in resp.data] == ["1"]
+
+
+async def test_list_news_redis_not_configured():
+    resp = await list_news(limit=None, container=_container(None))
+    assert not resp.success and resp.code == 500 and resp.data is None