mirror of
https://git.mirrors.martin98.com/https://github.com/langgenius/dify.git
synced 2025-05-14 00:18:17 +08:00
83 lines
1.8 KiB
Python
83 lines
1.8 KiB
Python
import os
|
|
from collections.abc import Callable
|
|
from typing import Literal
|
|
|
|
import httpx
|
|
import pytest
|
|
from _pytest.monkeypatch import MonkeyPatch
|
|
|
|
|
|
def mock_get(*args, **kwargs):
|
|
if kwargs.get("headers", {}).get("Authorization") != "Bearer test":
|
|
raise httpx.HTTPStatusError(
|
|
"Invalid API key",
|
|
request=httpx.Request("GET", ""),
|
|
response=httpx.Response(401),
|
|
)
|
|
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"items": [
|
|
{"title": "Model 1", "_id": "model1"},
|
|
{"title": "Model 2", "_id": "model2"},
|
|
]
|
|
},
|
|
request=httpx.Request("GET", ""),
|
|
)
|
|
|
|
|
|
def mock_stream(*args, **kwargs):
|
|
class MockStreamResponse:
|
|
def __init__(self):
|
|
self.status_code = 200
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
pass
|
|
|
|
def iter_bytes(self):
|
|
yield b"Mocked audio data"
|
|
|
|
return MockStreamResponse()
|
|
|
|
|
|
def mock_fishaudio(
|
|
monkeypatch: MonkeyPatch,
|
|
methods: list[Literal["list-models", "tts"]],
|
|
) -> Callable[[], None]:
|
|
"""
|
|
mock fishaudio module
|
|
|
|
:param monkeypatch: pytest monkeypatch fixture
|
|
:return: unpatch function
|
|
"""
|
|
|
|
def unpatch() -> None:
|
|
monkeypatch.undo()
|
|
|
|
if "list-models" in methods:
|
|
monkeypatch.setattr(httpx, "get", mock_get)
|
|
|
|
if "tts" in methods:
|
|
monkeypatch.setattr(httpx, "stream", mock_stream)
|
|
|
|
return unpatch
|
|
|
|
|
|
MOCK = os.getenv("MOCK_SWITCH", "false").lower() == "true"
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_fishaudio_mock(request, monkeypatch):
|
|
methods = request.param if hasattr(request, "param") else []
|
|
if MOCK:
|
|
unpatch = mock_fishaudio(monkeypatch, methods=methods)
|
|
|
|
yield
|
|
|
|
if MOCK:
|
|
unpatch()
|