65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""通知（お知らせ）读取链路测试：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
|