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

"""群管理模块 — 真实联调（QIWEI_ROOM_IDS / API-07/08/09）"""

import time

import pytest

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


def test_live_sync_by_room_ids(api, live_guid, live_room_ids, live_rooms_seeded):
    """POST /rooms/sync + roomIdList，从 .env QIWEI_ROOM_IDS 登记群"""
    data = api.assert_success(
        api.post(
            "/rooms/sync",
            json={
                "guid": live_guid,
                "roomIdList": live_room_ids,
            },
        )
    )
    assert data["roomCount"] == len(live_room_ids)
    synced_ids = {r["roomId"] for r in data["rooms"]}
    assert synced_ids == set(live_rooms_seeded)


def test_live_rooms_list_all_configured(api, live_room_ids, live_rooms_seeded):
    rooms = api.assert_success(api.get("/rooms"))
    listed = {r["roomId"] for r in rooms}
    for rid in live_room_ids:
        assert rid in listed


def test_live_room_detail_each(api, live_room_ids, live_rooms_seeded):
    for room_id in live_room_ids:
        room = api.assert_success(api.get(f"/rooms/{room_id}"))
        assert room["roomId"] == room_id
        assert room.get("roomName")


def test_live_room_health_each(api, live_room_ids, live_rooms_seeded):
    for room_id in live_room_ids:
        health = api.assert_success(api.get(f"/rooms/{room_id}/health"))
        assert health["roomId"] == room_id
        assert "healthScore" in health


def test_live_member_events_sync_each(api, live_guid, live_room_ids, live_rooms_seeded):
    """API-09 /room/getRoomIncrSync"""
    for room_id in live_room_ids:
        data = api.assert_success(
            api.post(
                "/rooms/member-events/sync",
                json={"guid": live_guid, "roomId": room_id, "ver": 0},
            )
        )
        assert "events" in data
        assert "nextVer" in data


def test_live_proxy_batch_room_detail_all(api, live_guid, live_room_ids):
    data = api.assert_success(
        api.post(
            "/qiwei/proxy",
            json={
                "method": "/room/batchGetRoomDetail",
                "params": {"guid": live_guid, "roomIdList": live_room_ids},
            },
        )
    )
    assert isinstance(data.get("roomList"), list)
    assert len(data["roomList"]) >= 1


def test_live_get_room_list_may_be_empty(api, live_guid):
    """getRoomList 可为空；与 QIWEI_ROOM_IDS 联调不冲突"""
    data = api.assert_success(
        api.post(
            "/qiwei/proxy",
            json={
                "method": "/room/getRoomList",
                "params": {"guid": live_guid, "nextStartIndex": 0},
            },
        )
    )
    assert "roomList" in data