cfc/cfc-langgraph/tests/test_memory.py

import pytest
from unittest.mock import AsyncMock, MagicMock, patch


class FakeEmbeddings:
    """最小 Embeddings 桩（O-14 终裁 2026-11-08 09:1x）。

    MemoryManager.__init__(embeddings) 已将 embeddings 改为必传
    （向量版记忆，langchain_chroma.Chroma 替换已废弃 langchain.memory）。
    本桩实现 langchain Embeddings 兼容接口，返回固定 4 维向量，
    仅用于单测构造 MemoryManager，不触发真实嵌入。
    """

    def embed_documents(self, texts):
        return [[0.0, 0.1, 0.2, 0.3] for _ in texts]

    def embed_query(self, text):
        return [0.0, 0.1, 0.2, 0.3]


@pytest.fixture
def fake_chroma():
    """用 MagicMock 替换 langchain_chroma.Chroma（隔离 chromadb 原生依赖）。

    单测验证的是 MemoryManager 的契约，而非向量库实现；因此不应被
    `chromadb_rust_bindings` 原生 DLL 缺失这类宿主机环境问题拖红（B-54）。
    """
    with patch("langchain_chroma.Chroma") as chroma_cls:
        chroma_cls.return_value = MagicMock()
        yield chroma_cls


@pytest.mark.asyncio
async def test_memory_manager_init(fake_chroma):
    """B-64 对齐：构造契约 —— embeddings 必传，集合名 user_memory。"""
    from app.memory.store import MemoryManager

    embeddings = FakeEmbeddings()
    mgr = MemoryManager(embeddings=embeddings)

    assert mgr is not None
    assert mgr.memory_vectorstore is not None
    assert mgr.embeddings is embeddings
    fake_chroma.assert_called_once()
    kwargs = fake_chroma.call_args.kwargs
    assert kwargs["collection_name"] == "user_memory"
    assert kwargs["embedding_function"] is embeddings


@pytest.mark.asyncio
async def test_save_conversation_writes_vectorstore(fake_chroma):
    """B-64 对齐：save_conversation 写入向量库并携带 user/conversation 元数据。"""
    from app.memory.store import MemoryManager

    mgr = MemoryManager(embeddings=FakeEmbeddings())
    mgr.memory_vectorstore.aadd_texts = AsyncMock()

    await mgr.save_conversation(
        user_id=1,
        conversation_id="conv-1",
        messages=[{"role": "user", "content": "你好"},
                  {"role": "assistant", "content": "你好呀"}],
    )

    mgr.memory_vectorstore.aadd_texts.assert_awaited_once()
    kwargs = mgr.memory_vectorstore.aadd_texts.call_args.kwargs
    assert len(kwargs["texts"]) == 1
    assert "[user] 你好" in kwargs["texts"][0]
    assert "[assistant] 你好呀" in kwargs["texts"][0]
    meta = kwargs["metadatas"][0]
    assert meta["user_id"] == 1
    assert meta["conversation_id"] == "conv-1"
    assert "timestamp" in meta


@pytest.mark.asyncio
async def test_recall_returns_contents(fake_chroma):
    """B-64 对齐：recall 按 user_id 过滤做相似度召回并返回纯文本列表。"""
    from app.memory.store import MemoryManager

    mgr = MemoryManager(embeddings=FakeEmbeddings())
    doc1, doc2 = MagicMock(), MagicMock()
    doc1.page_content = "历史记忆一"
    doc2.page_content = "历史记忆二"
    mgr.memory_vectorstore.similarity_search = MagicMock(return_value=[doc1, doc2])

    out = await mgr.recall(user_id=7, query="钙", k=2)

    assert out == ["历史记忆一", "历史记忆二"]
    kwargs = mgr.memory_vectorstore.similarity_search.call_args.kwargs
    assert kwargs["k"] == 2
    assert kwargs["filter"] == {"user_id": 7}


def test_legacy_memory_api_removed():
    """B-64 收口：langchain.memory 版旧接口已随迁移移除，防止回潮。"""
    from app.memory.store import MemoryManager

    assert not hasattr(MemoryManager, "get_working_memory")
    assert not hasattr(MemoryManager, "get_longterm_memory")