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

"""Swagger 模块 8（内容运营）— 真实联调（计划目标群用 QIWEI_ROOM_IDS）"""

import uuid

import pytest

from load_env import load_backend_env

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


@pytest.fixture(scope="module")
def m08_ctx(api_session, live_room_ids):
    suffix = uuid.uuid4().hex[:8]
    mat = api_session.assert_success(
        api_session.post(
            "/content/materials",
            json={
                "title": f"联调话术_{suffix}",
                "type": "script",
                "content": "真实联调素材内容",
                "tags": ["live"],
                "creatorId": 1,
            },
        ),
        status=201,
    )
    plan = api_session.assert_success(
        api_session.post(
            "/content/plans",
            json={
                "name": f"联调计划_{suffix}",
                "startDate": "2026-05-25",
                "endDate": "2026-05-31",
                "ownerId": 1,
            },
        ),
        status=201,
    )
    item = api_session.assert_success(
        api_session.post(
            f"/content/plans/{plan['id']}/items",
            json={
                "planId": plan["id"],
                "content": "针对真实客户群预告",
                "scheduledDate": "2026-05-30",
                "targetRoomIds": live_room_ids,
                "materialId": mat["id"],
            },
        ),
        status=201,
    )
    ctx = {
        "mat_id": mat["id"],
        "plan_id": plan["id"],
        "item_id": item["id"],
        "room_ids": live_room_ids,
    }
    yield ctx
    api_session.assert_success(api_session.delete(f"/content/materials/{mat['id']}"))


def test_m08_materials_list(api, m08_ctx):
    mats = api.assert_success(api.get("/content/materials"))
    assert any(m["id"] == m08_ctx["mat_id"] for m in mats)


def test_m08_plan_items(api, m08_ctx):
    items = api.assert_success(api.get(f"/content/plans/{m08_ctx['plan_id']}/items"))
    assert len(items) >= 1
    assert m08_ctx["room_ids"][0] in (items[0].get("targetRoomIds") or [])


def test_m08_plan_execution(api, m08_ctx):
    ex = api.assert_success(api.get(f"/content/plans/{m08_ctx['plan_id']}/execution"))
    assert isinstance(ex, dict)


def test_m08_broadcasts_list(api):
    listed = api.assert_success(api.get("/content/broadcasts"))
    assert isinstance(listed, list)


def test_m08_broadcast_real_optional(api, live_guid, m08_ctx):
    if load_backend_env().get("QIWEI_LIVE_ALLOW_BROADCAST", "").strip() != "1":
        pytest.skip("未开启 QIWEI_LIVE_ALLOW_BROADCAST=1，跳过真实群发")
    task = api.assert_success(
        api.post(
            "/content/broadcasts",
            json={
                "guid": live_guid,
                "sendType": 1,
                "toIdList": [m08_ctx["room_ids"][0]],
                "msgList": [{"type": 0, "msgData": {"content": "[模块8联调]测试"}}],
            },
        ),
        status=201,
    )
    status = api.assert_success(api.get(f"/content/broadcasts/{task['id']}"))
    assert status.get("groupMsgId") or status.get("id")