"""Unit tests for FastAPI endpoints in both app.py and studio_server.py. Uses FastAPI TestClient to test offline route contracts without GPU or network overhead. """ import pytest from fastapi.testclient import TestClient from app import fastapi_app as space_fastapi_app from studio_server import app as studio_fastapi_app @pytest.fixture def space_client(): headers = {} import os if os.environ.get("FLOW_API_KEY"): headers["Authorization"] = f"Bearer {os.environ['FLOW_API_KEY']}" return TestClient(space_fastapi_app, headers=headers) @pytest.fixture def studio_client(): headers = {} import os if os.environ.get("FLOW_API_KEY"): headers["Authorization"] = f"Bearer {os.environ['FLOW_API_KEY']}" return TestClient(studio_fastapi_app, headers=headers) def test_space_health_endpoint(space_client): """Verify Space /v1/health returns 200 and valid JSON schema.""" resp = space_client.get("/v1/health") assert resp.status_code == 200 data = resp.json() assert "service" in data assert "gpu" in data assert data["gpu"]["status"] == "healthy" def test_space_models_endpoint(space_client): """Verify Space /v1/models lists available model catalog.""" resp = space_client.get("/v1/models") assert resp.status_code == 200 data = resp.json() assert "data" in data assert len(data["data"]) > 0 def test_space_gpu_status_endpoint(space_client): """Verify Space /v1/gpu/status returns 200.""" resp = space_client.get("/v1/gpu/status") assert resp.status_code == 200 data = resp.json() assert "gpu" in data assert "total_vram_gb" in data["gpu"] def test_studio_stats_endpoint(studio_client): """Verify Studio server /api/stats returns 200 and statistics summary.""" resp = studio_client.get("/api/stats") assert resp.status_code == 200 data = resp.json() assert "personas_count" in data or "dictionary_entries" in data or "status" in data def test_studio_personas_endpoint(studio_client): """Verify Studio /api/personas lists registered personas.""" resp = studio_client.get("/api/personas") assert resp.status_code == 200 data = resp.json() assert "personas" in data assert len(data["personas"]) > 0 def test_studio_dialect_conversion_endpoint(studio_client): """Verify Studio /api/linguistics/convert endpoint returns translated dialect.""" payload = { "text": "Băieți, haideți să mergem puțin în oraș.", "level": 2 } resp = studio_client.post("/api/linguistics/convert", json=payload) assert resp.status_code == 200 data = resp.json() assert "converted_text" in data or "converted" in data or "dialect_text" in data def test_studio_media_route_aliases(studio_client): """Verify dual route aliases for frontend UI (/api/media/*).""" # 1. /api/media/stats r_stats = studio_client.get("/api/media/stats") assert r_stats.status_code == 200 assert "counts" in r_stats.json() assert "personas_count" in r_stats.json() # 2. /api/media/personas r_personas = studio_client.get("/api/media/personas") assert r_personas.status_code == 200 assert "personas" in r_personas.json() # 3. /api/media/dictionary r_dict = studio_client.get("/api/media/dictionary?page=1&limit=5") assert r_dict.status_code == 200 # 4. /api/media/datasets r_datasets = studio_client.get("/api/media/datasets") assert r_datasets.status_code == 200 assert "datasets" in r_datasets.json() # 5. /api/media/config r_config = studio_client.get("/api/media/config") assert r_config.status_code == 200 # 6. /api/media/vault r_vault = studio_client.get("/api/media/vault?type=ALL") assert r_vault.status_code == 200 assert "items" in r_vault.json() assert "assets" in r_vault.json() # 7. /api/media/vault/music & /api/media/vault/video r_vault_m = studio_client.get("/api/media/vault/music") assert r_vault_m.status_code == 200 assert "items" in r_vault_m.json() r_vault_v = studio_client.get("/api/media/vault/video") assert r_vault_v.status_code == 200 assert "items" in r_vault_v.json() # 8. /api/media/logs r_logs = studio_client.get("/api/media/logs") assert r_logs.status_code == 200 assert "logs" in r_logs.json() # 9. /api/media/dialect/convert r_dialect = studio_client.post("/api/media/dialect/convert", json={"text": "Salut prieteni", "level": 2}) assert r_dialect.status_code == 200 assert "dialect_text" in r_dialect.json() def test_space_models_catalog_includes_new_models(space_client): """Verify /v1/models includes petruhonk and Jackrong models.""" from app import resolve_model, list_gguf_files resp = space_client.get("/v1/models") assert resp.status_code == 200 data = resp.json() model_ids = [m["id"] for m in data["data"]] assert "Qwen3.8-9B-Distill-Heretic-Uncensored-Q8_0.gguf" in model_ids assert "DeepSeek-V4-Pro-Qwen3.5-9B-MTP-Q4_K_M.gguf" in model_ids assert "Qwen3.8-27B-TurboFCFusion-735-882-Here-Uncen-NEO-CODER-MAX-Q4_K_M.gguf" in model_ids assert "Wan-AI/Wan2.1-T2V-1.3B" in model_ids assert "MiniMaxAI/MiniMax-Music3" in model_ids # Verify alias resolution choices = list_gguf_files() assert resolve_model("premium", choices) == "Qwen3.8-9B-Q8_0.gguf" assert resolve_model("paid-premium", choices) == "Qwen3.8-9B-Q8_0.gguf" assert resolve_model("default", choices) == "Qwen3.8-9B-Q8_0.gguf" assert resolve_model("qwen-9b-distill-heretic", choices) == "Qwen3.8-9B-Distill-Heretic-Uncensored-Q8_0.gguf" assert resolve_model("deepseek-v4", choices) == "DeepSeek-V4-Pro-Qwen3.5-9B-MTP-Q4_K_M.gguf" assert resolve_model("deepseek-v4-pro", choices) == "DeepSeek-V4-Pro-Qwen3.5-9B-MTP-Q4_K_M.gguf" assert resolve_model("petruhonk/qwen3.8-9b-distill-uncensored-heretic", choices) == "Qwen3.8-9B-Distill-Heretic-Uncensored-Q8_0.gguf" assert resolve_model("jackrong/deepseek-v4-pro-qwen3.5-9b-mtp", choices) == "DeepSeek-V4-Pro-Qwen3.5-9B-MTP-Q4_K_M.gguf" assert resolve_model("qwen-turbo", choices) == "Qwen3.8-27B-TurboFCFusion-735-882-Here-Uncen-NEO-CODER-MAX-Q4_K_M.gguf" def test_studio_lyrics_compose_with_persona(studio_client): """Verify /api/media/lyrics/compose accepts persona_id and generates themed song sections.""" payload = { "topic": "Botanica vs Rîșcani", "genre": "Chișinău 808 Trap", "duration": 60, "persona_id": "ion_king_kong", "dialect_level": 3 } resp = studio_client.post("/api/media/lyrics/compose", json=payload) assert resp.status_code == 200 data = resp.json() assert "sections" in data assert len(data["sections"]) > 0 def test_studio_song_create_with_persona(studio_client, monkeypatch): """Verify /api/media/song/create passes persona_id and returns audio package. Policy: only MiniMax Music 3 is permitted for songs. Banned engines (voice_cloner, acestep, musicgen, synth808) must be rejected with HTTP 400. The MiniMax dispatch is mocked so the unit test stays offline (no ZeroGPU quota dependency). """ class _FakeDispatcher: def generate_music(self, *args, **kwargs): assert kwargs.get("engine") == "minimax_music3" return { "status": "success", "engine": "MiniMaxAI/MiniMax-Music3 (ZeroGPU)", "download_url": "/api/media/audio/stream/test_track.mp3", "audio_url": "/api/media/audio/stream/test_track.mp3", "filename": "test_track.mp3", "duration_seconds": 30, } def generate_image(self, *args, **kwargs): return {"image_url": "/api/media/images/test_cover.png", "filename": "test_cover.png"} from engine import media_creator as _mc_mod _RealMediaCreator = _mc_mod.MediaCreator def _fake_media_creator(*args, **kwargs): mc = _RealMediaCreator.__new__(_RealMediaCreator) mc.dispatcher = _FakeDispatcher() return mc monkeypatch.setattr( _mc_mod, "MediaCreator", _fake_media_creator, raising=False ) payload = { "topic": "King Kong Chișinău", "genre": "Chișinău 808 Trap", "duration": 30, "audio_engine": "minimax_music3", "persona_id": "ion_king_kong" } resp = studio_client.post("/api/media/song/create", json=payload) assert resp.status_code == 200 data = resp.json() assert "id" in data assert data["persona_id"] == "ion_king_kong" assert "audio_url" in data assert "full_lyrics" in data def test_space_audio_jobs_endpoint(space_client): """Verify /v1/audio/jobs creates background audio jobs and returns 202.""" payload = { "model": "MiniMaxAI/MiniMax-Music3", "instructions": "Moldovan balkan urban beat", "input": "Chișinău noaptea are ritmul lui", "duration": 60 } resp = space_client.post("/v1/audio/jobs", json=payload) assert resp.status_code == 202 data = resp.json() assert "job_id" in data assert data["status"] in ("queued", "processing") assert f"/v1/audio/jobs/{data['job_id']}" == data["check_status_url"] # Poll status poll_resp = space_client.get(data["check_status_url"]) assert poll_resp.status_code == 200 poll_data = poll_resp.json() assert poll_data["job_id"] == data["job_id"] # Also test /v1/audio/speech with ?async=true async_resp = space_client.post("/v1/audio/speech?async=true", json=payload) assert async_resp.status_code == 202 assert "job_id" in async_resp.json()