Underground-Digital/Workflow-Engine

1import os2from collections.abc import Callable3from typing import Literal4 5import httpx6import pytest7from _pytest.monkeypatch import MonkeyPatch8 9 10def mock_get(*args, **kwargs):11    if kwargs.get("headers", {}).get("Authorization") != "Bearer test":12        raise httpx.HTTPStatusError(13            "Invalid API key",14            request=httpx.Request("GET", ""),15            response=httpx.Response(401),16        )17 18    return httpx.Response(19        200,20        json={21            "items": [22                {"title": "Model 1", "_id": "model1"},23                {"title": "Model 2", "_id": "model2"},24            ]25        },26        request=httpx.Request("GET", ""),27    )28 29 30def mock_stream(*args, **kwargs):31    class MockStreamResponse:32        def __init__(self):33            self.status_code = 20034 35        def __enter__(self):36            return self37 38        def __exit__(self, exc_type, exc_val, exc_tb):39            pass40 41        def iter_bytes(self):42            yield b"Mocked audio data"43 44    return MockStreamResponse()45 46 47def mock_fishaudio(48    monkeypatch: MonkeyPatch,49    methods: list[Literal["list-models", "tts"]],50) -> Callable[[], None]:51    """52    mock fishaudio module53 54    :param monkeypatch: pytest monkeypatch fixture55    :return: unpatch function56    """57 58    def unpatch() -> None:59        monkeypatch.undo()60 61    if "list-models" in methods:62        monkeypatch.setattr(httpx, "get", mock_get)63 64    if "tts" in methods:65        monkeypatch.setattr(httpx, "stream", mock_stream)66 67    return unpatch68 69 70MOCK = os.getenv("MOCK_SWITCH", "false").lower() == "true"71 72 73@pytest.fixture74def setup_fishaudio_mock(request, monkeypatch):75    methods = request.param if hasattr(request, "param") else []76    if MOCK:77        unpatch = mock_fishaudio(monkeypatch, methods=methods)78 79    yield80 81    if MOCK:82        unpatch()83