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

"""
Swagger 模块 1（健康检查）+ 模块 2（企微）— 全部真实联调

覆盖 http://localhost:3101/api-docs 中：
  1-健康检查: GET /api/health
  2-企微: 全部 9 个接口
"""

from __future__ import annotations

import json
import time
import uuid
from pathlib import Path

import pytest
import requests

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

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


# ---------- 模块 1 ----------


def test_m01_health_live(require_pc_server):
    r = requests.get(f"{require_pc_server}/health", timeout=5)
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok"
    assert "qiwei" in body["modules"]


# ---------- 模块 2 工具 ----------


def _webhook_payload(live_guid: str, room_id: str, doc_suffix: str) -> tuple[dict, str]:
    payload = json.loads(FIXTURE.read_text(encoding="utf-8"))
    base = int(time.time() * 1000) + (hash(doc_suffix) % 100_000)
    docid = f"DOC_M12_{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


# ---------- 模块 2：按 Swagger 顺序 ----------


def test_m02_proxy_get_room_list(api, live_guid):
    data = api.assert_success(
        api.post(
            "/qiwei/proxy",
            json={
                "method": "/room/getRoomList",
                "params": {"guid": live_guid, "nextStartIndex": 0},
            },
        )
    )
    assert isinstance(data, dict)


def test_m02_proxy_get_session_list(api, live_guid, live_room_ids):
    data = api.assert_success(
        api.post(
            "/qiwei/proxy",
            json={"method": "/session/getSessionList", "params": {"guid": live_guid}},
        )
    )
    ids = set()
    for key in ("collectList", "shieldList", "topList", "markList"):
        for item in data.get(key) or []:
            if item.get("sessionType") == 1:
                ids.add(str(item["sessionId"]))
    for rid in live_room_ids:
        assert rid in ids


def test_m02_staff_status(api, live_guid):
    data = api.assert_success(api.get(f"/qiwei/staff/{live_guid}/status"))
    assert data.get("guid") == live_guid or data.get("userId")
    assert data["userOnlineStatus"] in (1, 2)


def test_m02_staff_batch_status(api, live_guid):
    data = api.assert_success(
        api.post("/qiwei/staff/batch-status", json={"guids": [live_guid]})
    )
    assert len(data) == 1


def test_m02_sync_messages(api, live_guid):
    data = api.assert_success(
        api.post("/qiwei/sync", json={"guid": live_guid, "msgSeq": 0, "limit": 20})
    )
    assert "messages" in data


def test_m02_webhook_register_doc(api, live_guid, live_room_id):
    suffix = uuid.uuid4().hex[:8]
    payload, docid = _webhook_payload(live_guid, live_room_id, suffix)
    data = api.assert_success(api.post("/qiwei/webhook", json=payload))
    assert data["docLinksFound"] >= 1
    doc = api.assert_success(api.get("/qiwei/room-docs", params={"roomId": live_room_id}))
    assert docid in doc["docId"]


def test_m02_messages_list(api, live_guid, live_room_id):
    msgs = api.assert_success(
        api.get("/qiwei/messages", params={"roomId": live_room_id, "guid": live_guid, "limit": 20})
    )
    assert isinstance(msgs, list)


def test_m02_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_m02_room_doc_anomalies_and_resolve(api, live_guid, live_room_id):
    suffix = uuid.uuid4().hex[:8]
    p1, doc1 = _webhook_payload(live_guid, live_room_id, f"x{suffix}")
    api.assert_success(api.post("/qiwei/webhook", json=p1))
    p2, doc2 = _webhook_payload(live_guid, live_room_id, f"y{suffix}")
    wh2 = api.assert_success(api.post("/qiwei/webhook", json=p2))
    assert wh2["processedCount"] >= 1

    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("newDocId") == doc2]
    assert hit, anomalies

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