ai-kefu-admin/tests/test_channel_hygiene.py

"""内部状态卫生(同步自上游 ai-kefu tests/test_channel_hygiene.py)。

上游两测分别锚 AIService.extract_info 内部 `_` 过滤与 _merge_extracted
纯函数的出口剥离;jc 主线把这两处卫生逻辑收敛进 stream_process_message
的调用点(入口侧 existing_info 过滤 / 出口侧 cc_reply 剥离 + `_` 回显
丢弃,不再有独立函数)——断言按 jc 锚点重锚到 manager 整轮,语义等价:
`_` 键不进提示词、cc_reply 消费即剥离不落 raw_info、extractor 回显的
`_` 状态不得覆盖 Python 管理的状态机。
"""
import asyncio

from app.engine.services import email_service
from app.engine.services.conversation import ConversationManager
from tests.conftest import FakeAI, build_jiancai_runtime

KE_IP = "41.90.64.1"   # 肯尼亚段(iptocc 实测 → KE)


async def _send(manager, sid, text):
    async for _ in manager.stream_process_message(sid, text):
        pass


def _no_email(monkeypatch):
    """卫生测试不关心邮件侧效应——桩掉避免真实 SMTP 配置探测。"""

    async def fake_lead(*a, **k):
        return True

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)


def test_extract_info_filters_underscore_keys(async_db, monkeypatch):
    # 入口侧(上游锚 AIService 内部过滤,jc 锚调用点): `_` 前缀内部态
    # (_cc 状态机等)绝不进 LLM 输入,正常键照常透传。
    _no_email(monkeypatch)
    captured = {}

    class RecAI(FakeAI):
        async def extract_info(self, messages, existing_info):
            captured["existing"] = existing_info
            return {}

    async def run():
        async with async_db() as db:
            manager = ConversationManager(
                db, runtime=build_jiancai_runtime(), ai_service=RecAI())
            conv = await manager.create_conversation(client_info={"ip": KE_IP})
            profile = await manager._get_profile(conv.id)
            profile.raw_info = {"name": "Sam",
                                "_cc": {"whatsapp": {"e164": "+254712345678"}}}
            db.add(profile)
            await db.commit()
            await _send(manager, conv.session_id, "hello are you there")

    asyncio.run(run())
    assert captured["existing"] == {"name": "Sam"}   # _cc 不进提示词,name 保留


def test_extractor_output_strips_cc_reply_and_underscore_echo(async_db, monkeypatch):
    # 出口侧(上游锚 _merge_extracted,jc 锚 stream_process_message 内联):
    # cc_reply 提取后即剥离不落 raw_info;extractor 回显的任何 `_` 键
    # (含 _cc 状态机回显)防御性丢弃——Python 管理的状态机不被覆盖;
    # 正常键(含联系方式深合并)照常进 raw_info。
    _no_email(monkeypatch)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(
                db, runtime=build_jiancai_runtime(),
                ai_service=FakeAI(extract={
                    "联系方式": {"whatsapp": "0712345678"},
                    "cc_reply": {"affirm": True},
                    "_cc": {"whatsapp": {"e164": "HACKED"}},
                    "_evil": "x",
                    "name": "Sam2"}))
            conv = await manager.create_conversation(client_info={"ip": KE_IP})
            await _send(manager, conv.session_id, "my whatsapp is 0712345678")
            profile = await manager._get_profile(conv.id)
            return dict(profile.raw_info or {})

    raw = asyncio.run(run())
    assert "cc_reply" not in raw and "_evil" not in raw
    assert raw["name"] == "Sam2"
    assert raw["联系方式"]["whatsapp"] == "0712345678"
    # _cc 在场但属 Python 建态(判定树按 KE IP 补全),extractor 回显未生效
    st = raw["_cc"]["whatsapp"]
    assert st["e164"] == "+254712345678" and st["source"] == "ip_proposed"