ai-kefu-jc/tests/test_lead_push_wiring.py

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

from sqlalchemy import select

from app.models.database import ConversationSource, LeadPushLog
from app.services import email_service, lead_push_service
from app.services.conversation import ConversationManager


class FakeAI:
    """同 test_lead_email_trigger 的确定性桩。"""

    def __init__(self, extract=None):
        self._extract_seq = extract if isinstance(extract, list) else [extract or {}]

    async def stream_chat(self, messages, context=""):
        for token in ("ok ", "reply"):
            yield token

    async def extract_info(self, messages, existing_info):
        idx = min(max(len(messages) // 2 - 1, 0), len(self._extract_seq) - 1)
        return self._extract_seq[idx]

    async def evaluate(self, profile, recent_messages):
        return {"intent_level": "interested", "reasoning": "留号",
                "contact_info": {}, "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=""):
        sent.append(("lead", dict(contact_info)))
        return lead_succeeds

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

    async def fake_fix(contact_info, count=1):
        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: fired.extend(logs))
    monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, 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"}})
    (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"}})
    (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"}})
    (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"}}])
    (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"}})
    (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"}})
    (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"}})

    sent = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level=""):
        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: fired.extend(logs))

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, 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):
        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=""):
        sent.append("lead")
        return True

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

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, 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"}})
    sent = []
    read_result = {}
    commit_gate: asyncio.Event | None = None
    fired = []

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

    async def fake_run(log_id, payload):
        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):
        fired.append(True)
        real_fire(logs)

    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, 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"}