lami/lami-base-v1/backend/tests_python/test_19_qiwei_docs_live.py

"""企微模块 — 台账 / webhook / 消息（真实 roomId + guid，补 test_13 未覆盖接口）"""

import json
import time
import uuid
from pathlib import Path

import pytest

pytestmark = [pytest.mark.integration, pytest.mark.qiwei_live]

FIXTURE = Path(__file__).parent / "fixtures" / "qiwei_webhook_sample.json"


def _live_webhook_payload(live_guid: str, room_id: str, doc_suffix: str) -> dict:
    payload = json.loads(FIXTURE.read_text(encoding="utf-8"))
    # 毫秒级 + 后缀哈希，避免同群连续 webhook 被去重
    base = int(time.time() * 1000) + (hash(doc_suffix) % 100_000)
    docid = f"DOC_LIVE_{doc_suffix}"
    for i, item in enumerate(payload.get("data", [])):
        item["guid"] = live_guid
        item["fromRoomId"] = room_id
        item["timestamp"] = base + i
        if item.get("msgType") == 13:
            item["msgData"]["linkUrl"] = (
                f"https://doc.weixin.qq.com/txdoc/excel?docid={docid}"
            )
    return payload, docid


def test_live_webhook_register_room_doc(api, live_guid, live_room_id):
    suffix = uuid.uuid4().hex[:8]
    payload, docid = _live_webhook_payload(live_guid, live_room_id, suffix)
    data = api.assert_success(api.post("/qiwei/webhook", json=payload))
    assert data["processedCount"] >= 1
    assert data["docLinksFound"] >= 1

    doc = api.assert_success(api.get("/qiwei/room-docs", params={"roomId": live_room_id}))
    assert doc["roomId"] == live_room_id
    assert docid in doc["docId"]


def test_live_messages_list(api, live_guid, live_room_id):
    """先 sync 真消息，再查列表（可与 webhook 并存）"""
    api.assert_success(
        api.post("/qiwei/sync", json={"guid": live_guid, "msgSeq": 0, "limit": 10})
    )
    msgs = api.assert_success(
        api.get("/qiwei/messages", params={"roomId": live_room_id, "limit": 20})
    )
    assert isinstance(msgs, list)


def test_live_room_docs_list(api, live_room_id):
    docs = api.assert_success(api.get("/qiwei/room-docs"))
    assert isinstance(docs, list)
    assert any(d.get("roomId") == live_room_id for d in docs)


def test_live_room_doc_anomaly_and_resolve(api, live_guid, live_room_id):
    """同一真群连续登记两个不同 docid → 多表异常 → 人工解决"""
    suffix = uuid.uuid4().hex[:8]
    p1, doc1 = _live_webhook_payload(live_guid, live_room_id, f"a{suffix}")
    api.assert_success(api.post("/qiwei/webhook", json=p1))

    p2, doc2 = _live_webhook_payload(live_guid, live_room_id, f"b{suffix}")
    api.assert_success(api.post("/qiwei/webhook", json=p2))

    anomalies = api.assert_success(api.get("/qiwei/room-doc-anomalies"))
    hit = [
        a
        for a in anomalies
        if a.get("roomId") == live_room_id
        and a.get("type") == "multi_doc"
        and a.get("newDocId") == doc2
    ]
    assert hit, f"应产生 multi_doc 异常 newDocId={doc2}, 实际={anomalies}"

    resolved = api.assert_success(
        api.put(
            f"/qiwei/room-doc-anomalies/{live_room_id}/resolve",
            json={"newDocId": doc2, "resolution": "keep_new"},
        )
    )
    assert resolved["roomId"] == live_room_id
    assert resolved["status"] == "resolved"