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

"""工作台与通知 — 真实联调（模板 + API-14 sendText）"""

import pytest

from load_env import load_backend_env

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


@pytest.fixture(scope="session")
def live_staff_id(api_session, live_guid: str) -> int:
    """创建绑定真实 guid 的人员，供通知 receiverId 使用"""
    digits = "".join(c for c in live_guid if c.isdigit())[:8] or "00000001"
    phone = f"139{digits}"
    created = api_session.assert_success(
        api_session.post(
            "/staff",
            json={
                "name": "联调测试员",
                "role": "operator",
                "storeId": 1,
                "phone": phone,
            },
        ),
        status=201,
    )
    staff_id = created["id"]
    api_session.assert_success(
        api_session.put(
            f"/staff/{staff_id}",
            json={"guid": live_guid, "status": 1},
        )
    )
    return staff_id


def test_live_notification_template(api, live_room_ids, live_rooms_seeded):
    room = api.assert_success(api.get(f"/rooms/{live_room_ids[0]}"))
    tpl = api.assert_success(
        api.get(
            "/notifications/template",
            params={"issueType": "missing_doc", "roomName": room["roomName"]},
        )
    )
    assert room["roomName"] in tpl["content"]
    assert "整改" in tpl["title"]


def test_live_send_notification_via_qiwe(api, live_guid, live_staff_id, live_room_ids):
    env = load_backend_env()
    if env.get("QIWEI_LIVE_NOTIFY", "").strip() != "1":
        pytest.skip(
            "未开启真实企微通知：backend/.env 设置 QIWEI_LIVE_NOTIFY=1 "
            "（会向 QIWEI_ROOM_IDS 第一个群发送一条测试文案）"
        )

    room = api.assert_success(api.get(f"/rooms/{live_room_ids[0]}"))
    tpl = api.assert_success(
        api.get(
            "/notifications/template",
            params={"issueType": "missing_doc", "roomName": room["roomName"]},
        )
    )

    sent = api.assert_success(
        api.post(
            "/notifications/send",
            json={
                "type": "compliance",
                "receiverId": live_staff_id,
                "title": tpl["title"],
                "content": tpl["content"] + " [自动化联调]",
                "refId": 1,
                "guid": live_guid,
                "toId": live_room_ids[0],
            },
        ),
        status=201,
    )
    assert sent["channel"] == "qiwe"
    assert sent["status"] == "sent", sent

    listed = api.assert_success(
        api.get("/notifications", params={"receiverId": live_staff_id, "type": "compliance"})
    )
    assert any(n["id"] == sent["id"] for n in listed)

    logs = api.assert_success(api.get("/audit-logs", params={"action": "notify", "limit": 5}))
    assert any("QiWe" in (log.get("detail") or "") for log in logs)