SupplyAgent/tests/supply_infra/pipeline/test_aigc_safety.py

from __future__ import annotations

from supply_infra.aigc.client import AigcClient
from supply_infra.config import InfraSettings


def test_aigc_client_requires_token_for_live_calls(monkeypatch) -> None:
    settings = InfraSettings(_env_file=None)
    monkeypatch.setattr(
        "supply_infra.aigc.client.get_infra_settings",
        lambda: settings,
    )
    client = AigcClient(token="")
    try:
        client.create_video_crawler_plan(["123"])
    except RuntimeError as exc:
        assert "AIGC_API_TOKEN" in str(exc)
    else:
        raise AssertionError("missing token must raise")


def test_aigc_client_posts_to_platform(monkeypatch) -> None:
    settings = InfraSettings(_env_file=None)
    monkeypatch.setattr(
        "supply_infra.aigc.client.get_infra_settings",
        lambda: settings,
    )

    class _Response:
        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict:
            return {"code": 0, "data": {"id": "plan-1"}}

    calls: list[dict] = []

    def _post(*, url, json, headers, timeout):
        calls.append({"url": url, "json": json})
        return _Response()

    monkeypatch.setattr("supply_infra.aigc.client.requests.post", _post)
    client = AigcClient(token="secret")
    result = client.create_video_crawler_plan(["123"])
    assert result["success"] is True
    assert result["crawler_plan_id"] == "plan-1"
    assert len(calls) == 1