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

"""
Swagger 模块 3（人员）— 真实联调

使用 backend/.env 的 QIWEI_GUID 回填人员，并调用 QiWe 校验在线状态。
覆盖 api-docs 中 6 个接口。
"""

from __future__ import annotations

import uuid

import pytest

from load_env import load_backend_env

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


@pytest.fixture(scope="module")
def live_staff_record(api_session, live_guid):
    """创建人员 → 绑定真实 guid → 用例结束后删除"""
    env = load_backend_env()
    suffix = uuid.uuid4().hex[:8]
    phone = f"139{suffix[:8]}"

    created = api_session.assert_success(
        api_session.post(
            "/staff",
            json={
                "name": f"联调人员_{suffix}",
                "role": "operator",
                "storeId": 1,
                "phone": phone,
            },
        ),
        status=201,
    )
    staff_id = created["id"]

    updated = api_session.assert_success(
        api_session.put(
            f"/staff/{staff_id}",
            json={"guid": live_guid, "status": 1},
        )
    )
    assert updated["guid"] == live_guid

    yield {"id": staff_id, "phone": phone, "name": created["name"], "guid": live_guid}

    api_session.assert_success(api_session.delete(f"/staff/{staff_id}"))


def test_m03_create_staff(live_staff_record):
    assert live_staff_record["id"] > 0
    assert live_staff_record["guid"]


def test_m03_list_staff(api, live_staff_record):
    listed = api.assert_success(
        api.get(
            "/staff",
            params={"role": "operator", "storeId": 1, "status": 1},
        )
    )
    assert any(s["id"] == live_staff_record["id"] for s in listed)


def test_m03_staff_detail(api, live_staff_record):
    detail = api.assert_success(api.get(f"/staff/{live_staff_record['id']}"))
    assert detail["guid"] == live_staff_record["guid"]
    assert detail["phone"] == live_staff_record["phone"]


def test_m03_active_guids(api, live_guid, live_staff_record):
    data = api.assert_success(api.get("/staff/active-guids"))
    assert live_guid in data["guids"]
    assert data["count"] >= 1


def test_m03_qiwei_online_matches_staff(api, live_guid, live_staff_record):
    """人员 guid 与 QiWe 在线接口一致（跨模块真实校验）"""
    online = api.assert_success(api.get(f"/qiwei/staff/{live_guid}/status"))
    assert online.get("guid") == live_guid or online.get("userId")
    assert online["userOnlineStatus"] in (1, 2)
    assert online.get("nickname") or online.get("userId")


def test_m03_update_staff(api, live_staff_record):
    updated = api.assert_success(
        api.put(
            f"/staff/{live_staff_record['id']}",
            json={"name": live_staff_record["name"] + "_已更新"},
        )
    )
    assert updated["name"].endswith("_已更新")