ai-kefu-jc/tests/test_lead_push_service.py

"""询盘推送 UniService:模型/服务单元测试(设计 2026-09-28-lead-push-to-uniservice)。"""
import asyncio

from sqlalchemy import select

from app.models.database import (
    Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog,
    Message, MessageRole,
)


def test_lead_push_log_roundtrip(async_db):
    async def run():
        async with async_db() as db:
            conv = Conversation(session_id="s1")
            db.add(conv)
            await db.flush()
            ev = Evaluation(
                conversation_id=conv.id,
                intent_level=IntentLevel.pending, contact_info={},
            )
            db.add(ev)
            await db.flush()
            db.add(LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type="lead", status="pending",
            ))
            await db.commit()
            row = (await db.execute(select(LeadPushLog))).scalar_one()
            assert row.id and len(row.id) == 36          # UUID 主键,同既有四表
            assert row.status == "pending" and row.attempts == 0
            assert row.outcome is None and row.response_status is None
    asyncio.run(run())


def test_push_settings_defaults():
    from app.config import Settings
    s = Settings(_env_file=None)   # 隔离本机 .env(联调时会写入 PUSH_* 真值)
    assert s.PUSH_ENABLED is False
    assert s.PUSH_API_URL == (
        "https://uniservice.server.hnxinyouhe.com/open/v1/inquiry/push")
    assert s.PUSH_API_KEY == ""
    assert s.PUSH_EXTERNAL_KEY_PREFIX == "ai-kefu-jc"  # 与汽车项目线索键空间隔离
    assert s.PUSH_TIMEOUT_SECONDS == 10.0
    assert s.PUSH_MAX_ATTEMPTS == 3
    assert s.PUSH_RETRY_BACKOFF_BASE == 1.0


def test_fmt_cst_converts_utc_and_naive():
    from datetime import datetime, timezone
    from app.services.lead_push_service import _fmt_cst
    aware = datetime(2026, 9, 28, 2, 30, 0, tzinfo=timezone.utc)
    assert _fmt_cst(aware) == "2026-09-28T10:30:00+08:00"   # UTC→+08:00,绝不发 UTC
    naive = datetime(2026, 9, 28, 2, 30, 0)                 # SQLite 读回形态,按 UTC 补
    assert _fmt_cst(naive) == "2026-09-28T10:30:00+08:00"
    micro = datetime(2026, 9, 28, 2, 30, 0, 123456, tzinfo=timezone.utc)
    assert _fmt_cst(micro) == "2026-09-28T10:30:00+08:00"   # 微秒归一整秒(契约示例精度)


def test_map_intent():
    from app.services.lead_push_service import map_intent
    assert map_intent("interested", True) == "high"
    assert map_intent("pending", True) == "medium"
    assert map_intent("not_interested", True) == "low"      # 销售优先级解读,设计 7.2 定案
    assert map_intent("pending", False) == "unknown"        # 未评估≠待确认(D2 语义)
    assert map_intent("bogus", True) == "unknown"
    assert map_intent(None, True) == "unknown"
    assert map_intent(IntentLevel.pending, True) == "medium"   # DB 读回枚举成员形态


def _seed_push_ctx(async_db, client_info=None):
    """建 conv/profile/ev 三件套,返回 (conv, profile, ev)。"""
    from app.models.database import ConversationSource

    async def run():
        async with async_db() as db:
            conv = Conversation(
                session_id="s-push", source=ConversationSource.widget,
                client_info=client_info or {"ip": "8.8.8.8", "lang": "en"},
            )
            db.add(conv)
            await db.flush()
            profile = CustomerProfile(
                conversation_id=conv.id, raw_info={"目标国家": "肯尼亚"})
            ev = Evaluation(
                conversation_id=conv.id,
                intent_level=IntentLevel.interested, contact_info={})
            db.add_all([profile, ev])
            await db.commit()
    asyncio.run(run())


def test_dispatch_disabled_returns_none(async_db, monkeypatch):
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", False)
    _seed_push_ctx(async_db)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            return await svc.dispatch_lead_push(
                db, ev, conv, profile, {"phone": "+254722111117"},
                "lead", {"intent_level": "interested"}, True)
    assert asyncio.run(run()) is None
    assert asyncio.run(_select_logs(async_db)) == []   # 开关关闭零落库


def test_dispatch_wechat_only_marks_skipped(async_db, monkeypatch):
    """wechat-only: 我方发邮件(wechat 算 confirmed)但对方必拒三选一 → skipped 留痕不派发。"""
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            result = await svc.dispatch_lead_push(
                db, ev, conv, profile, {"wechat": "li_ming_88"},
                "lead", {"intent_level": "interested"}, True)
            await db.commit()
            return result
    assert asyncio.run(run()) is None
    logs = asyncio.run(_select_logs(async_db))
    assert len(logs) == 1
    assert logs[0].status == "skipped"
    assert "phone/whatsapp/email" in logs[0].error_message


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


def test_dispatch_pending_payload_structure(async_db, monkeypatch):
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            db.add(Message(conversation_id=conv.id, role=MessageRole.user,
                           content="my whatsapp is +254722111117"))
            await db.commit()
            confirmed = {"phone": "+254722111117", "whatsapp": "+254722111117"}
            log = await svc.dispatch_lead_push(
                db, ev, conv, profile, confirmed, "lead",
                {"intent_level": "interested", "reasoning": "询库存留号"}, True)
            await db.commit()
            return conv, log
    conv, log = asyncio.run(run())
    assert log is not None and log.status == "pending"
    p = log.request_payload
    assert p["event"] == "new"
    assert p["external_key"] == f"{svc.settings.PUSH_EXTERNAL_KEY_PREFIX}-{conv.id}"  # 不硬编码前缀: 联调 .env 会写真实 PUSH_* 值(同 test_push_settings_defaults 的隔离教训)
    assert p["occurred_at"].endswith("+08:00")
    assert p["contact"]["phone"] == "+254722111117"
    assert p["contact"]["skype"] is None
    assert p["intent_level"] == "high" and p["intent_reason"] == "询库存留号"
    assert p["profile"]["目标国家"] == "肯尼亚"
    assert p["profile"]["source"] == "widget"
    assert p["profile"]["country"]            # 8.8.8.8 → 国名非空
    assert p["conversation"][0]["role"] == "customer"
    assert p["conversation"][0]["at"].endswith("+08:00")


def test_dispatch_correction_event_rules(async_db, monkeypatch):
    """7.1 保护规则: 同会话有 success 前科 → update;没有 → new。"""
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)

    async def run(with_prior_success):
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            if with_prior_success:
                db.add(LeadPushLog(
                    evaluation_id=ev.id, conversation_id=conv.id,
                    push_type="lead", status="success"))
                await db.flush()
            log = await svc.dispatch_lead_push(
                db, ev, conv, profile, {"phone": "+254700000001"},
                "correction", {"intent_level": "interested"}, True)
            await db.commit()
            return log.request_payload["event"]
    assert asyncio.run(run(False)) == "new"
    assert asyncio.run(run(True)) == "update"


def test_dispatch_swallows_exceptions(async_db, monkeypatch):
    """红线: dispatch 内部炸(如 payload 组装抛错)绝不向上抛。"""
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)
    async def boom(*a, **kw):
        raise RuntimeError("boom")
    monkeypatch.setattr(svc, "_build_payload", boom)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            return await svc.dispatch_lead_push(
                db, ev, conv, profile, {"phone": "+254700000001"},
                "lead", {}, True)
    assert asyncio.run(run()) is None


def test_dispatch_db_failure_does_not_poison_caller_tx(async_db, monkeypatch):
    """红线(DB 层): flush 失败只回滚推送写入,调用方事务仍可提交(防重复发邮件)。"""
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)

    async def bad_payload(*a, **kw):
        return {"bad": object()}   # JSON 不可序列化 → flush 必炸

    monkeypatch.setattr(svc, "_build_payload", bad_payload)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            ev.reasoning = "caller update after dispatch failure"
            db.add(ev)
            result = await svc.dispatch_lead_push(
                db, ev, conv, profile, {"phone": "+254700000001"},
                "lead", {}, True)
            logs = list((await db.execute(select(LeadPushLog))).scalars())
            await db.commit()
            return result, logs
    result, logs = asyncio.run(run())
    assert result is None and logs == []

    async def reload():
        async with async_db() as db:
            from app.models.database import Evaluation
            ev = (await db.execute(select(Evaluation))).scalar_one()
            return ev.reasoning
    assert asyncio.run(reload()) == "caller update after dispatch failure"


def test_dispatch_returning_with_prior_success_still_new(async_db, monkeypatch):
    """不变量: 仅 correction 可 update——returning 即使同会话已 success 也走 new。"""
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
    _seed_push_ctx(async_db)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, CustomerProfile, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            profile = (await db.execute(select(CustomerProfile))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            db.add(LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type="lead", status="success"))
            await db.flush()
            log = await svc.dispatch_lead_push(
                db, ev, conv, profile, {"phone": "+254700000001"},
                "returning", {"intent_level": "interested"}, True)
            await db.commit()
            return log.request_payload["event"]
    assert asyncio.run(run()) == "new"


def _mk_push_log(async_db, status="pending"):
    async def run():
        async with async_db() as db:
            from app.models.database import ConversationSource
            conv = Conversation(session_id=f"s-{status}-{id(async_db)}",
                                source=ConversationSource.widget)
            db.add(conv)
            await db.flush()
            ev = Evaluation(conversation_id=conv.id,
                            intent_level=IntentLevel.pending, contact_info={})
            db.add(ev)
            await db.flush()
            log = LeadPushLog(evaluation_id=ev.id, conversation_id=conv.id,
                              push_type="lead", status=status,
                              request_payload={"external_key": "ai-kefu-x",
                                               "event": "new"})
            db.add(log)
            await db.commit()
            return log.id
    return asyncio.run(run())


def _run_push_case(async_db, monkeypatch, do_post, log_id):
    """注入 _do_post/_sleep/async_session 后驱动 _run_push,返回退避序列。"""
    from app.services import lead_push_service as svc
    sleeps = []

    async def fake_sleep(s):
        sleeps.append(s)

    monkeypatch.setattr(svc, "_do_post", do_post)
    monkeypatch.setattr(svc, "_sleep", fake_sleep)
    monkeypatch.setattr(svc, "async_session", async_db)

    async def run():
        await svc._run_push(log_id, {"external_key": "ai-kefu-x", "event": "new"})
    asyncio.run(run())
    return sleeps


def _get_log(async_db, log_id):
    async def run():
        async with async_db() as db:
            return await db.get(LeadPushLog, log_id)
    return asyncio.run(run())


def _resp(status, body=None):
    import httpx
    return httpx.Response(status, json=body or {})


def test_run_push_success_accepted(async_db, monkeypatch):
    async def ok(payload):
        return _resp(200, {"success": True, "data": {
            "accepted": 1, "duplicated": 0, "failed": []}}), None
    lid = _mk_push_log(async_db)
    _run_push_case(async_db, monkeypatch, ok, lid)
    log = _get_log(async_db, lid)
    assert log.status == "success" and log.outcome == "accepted"
    assert log.response_status == 200 and log.attempts == 1


def test_run_push_success_duplicated(async_db, monkeypatch):
    async def dup(payload):
        return _resp(200, {"success": True, "data": {
            "accepted": 0, "duplicated": 1, "failed": []}}), None
    lid = _mk_push_log(async_db)
    _run_push_case(async_db, monkeypatch, dup, lid)
    log = _get_log(async_db, lid)
    assert log.status == "success" and log.outcome == "duplicated"


def test_run_push_200_item_rejected_no_retry(async_db, monkeypatch):
    """200 但本条在 failed[](如联系方式无效): failed+对方 reason,不重试。"""
    calls = []

    async def rejected(payload):
        calls.append(1)
        return _resp(200, {"success": True, "data": {
            "accepted": 0, "duplicated": 0,
            "failed": [{"external_key": "ai-kefu-x", "reason": "无有效联系方式"}]}}), None
    lid = _mk_push_log(async_db)
    sleeps = _run_push_case(async_db, monkeypatch, rejected, lid)
    log = _get_log(async_db, lid)
    assert log.status == "failed" and "无有效联系方式" in log.error_message
    assert len(calls) == 1 and sleeps == []


def test_run_push_5xx_retries_then_failed(async_db, monkeypatch):
    calls = []

    async def always_500(payload):
        calls.append(1)
        return _resp(500), None
    lid = _mk_push_log(async_db)
    sleeps = _run_push_case(async_db, monkeypatch, always_500, lid)
    log = _get_log(async_db, lid)
    assert len(calls) == 3 and log.attempts == 3 and log.status == "failed"
    assert log.response_status == 500
    assert sleeps == [1.0, 2.0]        # 指数退避: 3 次尝试 2 次间隔


def test_run_push_401_fails_fast(async_db, monkeypatch):
    calls = []

    async def unauthorized(payload):
        calls.append(1)
        return _resp(401), None
    lid = _mk_push_log(async_db)
    sleeps = _run_push_case(async_db, monkeypatch, unauthorized, lid)
    log = _get_log(async_db, lid)
    assert len(calls) == 1 and log.attempts == 1 and log.status == "failed"
    assert sleeps == []


def test_run_push_transport_error_then_success(async_db, monkeypatch):
    calls = []

    async def flaky(payload):
        calls.append(1)
        if len(calls) == 1:
            return None, "ConnectError('timeout')"
        return _resp(200, {"success": True, "data": {
            "accepted": 1, "duplicated": 0, "failed": []}}), None
    lid = _mk_push_log(async_db)
    sleeps = _run_push_case(async_db, monkeypatch, flaky, lid)
    log = _get_log(async_db, lid)
    assert log.status == "success" and log.attempts == 2 and sleeps == [1.0]


def test_run_push_crash_guard(async_db, monkeypatch):
    """红线: _do_post 抛非 HTTP 异常也不得向外传播。"""
    async def boom(payload):
        raise RuntimeError("boom")
    lid = _mk_push_log(async_db)
    _run_push_case(async_db, monkeypatch, boom, lid)   # 不 raise 即通过
    log = _get_log(async_db, lid)
    assert log.status == "pending"      # 结果未写,留给启动清扫兜底


def test_run_push_200_nondict_json_normalized(async_db, monkeypatch):
    """防御: 200 但 body 是合法 JSON 非对象(如数组)——归一为空 data,不 crash。"""
    async def weird(payload):
        return _resp(200, ["not", "a", "dict"]), None
    lid = _mk_push_log(async_db)
    _run_push_case(async_db, monkeypatch, weird, lid)
    log = _get_log(async_db, lid)
    assert log.status == "success" and log.outcome == "accepted"


def test_run_push_200_html_body_fails(async_db, monkeypatch):
    """200 但 body 非 JSON(如重定向到 HTML 登录页)——不得误记 success。"""
    import httpx

    async def html(payload):
        return httpx.Response(200, text="<html>login</html>"), None
    lid = _mk_push_log(async_db)
    _run_push_case(async_db, monkeypatch, html, lid)
    log = _get_log(async_db, lid)
    assert log.status == "failed" and "non-JSON" in log.error_message
    assert log.response_status == 200


def test_fire_pending_creates_tasks(monkeypatch):
    """create_task 派发(不阻塞语义的结构保证);非 pending 不派。"""
    from types import SimpleNamespace
    from app.services import lead_push_service as svc
    pushed = []

    async def fake_run(log_id, payload):
        pushed.append((log_id, payload))

    monkeypatch.setattr(svc, "_run_push", fake_run)
    logs = [
        SimpleNamespace(id="id-pending", status="pending",
                        request_payload={"event": "new"}),
        SimpleNamespace(id="id-skipped", status="skipped",
                        request_payload=None),
    ]

    async def run():
        svc.fire_pending_pushes(logs)
        await asyncio.sleep(0)   # 同 loop 内让 create_task 跑一轮
    asyncio.run(run())
    assert pushed == [("id-pending", {"event": "new"})]


def test_sweep_stale_pending(async_db, monkeypatch):
    from datetime import datetime, timezone
    from app.services import lead_push_service as svc
    monkeypatch.setattr(svc, "async_session", async_db)
    _seed_push_ctx(async_db)

    async def run():
        async with async_db() as db:
            from app.models.database import Conversation, Evaluation
            conv = (await db.execute(select(Conversation))).scalar_one()
            ev = (await db.execute(select(Evaluation))).scalar_one()
            stale = LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type="lead", status="pending",
                updated_at=datetime(2020, 1, 1, tzinfo=timezone.utc))
            fresh = LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type="lead", status="pending")
            done = LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type="lead", status="success",
                updated_at=datetime(2020, 1, 1, tzinfo=timezone.utc))
            db.add_all([stale, fresh, done])
            await db.commit()
        n = await svc.sweep_stale_pending(stale_minutes=30)
        assert n == 1
        async with async_db() as db:
            logs = list((await db.execute(select(LeadPushLog))).scalars())
            by = {(l.status, l.error_message or ""): l for l in logs}
            assert ("failed", "interrupted by restart") in by
            statuses = sorted(l.status for l in logs)
            assert statuses == ["failed", "pending", "success"]
    asyncio.run(run())