ai-kefu-admin/tests/test_lead_push_wiring.py

"""询盘推送接线集成测试: 跟随邮件三类型/SMTP解耦/commit后派发(设计 §8)。"""
import asyncio

from sqlalchemy import select

from app.engine.models import ConversationSource, LeadPushLog
from app.engine.services import email_service, lead_push_service
from app.engine.services.conversation import ConversationManager
from tests.conftest import FakeAI, build_jiancai_runtime

# 本文件场景依赖"interested"触发推送——conftest FakeAI(超集)经 evaluate
# 参数注入,与原本地拷贝的固定返回值逐字等价(setdefault 补 contact_info={})
_EVAL = {"intent_level": "interested", "reasoning": "留号",
         "info_sufficient": True}


def _setup(async_db, monkeypatch, ai, lead_succeeds=True):
    """邮件桩 + 推送桩: fire 只收集不真发(避免测试触网)。"""
    sent = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level="", cc_info=None,
                        config_data=None):
        sent.append(("lead", dict(contact_info)))
        return lead_succeeds

    async def fake_nudge(contact_info, cc_info=None, config_data=None):
        sent.append(("returning", dict(contact_info)))
        return lead_succeeds

    async def fake_fix(contact_info, count=1, cc_info=None, config_data=None):
        sent.append(("correction", dict(contact_info)))
        return lead_succeeds

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
    monkeypatch.setattr(email_service, "send_returning_lead_email", fake_nudge)
    monkeypatch.setattr(email_service, "send_correction_email", fake_fix)

    fired = []
    monkeypatch.setattr(lead_push_service, "fire_pending_pushes",
                        lambda logs, session_factory=None, config_data=None:
                        fired.extend(logs))
    monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, runtime=build_jiancai_runtime(),
                                          ai_service=ai)
            conv = await manager.create_conversation(
                source=ConversationSource.widget)
            return manager, conv
    return asyncio.run(run()), sent, fired


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


def _logs(async_db):
    async def run():
        async with async_db() as db:
            return list((await db.execute(select(LeadPushLog))).scalars())
    return asyncio.run(run())


def test_lead_email_triggers_pending_push(async_db, monkeypatch):
    ai = FakeAI(extract={"联系方式": {"手机": "+254722111117"}}, evaluate=_EVAL)
    (manager, conv), sent, fired = _setup(async_db, monkeypatch, ai)
    _send(manager, conv.session_id, "my whatsapp is +254722111117")
    assert [s[0] for s in sent] == ["lead"]
    assert len(fired) == 1 and fired[0].status == "pending"
    assert fired[0].push_type == "lead"
    assert fired[0].request_payload["event"] == "new"
    assert fired[0].request_payload["contact"]["phone"] == "+254722111117"
    asyncio.run(manager.db.close())


def test_smtp_failure_still_pushes(async_db, monkeypatch):
    """设计定案: 推送与 SMTP 成败解耦——邮件失败询盘照推。"""
    ai = FakeAI(extract={"联系方式": {"手机": "+254722111117"}}, evaluate=_EVAL)
    (manager, conv), sent, fired = _setup(
        async_db, monkeypatch, ai, lead_succeeds=False)
    _send(manager, conv.session_id, "my whatsapp is +254722111117")
    assert sent and sent[0][0] == "lead"   # fake 返回 False,email_sent 不置位
    assert len(fired) == 1 and fired[0].status == "pending"
    asyncio.run(manager.db.close())


def test_wechat_only_push_marked_skipped(async_db, monkeypatch):
    ai = FakeAI(extract={"联系方式": {"微信": "li_ming_88"}}, evaluate=_EVAL)
    (manager, conv), sent, fired = _setup(async_db, monkeypatch, ai)
    _send(manager, conv.session_id, "add my wechat li_ming_88")
    assert [s[0] for s in sent] == ["lead"]          # 我方邮件照发(wechat 算 confirmed)
    assert fired == []                                # 无后台任务
    logs = _logs(async_db)
    assert len(logs) == 1 and logs[0].status == "skipped"
    asyncio.run(manager.db.close())


def test_correction_push_event_update_after_success(async_db, monkeypatch):
    """改号 → 更正推送;前次已 success(手动置位模拟 _run_push)→ event=update。"""
    ai = FakeAI(extract=[{"联系方式": {"手机": "+254722111117"}},
                         {"联系方式": {"手机": "+254799999999"}}],
                evaluate=_EVAL)
    (manager, conv), sent, fired = _setup(async_db, monkeypatch, ai)
    _send(manager, conv.session_id, "my whatsapp is +254722111117")
    assert len(fired) == 1 and fired[0].push_type == "lead"

    async def mark_success():
        async with async_db() as db:
            for row in (await db.execute(select(LeadPushLog))).scalars():
                row.status = "success"
            await db.commit()
    asyncio.run(mark_success())

    _send(manager, conv.session_id, "wrong, use +254799999999")
    assert [s[0] for s in sent] == ["lead", "correction"]
    assert len(fired) == 2
    assert fired[1].push_type == "correction"
    assert fired[1].request_payload["event"] == "update"
    asyncio.run(manager.db.close())


def test_returning_push_follows_nudge(async_db, monkeypatch):
    """24h 内同值另已发过 → 回访轻提醒 → 推送 push_type=returning,event=new。"""
    ai1 = FakeAI(extract={"联系方式": {"email": "same@x.com"}}, evaluate=_EVAL)
    (m1, c1), _s1, f1 = _setup(async_db, monkeypatch, ai1)
    _send(m1, c1.session_id, "mail same@x.com")
    assert len(f1) == 1 and f1[0].push_type == "lead"
    asyncio.run(m1.db.close())

    ai2 = FakeAI(extract={"联系方式": {"email": "same@x.com"}}, evaluate=_EVAL)
    (m2, c2), sent2, f2 = _setup(async_db, monkeypatch, ai2)
    _send(m2, c2.session_id, "im back, same@x.com")
    assert [s[0] for s in sent2] == ["returning"]
    assert len(f2) == 1 and f2[0].push_type == "returning"
    assert f2[0].request_payload["event"] == "new"
    asyncio.run(m2.db.close())


def test_push_disabled_writes_nothing(async_db, monkeypatch):
    """PUSH_ENABLED=false(默认): 零日志零派发,既有行为完全不变。"""
    monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", False)
    ai = FakeAI(extract={"联系方式": {"手机": "+254722111117"}}, evaluate=_EVAL)

    sent = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level="", cc_info=None,
                        config_data=None):
        sent.append("lead")
        return True

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
    fired = []
    monkeypatch.setattr(lead_push_service, "fire_pending_pushes",
                        lambda logs, session_factory=None, config_data=None:
                        fired.extend(logs))

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, runtime=build_jiancai_runtime(),
                                          ai_service=ai)
            conv = await manager.create_conversation(
                source=ConversationSource.widget)
            return manager, conv
    (manager, conv) = asyncio.run(run())
    _send(manager, conv.session_id, "my whatsapp is +254722111117")
    assert sent == ["lead"] and fired == [] and _logs(async_db) == []
    asyncio.run(manager.db.close())


def test_fire_end_to_end_single_loop(async_db, monkeypatch):
    """真 fire + 假 _run_push: 同一事件循环内 stream 完成后任务确实被派发执行。"""
    pushed = []

    async def fake_run(log_id, payload, session_factory=None, config_data=None):
        pushed.append(log_id)

    monkeypatch.setattr(lead_push_service, "_run_push", fake_run)
    monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True)

    sent = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level="", cc_info=None,
                        config_data=None):
        sent.append("lead")
        return True

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
    ai = FakeAI(extract={"联系方式": {"手机": "+254722111117"}}, evaluate=_EVAL)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, runtime=build_jiancai_runtime(),
                                          ai_service=ai)
            conv = await manager.create_conversation(
                source=ConversationSource.widget)
            async for _ in manager.stream_process_message(
                    conv.session_id, "my whatsapp is +254722111117"):
                pass
            await asyncio.sleep(0)   # 让 create_task 的任务跑一轮
    asyncio.run(run())
    assert sent == ["lead"] and len(pushed) == 1


def test_fire_happens_only_after_commit(async_db, monkeypatch):
    """顺序钉住(设计 8.1): fire 在主事务 commit 之后——后台任务从独立连接
    必须能读到该 pending 行;若 fire 被挪到 commit 前,此处 NoResultFound 变红。

    加固(变异实测): fire-and-forget 任务里的异常只会被 shutdown 吞掉、
    单次 sleep(0) 也不够任务跑完真实 aiosqlite I/O——原形态变异后仍绿。
    改为: 任务把读取结果记回 read_result;fire 后的首次 commit 让位
    (gate,≤50ms)给后台读先行——broken 序(fire 在 commit 前)时读窗口
    确定性落在提交前 → NoResultFound 被记录 → 变红;正确序 gate 不触发。
    """
    from sqlalchemy import select as sa_select
    ai = FakeAI(extract={"联系方式": {"手机": "+254722111117"}}, evaluate=_EVAL)
    sent = []
    read_result = {}
    commit_gate: asyncio.Event | None = None
    fired = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level="", cc_info=None,
                        config_data=None):
        sent.append("lead")
        return True

    async def fake_run(log_id, payload, session_factory=None, config_data=None):
        async with async_db() as db:
            try:
                row = (await db.execute(
                    sa_select(LeadPushLog).where(LeadPushLog.id == log_id)
                )).scalar_one()
                read_result["status"] = row.status
            except Exception as exc:   # NoResultFound → fire 先于 commit
                read_result["error"] = repr(exc)
            finally:
                if commit_gate is not None:
                    commit_gate.set()

    real_fire = lead_push_service.fire_pending_pushes

    def recording_fire(logs, session_factory=None, config_data=None):
        fired.append(True)
        real_fire(logs, session_factory, config_data)

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
    monkeypatch.setattr(lead_push_service, "_run_push", fake_run)
    monkeypatch.setattr(lead_push_service, "fire_pending_pushes", recording_fire)
    monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, runtime=build_jiancai_runtime(),
                                          ai_service=ai)
            conv = await manager.create_conversation(
                source=ConversationSource.widget)

            # commit 闸门: 仅当 fire 已发生(broken 序才会如此)时,主事务
            # commit 让位给后台读(否则后台读与 commit 落地是毫秒级竞态)
            nonlocal commit_gate
            commit_gate = asyncio.Event()
            real_commit = manager.db.commit

            async def gated_commit():
                if fired:
                    try:
                        await asyncio.wait_for(commit_gate.wait(), timeout=0.05)
                    except asyncio.TimeoutError:
                        pass
                await real_commit()
            manager.db.commit = gated_commit

            async for _ in manager.stream_process_message(
                    conv.session_id, "my whatsapp is +254722111117"):
                pass
            for _ in range(400):            # 等后台任务读完成(≤2s,实际毫秒级)
                if read_result:
                    break
                await asyncio.sleep(0.005)
    asyncio.run(run())
    assert sent == ["lead"]
    assert read_result == {"status": "pending"}