mindfulness/server/tests/test_reco_engine.py

from __future__ import annotations

from datetime import datetime, timezone

import pytest

from app.features.personalized_reco.content_repository.interface import ContentRepository
from app.features.personalized_reco.content_repository.types import ContentProfileDTO
from app.features.personalized_reco.reco_engine.orchestrator import recommend
from app.features.personalized_reco.reco_engine.types import RecoConstraints
from app.features.user_profile_scoring.types import ProfileAnswered, UserProfileV1_2, UserStageOneHot


class _FakeRepo(ContentRepository):
    def __init__(self, candidates_by_level: dict[int, list[ContentProfileDTO]]):
        self._candidates_by_level = candidates_by_level

    async def fetch_candidates(  # type: ignore[override]
        self,
        *,
        scene: str,
        user_profile: object,
        fallback_level: int,
        limit: int,
        locale: str,
        exclude_content_ids: list[int] | None = None,
    ) -> list[ContentProfileDTO]:
        # 单测：简化实现，只按 level 返回，忽略 limit/locale/exclude
        return list(self._candidates_by_level.get(int(fallback_level), []))[: int(limit)]

    async def fetch_contents_by_ids(self, *, content_ids: list[int], locale: str) -> list[ContentProfileDTO]:  # type: ignore[override]
        raise NotImplementedError


def _now() -> datetime:
    return datetime(2026, 2, 2, 12, 0, 0, tzinfo=timezone.utc)


def _user_profile(*, stage: str = "unknown", emotion_score: float | None = 0.5) -> UserProfileV1_2:
    if stage == "expecting":
        st = UserStageOneHot(expecting=1, parenting=0, unknown=0)  # type: ignore[arg-type]
    elif stage == "parenting":
        st = UserStageOneHot(expecting=0, parenting=1, unknown=0)  # type: ignore[arg-type]
    else:
        st = UserStageOneHot(expecting=0, parenting=0, unknown=1)  # type: ignore[arg-type]

    return UserProfileV1_2(
        profile_generated_at=_now(),
        profile_confidence=1.0,
        profile_answered=ProfileAnswered(stage=True, emotion=emotion_score is not None, context=False, need=False),
        stage=st,
        emotion_score=emotion_score,
        context={},
        need={},
    )


def _content(
    *,
    content_id: int,
    text: str = "hello",
    stage: str = "general",
    personalization_power: float = 0.0,
    risk_flags: list[str] | None = None,
    author_id: str | None = None,
    template_id: str | None = None,
    need_suitability: dict[str, float] | None = None,
) -> ContentProfileDTO:
    return ContentProfileDTO(
        content_id=int(content_id),
        text=text,
        stage=stage,  # type: ignore[arg-type]
        emotion_score=None,
        context_suitability={},
        need_suitability=need_suitability or {},
        personalization_power=float(personalization_power),
        risk_flags=risk_flags or [],
        author_id=author_id,
        template_id=template_id,
        review_confidence=0.7,
    )


@pytest.mark.asyncio
async def test_reco_engine_k_zero_returns_empty() -> None:
    repo = _FakeRepo({0: [_content(content_id=1)]})
    res = await recommend(
        repo=repo,
        scene="feed",
        user_profile=_user_profile(),
        already_recommended_ids=[],
        touched_or_viewed_ids=[],
        k=0,
        now=_now(),
        locale=None,
    )
    assert res.items == []
    assert res.meta.served_k == 0


@pytest.mark.asyncio
async def test_reco_engine_pool_empty_sets_empty_reason() -> None:
    repo = _FakeRepo({0: []})
    res = await recommend(
        repo=repo,
        scene="feed",
        user_profile=_user_profile(),
        already_recommended_ids=[],
        touched_or_viewed_ids=[],
        k=1,
        now=_now(),
        locale="en",
    )
    assert res.items == []
    assert res.meta.empty_reason == "pool_empty"


@pytest.mark.asyncio
async def test_reco_engine_hard_filter_all_sets_empty_reason() -> None:
    repo = _FakeRepo({0: [_content(content_id=1, risk_flags=["block_health_medical"])]})
    res = await recommend(
        repo=repo,
        scene="push",
        user_profile=_user_profile(stage="unknown"),
        already_recommended_ids=[],
        touched_or_viewed_ids=[],
        k=1,
        now=_now(),
        locale="en",
    )
    assert res.items == []
    assert res.meta.empty_reason == "hard_filter_all"
    assert res.meta.risk_filtered_count_by_flag.get("block_health_medical", 0) == 1


@pytest.mark.asyncio
async def test_reco_engine_freqcap_all_sets_empty_reason() -> None:
    # Push 场景：作者冷却命中，导致 after_freqcap=0
    repo = _FakeRepo({0: [_content(content_id=1, author_id="a1")]})
    res = await recommend(
        repo=repo,
        scene="push",
        user_profile=_user_profile(stage="unknown"),
        already_recommended_ids=[],
        touched_or_viewed_ids=[],
        k=1,
        now=_now(),
        locale="en",
        constraints=RecoConstraints(recent_author_ids=["a1"]),
    )
    assert res.items == []
    assert res.meta.empty_reason == "freqcap_all"


@pytest.mark.asyncio
async def test_reco_engine_returns_item_and_explanations_enabled() -> None:
    repo = _FakeRepo({0: [_content(content_id=1, text="t1", personalization_power=0.0)]})
    res = await recommend(
        repo=repo,
        scene="feed",
        user_profile=_user_profile(stage="unknown"),
        already_recommended_ids=[],
        touched_or_viewed_ids=[],
        k=1,
        now=_now(),
        locale="en",
    )
    assert len(res.items) == 1
    assert res.items[0].content_id == 1
    assert res.items[0].text == "t1"
    assert res.items[0].explanations is not None