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

"""内容运营 — 真实联调（仅查询类；群发写入默认跳过）"""

import pytest

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


def test_live_proxy_send_group_msg_status_shape(api, live_guid):
    """
    仅验证 QiWe 群发状态接口可达（使用占位 groupMsgId，预期业务错误而非 Token 错误）。
    不向真实群发送消息。
    """
    resp = api.post(
        "/qiwei/proxy",
        json={
            "method": "/msg/sendGroupMsgStatus",
            "params": {
                "guid": live_guid,
                "groupMsgId": "live-test-placeholder-not-real",
                "endDetailId": 2,
            },
        },
    )
    body = api.parse_json(resp)
    # 成功或 QiWe 业务错误均可，说明 Token+guid 已打通
    if body.get("success"):
        assert isinstance(body.get("data"), dict)
    else:
        err = body.get("error") or {}
        assert err.get("code") in ("QIWEI_API_ERROR", "INTERNAL_ERROR")
        assert "QIWEI_TOKEN_MISSING" not in str(err.get("code", ""))


def test_live_create_broadcast(api, live_guid, live_room_id, live_rooms_seeded):
    from load_env import load_backend_env

    if load_backend_env().get("QIWEI_LIVE_ALLOW_BROADCAST", "").strip() != "1":
        pytest.skip(
            "未开启真实群发：在 backend/.env 设置 QIWEI_LIVE_ALLOW_BROADCAST=1 "
            "（会向 QIWEI_ROOM_IDS 中的群发送测试文案）"
        )
    """若需自动化群发联调，去掉 skip 并确认 toIdList 为测试群"""
    api.assert_success(
        api.post(
            "/content/broadcasts",
            json={
                "guid": live_guid,
                "sendType": 1,
                "toIdList": [live_room_id],
                "msgList": [{"type": 0, "msgData": {"content": "自动化联调测试"}}],
            },
        ),
        status=201,
    )