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

"""Swagger 模块 7（风控）— 真实联调（扫描使用真实 roomId）"""

import uuid

import pytest

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


@pytest.fixture(scope="module")
def m07_keyword(api_session, live_room_id):
    suffix = uuid.uuid4().hex[:6]
    kw = api_session.assert_success(
        api_session.post(
            "/risk/keywords",
            json={
                "keyword": f"联调敏感词{suffix}",
                "category": "complaint",
                "severity": 3,
                "enabled": 1,
                "remark": "module7-live",
            },
        ),
        status=201,
    )
    yield {"id": kw["id"], "keyword": kw["keyword"], "room_id": live_room_id}
    api_session.assert_success(api_session.delete(f"/risk/keywords/{kw['id']}"))


def test_m07_keywords_list(api, m07_keyword):
    listed = api.assert_success(api.get("/risk/keywords"))
    assert any(k["id"] == m07_keyword["id"] for k in listed)


def test_m07_keywords_scan_real_room(api, m07_keyword):
    scan = api.assert_success(
        api.post(
            "/risk/keywords/scan",
            json={
                "content": f"测试{m07_keyword['keyword']}消息",
                "roomId": m07_keyword["room_id"],
                "messageId": 90001,
            },
        )
    )
    assert "hitCount" in scan or "hits" in scan or "alerts" in scan


def test_m07_update_keyword(api, m07_keyword):
    updated = api.assert_success(
        api.put(
            f"/risk/keywords/{m07_keyword['id']}",
            json={"remark": "已更新"},
        )
    )
    assert updated["remark"] == "已更新"


def test_m07_alerts_list(api):
    alerts = api.assert_success(api.get("/risk/alerts"))
    assert isinstance(alerts, list)


def test_m07_handle_alert_if_any(api):
    alerts = api.assert_success(api.get("/risk/alerts"))
    if not alerts:
        pytest.skip("无预警可处理")
    handled = api.assert_success(
        api.put(
            f"/risk/alerts/{alerts[0]['id']}",
            json={"status": "handled", "handlerId": 1, "note": "模块7联调"},
        )
    )
    assert handled.get("status") in ("handled", "ignored")


def test_m07_work_orders_list(api):
    orders = api.assert_success(api.get("/risk/work-orders"))
    assert isinstance(orders, list)


def test_m07_update_work_order_if_any(api):
    orders = api.assert_success(api.get("/risk/work-orders"))
    if not orders:
        pytest.skip("无工单可更新")
    updated = api.assert_success(
        api.put(
            f"/risk/work-orders/{orders[0]['id']}",
            json={"status": "in_progress"},
        )
    )
    assert updated.get("status") == "in_progress"