life-echo/api/tests/test_memory_consistency_rules.py

"""Memory 检索：无 recent fallback；compaction 后事实 stale。"""

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

import pytest

from app.features.memory import repo as memory_repo


@pytest.mark.asyncio
async def test_search_facts_async_does_not_use_recent_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[bool] = []

    async def boom(*_a, **_k):
        calls.append(True)
        raise AssertionError("get_facts_for_user should not run")

    monkeypatch.setattr(memory_repo, "get_facts_for_user", boom)

    mock_session = MagicMock()
    result = MagicMock()
    result.unique.return_value.scalars.return_value.all.return_value = []
    mock_session.execute = AsyncMock(return_value=result)

    out = await memory_repo.search_facts_for_user_async(
        mock_session, "user-1", "no_such_subject_xyz", 5
    )
    assert out == []
    assert calls == []


@pytest.mark.asyncio
async def test_search_facts_async_empty_query_returns_empty_without_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[bool] = []

    async def boom(*_a, **_k):
        calls.append(True)
        raise AssertionError("get_facts_for_user should not run")

    monkeypatch.setattr(memory_repo, "get_facts_for_user", boom)

    mock_session = MagicMock()
    out = await memory_repo.search_facts_for_user_async(mock_session, "user-1", "", 5)
    assert out == []
    assert calls == []


@pytest.mark.asyncio
async def test_mark_facts_stale_for_excluded_chunk_returns_rowcount() -> None:
    session = MagicMock()
    session.execute = AsyncMock(return_value=MagicMock(rowcount=3))
    n = await memory_repo.mark_facts_stale_for_excluded_chunk(
        session, user_id="u1", chunk_id="chunk-9"
    )
    assert n == 3