SupplyAgent/tests/supply_infra/pipeline/test_aigc_safety.py

from __future__ import annotations

from unittest.mock import MagicMock

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


def test_bind_skips_duplicate_existing_crawler_source(monkeypatch) -> None:
    settings = InfraSettings(_env_file=None)
    monkeypatch.setattr(
        "supply_infra.aigc.client.get_infra_settings",
        lambda: settings,
    )
    client = AigcClient(token="secret")
    monkeypatch.setattr(
        client,
        "_get_produce_plan_detail",
        lambda _plan_id: (
            {
                "name": "produce-plan",
                "inputSourceGroups": [
                    {
                        "inputSources": [
                            {
                                "contentType": 1,
                                "inputSourceType": 2,
                                "inputSourceValue": "crawler-1",
                                "inputSourceModal": 4,
                                "inputSourceChannel": 2,
                            }
                        ]
                    }
                ],
            },
            None,
        ),
    )
    post = MagicMock()
    monkeypatch.setattr(client, "_post", post)

    result = client.bind_crawler_to_produce_plan(
        "crawler-1",
        "produce-1",
        crawler_plan_name="crawler-name",
    )

    assert result["success"] is True
    assert result["already_bound"] is True
    post.assert_not_called()