wehub-resource-sync
212 lines
7.4 KiB
Python
212 lines
7.4 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Tests for GET /v1/models listing audio models (INV-02).
|
|
|
|
Verifies that audio_stt and audio_tts models appear in the /v1/models
|
|
response with correct fields, and that they coexist with other engine types.
|
|
|
|
All tests use FastAPI TestClient with a mocked EnginePool — no mlx-audio
|
|
or real model loading required.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_engine_entry(
|
|
model_id: str,
|
|
model_type: str,
|
|
engine_type: str,
|
|
engine=None,
|
|
is_pinned: bool = False,
|
|
is_loading: bool = False,
|
|
) -> MagicMock:
|
|
"""Build a minimal mock EngineEntry."""
|
|
entry = MagicMock()
|
|
entry.model_id = model_id
|
|
entry.model_type = model_type
|
|
entry.engine_type = engine_type
|
|
entry.engine = engine
|
|
entry.is_pinned = is_pinned
|
|
entry.is_loading = is_loading
|
|
entry.estimated_size = 1024 * 1024 * 500 # 500 MB
|
|
entry.last_access = 0.0
|
|
return entry
|
|
|
|
|
|
def _make_pool(entries: list) -> MagicMock:
|
|
"""Build a mock EnginePool with the given entries."""
|
|
pool = MagicMock()
|
|
pool.preload_pinned_models = AsyncMock()
|
|
pool.check_ttl_expirations = AsyncMock()
|
|
pool.shutdown = AsyncMock()
|
|
pool.get_model_ids.return_value = [e.model_id for e in entries]
|
|
pool.get_entry.side_effect = lambda mid: next(
|
|
(e for e in entries if e.model_id == mid), None
|
|
)
|
|
pool.get_status.return_value = {
|
|
"final_ceiling": 32 * 1024**3,
|
|
"current_model_memory": 0,
|
|
"model_count": len(entries),
|
|
"loaded_count": sum(1 for e in entries if e.engine is not None),
|
|
"models": [
|
|
{
|
|
"id": e.model_id,
|
|
"model_type": e.model_type,
|
|
"engine_type": e.engine_type,
|
|
"loaded": e.engine is not None,
|
|
"pinned": e.is_pinned,
|
|
"is_loading": e.is_loading,
|
|
"estimated_size": e.estimated_size,
|
|
"last_access": e.last_access,
|
|
}
|
|
for e in entries
|
|
],
|
|
}
|
|
return pool
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestModelsListAudio
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestModelsListAudio:
|
|
"""GET /v1/models must include audio models with correct fields."""
|
|
|
|
@pytest.fixture
|
|
def stt_entry(self):
|
|
return _make_engine_entry(
|
|
"whisper-large-v3", "audio_stt", "stt", engine=MagicMock()
|
|
)
|
|
|
|
@pytest.fixture
|
|
def tts_entry(self):
|
|
return _make_engine_entry(
|
|
"qwen3-tts", "audio_tts", "tts", engine=None
|
|
)
|
|
|
|
@pytest.fixture
|
|
def llm_entry(self):
|
|
return _make_engine_entry(
|
|
"llama-3b", "llm", "batched", engine=None
|
|
)
|
|
|
|
@pytest.fixture
|
|
def client_with_stt(self, stt_entry):
|
|
"""TestClient with a pool containing only an STT model."""
|
|
from omlx.server import app
|
|
|
|
mock_pool = _make_pool([stt_entry])
|
|
with patch("omlx.server._server_state") as mock_state:
|
|
mock_state.engine_pool = mock_pool
|
|
mock_state.global_settings = None
|
|
mock_state.process_memory_enforcer = None
|
|
mock_state.hf_downloader = None
|
|
mock_state.ms_downloader = None
|
|
mock_state.mcp_manager = None
|
|
mock_state.api_key = None
|
|
mock_state.settings_manager = MagicMock()
|
|
mock_state.settings_manager.get_settings.return_value = MagicMock(
|
|
model_alias=None, is_hidden=False
|
|
)
|
|
with TestClient(app, raise_server_exceptions=False) as client:
|
|
yield client, mock_pool
|
|
|
|
@pytest.fixture
|
|
def client_with_mixed(self, stt_entry, tts_entry, llm_entry):
|
|
"""TestClient with a pool containing STT + TTS + LLM models."""
|
|
from omlx.server import app
|
|
|
|
mock_pool = _make_pool([stt_entry, tts_entry, llm_entry])
|
|
with patch("omlx.server._server_state") as mock_state:
|
|
mock_state.engine_pool = mock_pool
|
|
mock_state.global_settings = None
|
|
mock_state.process_memory_enforcer = None
|
|
mock_state.hf_downloader = None
|
|
mock_state.ms_downloader = None
|
|
mock_state.mcp_manager = None
|
|
mock_state.api_key = None
|
|
mock_state.settings_manager = MagicMock()
|
|
mock_state.settings_manager.get_settings.return_value = MagicMock(
|
|
model_alias=None, is_hidden=False
|
|
)
|
|
with TestClient(app, raise_server_exceptions=False) as client:
|
|
yield client, mock_pool
|
|
|
|
def test_models_list_returns_200(self, client_with_stt):
|
|
client, _ = client_with_stt
|
|
response = client.get("/v1/models")
|
|
assert response.status_code == 200
|
|
|
|
def test_models_list_includes_stt_model(self, client_with_stt):
|
|
"""audio_stt model appears in /v1/models response."""
|
|
client, _ = client_with_stt
|
|
response = client.get("/v1/models")
|
|
assert response.status_code == 200
|
|
|
|
body = response.json()
|
|
assert "data" in body
|
|
model_ids = [m["id"] for m in body["data"]]
|
|
assert "whisper-large-v3" in model_ids
|
|
|
|
def test_stt_model_has_required_openai_fields(self, client_with_stt):
|
|
"""Each model entry has id, object, owned_by per OpenAI spec."""
|
|
client, _ = client_with_stt
|
|
response = client.get("/v1/models")
|
|
body = response.json()
|
|
|
|
stt_model = next(
|
|
(m for m in body["data"] if m["id"] == "whisper-large-v3"), None
|
|
)
|
|
assert stt_model is not None, "whisper-large-v3 not found in /v1/models"
|
|
assert "id" in stt_model
|
|
assert "object" in stt_model
|
|
assert "owned_by" in stt_model
|
|
|
|
def test_stt_model_object_field_value(self, client_with_stt):
|
|
"""Model object field is 'model'."""
|
|
client, _ = client_with_stt
|
|
response = client.get("/v1/models")
|
|
body = response.json()
|
|
|
|
stt_model = next(
|
|
(m for m in body["data"] if m["id"] == "whisper-large-v3"), None
|
|
)
|
|
assert stt_model is not None
|
|
assert stt_model["object"] == "model"
|
|
|
|
def test_models_list_includes_tts_model(self, client_with_mixed):
|
|
"""audio_tts model appears in /v1/models response."""
|
|
client, _ = client_with_mixed
|
|
response = client.get("/v1/models")
|
|
assert response.status_code == 200
|
|
|
|
body = response.json()
|
|
model_ids = [m["id"] for m in body["data"]]
|
|
assert "qwen3-tts" in model_ids
|
|
|
|
def test_audio_models_coexist_with_llm(self, client_with_mixed):
|
|
"""Audio models and LLM appear together in the same /v1/models response."""
|
|
client, _ = client_with_mixed
|
|
response = client.get("/v1/models")
|
|
body = response.json()
|
|
model_ids = {m["id"] for m in body["data"]}
|
|
|
|
assert "whisper-large-v3" in model_ids
|
|
assert "qwen3-tts" in model_ids
|
|
assert "llama-3b" in model_ids
|
|
|
|
def test_models_list_response_top_level_fields(self, client_with_stt):
|
|
"""Response top-level has 'object' and 'data' fields."""
|
|
client, _ = client_with_stt
|
|
body = client.get("/v1/models").json()
|
|
assert body.get("object") == "list"
|
|
assert isinstance(body.get("data"), list)
|