166 lines
5.6 KiB
Python
166 lines
5.6 KiB
Python
from __future__ import annotations
|
||
|
||
import importlib
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
def _set_min_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
# 让 Settings 可构造（引擎不会在单测中真正连接 DB/Redis）
|
||
monkeypatch.setenv(
|
||
"DATABASE_URL",
|
||
"mysql+aiomysql://u:p@127.0.0.1:3306/mindfulness_dev_test?charset=utf8mb4",
|
||
)
|
||
monkeypatch.setenv("REDIS_URL", "redis://127.0.0.1:6379/0")
|
||
monkeypatch.setenv("CELERY_BROKER_URL", "redis://127.0.0.1:6379/0")
|
||
|
||
|
||
def _user_profile_dict() -> dict[str, Any]:
|
||
return {
|
||
"profile_version": "v1.2",
|
||
"profile_source": "questionnaire",
|
||
"profile_generated_at": "2026-02-02T12:00:00Z",
|
||
"profile_confidence": 1.0,
|
||
"profile_answered": {"stage": True, "emotion": False, "context": False, "need": False},
|
||
"stage": {"unknown": 1},
|
||
"emotion_score": None,
|
||
"context": {},
|
||
"need": {},
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def client(monkeypatch: pytest.MonkeyPatch):
|
||
_set_min_env(monkeypatch)
|
||
|
||
# 清理 settings cache，避免被其他测试污染
|
||
from app.core import config as config_mod
|
||
|
||
config_mod.get_settings.cache_clear()
|
||
|
||
# 重新加载 main，确保使用最新 env
|
||
import app.main as main_mod
|
||
|
||
importlib.reload(main_mod)
|
||
|
||
app = main_mod.create_app()
|
||
|
||
# override repo（避免依赖真实 DB）
|
||
from app.api.v1 import reco as reco_mod
|
||
from app.features.personalized_reco.content_repository.types import ContentProfileDTO
|
||
|
||
class _FakeRepo:
|
||
async def fetch_candidates(self, **kwargs): # type: ignore[no-untyped-def]
|
||
# 返回一条可下发内容
|
||
return [
|
||
ContentProfileDTO(
|
||
content_id=1,
|
||
text="t1",
|
||
stage="general",
|
||
emotion_score=None,
|
||
context_suitability={},
|
||
need_suitability={},
|
||
personalization_power=0.0,
|
||
risk_flags=[],
|
||
author_id=None,
|
||
template_id=None,
|
||
review_confidence=0.7,
|
||
)
|
||
]
|
||
|
||
async def fetch_contents_by_ids(self, **kwargs): # type: ignore[no-untyped-def]
|
||
return []
|
||
|
||
async def _override_repo(): # type: ignore[no-untyped-def]
|
||
return _FakeRepo()
|
||
|
||
app.dependency_overrides[reco_mod.get_reco_repo] = _override_repo
|
||
|
||
# 清空限流计数，避免跨测试污染
|
||
import app.api.limits as limits_mod
|
||
|
||
limits_mod._reco_rate_limiter._counters.clear() # type: ignore[attr-defined]
|
||
limits_mod._reco_rate_limiter._last_gc_bucket = 0 # type: ignore[attr-defined]
|
||
|
||
return TestClient(app)
|
||
|
||
|
||
def test_accept_language_mapping_to_tc(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
# 固定时间，避免跨分钟 flake
|
||
import app.api.limits as limits_mod
|
||
|
||
monkeypatch.setattr(limits_mod.time, "time", lambda: 1738497600.0) # 2025-02-02 12:00:00Z 的某个时间戳
|
||
|
||
resp = client.post(
|
||
"/v1/reco/feed",
|
||
json={"user_profile": _user_profile_dict(), "already_recommended_ids": [], "touched_or_viewed_ids": []},
|
||
headers={"Accept-Language": "zh-TW,zh;q=0.9"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["meta"]["config_snapshot"]["locale"] == "tc"
|
||
|
||
|
||
def test_x_now_header_priority_over_body_now(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
import app.api.limits as limits_mod
|
||
|
||
monkeypatch.setattr(limits_mod.time, "time", lambda: 1738497600.0)
|
||
|
||
resp = client.post(
|
||
"/v1/reco/push",
|
||
json={
|
||
"user_profile": _user_profile_dict(),
|
||
"now": "2026-02-01T00:00:00Z",
|
||
"already_recommended_ids": [],
|
||
"touched_or_viewed_ids": [],
|
||
},
|
||
headers={"X-Now": "2026-02-02T12:00:00Z", "Accept-Language": "en-US"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# meta.config_snapshot 里没有 now，但 served_k 应该正常
|
||
assert data["meta"]["served_k"] == 1
|
||
|
||
|
||
def test_rate_limit_10_per_minute(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
import app.api.limits as limits_mod
|
||
|
||
monkeypatch.setattr(limits_mod.time, "time", lambda: 1738497600.0)
|
||
|
||
body = {"user_profile": _user_profile_dict(), "already_recommended_ids": [], "touched_or_viewed_ids": []}
|
||
for _ in range(10):
|
||
r = client.post("/v1/reco/widget", json=body)
|
||
assert r.status_code == 200
|
||
|
||
r = client.post("/v1/reco/widget", json=body)
|
||
assert r.status_code == 429
|
||
assert r.json()["detail"] == "rate_limited"
|
||
|
||
|
||
def test_celery_tasks_can_call_generate(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
_set_min_env(monkeypatch)
|
||
from app.core import config as config_mod
|
||
|
||
config_mod.get_settings.cache_clear()
|
||
|
||
import app.tasks.reco as reco_tasks
|
||
|
||
# monkeypatch async runner，避免依赖 DB
|
||
async def _fake_run_reco_async(**kwargs): # type: ignore[no-untyped-def]
|
||
from app.features.personalized_reco.observability.types import RecoMeta
|
||
from app.features.personalized_reco.reco_engine.types import RecoEngineResult, RecommendedItem
|
||
|
||
return RecoEngineResult(
|
||
items=[RecommendedItem(content_id=1, text="t1", final_score=1.0, fallback_level_final=0, explanations={})],
|
||
meta=RecoMeta(scene="push", served_k=1),
|
||
)
|
||
|
||
monkeypatch.setattr(reco_tasks, "_run_reco_async", _fake_run_reco_async)
|
||
|
||
out = reco_tasks.generate(scene="push", user_profile=_user_profile_dict(), k=1)
|
||
assert out["meta"]["served_k"] == 1
|
||
|