c491779248
Issue #2: Manifest schema + Sidecar foundation - sidecar/manifest.py: YAML manifest loading and profile validation - sidecar/app.py: FastAPI sidecar service with /models/available, /models/status endpoints - Router GET /v1/models: proxies to sidecar, returns OpenAI-compatible model list - Tests: 12 manifest tests, 6 sidecar endpoint tests, 3 router tests (21 total) Issue #3: Sidecar model switch + Router request queue - Sidecar POST /models/switch: stops current llama-server, starts new one, polls for readiness - Switch lock prevents concurrent switches (threading.Lock for TestClient compatibility) - Router request queue: max 10 requests, 120s hard timeout, 429 when full - Router automatic model detection: extracts model from chat body, matches against sidecar status - Full proxy endpoint with Sidecar → Main PC routing and fallback chain - Tests: 5 sidecar switch tests, 4 queue tests, 3 router integration tests (12 total) Total: 33 tests, all passing
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""Tests for sidecar HTTP endpoints — Issue #2."""
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, mock_open
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sidecar.app import app as sidecar_app
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_sidecar_state():
|
|
"""Reset shared sidecar state between tests."""
|
|
import sidecar.app
|
|
old_active = sidecar.app._active_profile
|
|
old_proc = sidecar.app._llama_server_process
|
|
sidecar.app._active_profile = None
|
|
sidecar.app._llama_server_process = None
|
|
yield
|
|
sidecar.app._active_profile = old_active
|
|
sidecar.app._llama_server_process = old_proc
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_manifest(tmp_path):
|
|
"""Create a temporary manifest file for testing."""
|
|
manifest_file = tmp_path / "manifest.yaml"
|
|
manifest_file.write_text(
|
|
"- id: qwen-3-8b\n"
|
|
" name: \"Qwen 3 8B\"\n"
|
|
" model_path: /home/bigt/AI/llm/qwen/qwen3-8b-q4.gguf\n"
|
|
" flags:\n"
|
|
" n_ctx: 8192\n"
|
|
" n_gpu_layers: 35\n"
|
|
"- id: llama-4-maverick\n"
|
|
" name: \"Llama 4 Maverick\"\n"
|
|
" model_path: /home/bigt/AI/llm/llama4/llama4-maverick-q4.gguf\n"
|
|
)
|
|
return manifest_file
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_manifest):
|
|
"""Create a test client with a temporary manifest."""
|
|
with patch("sidecar.app.MANIFEST_PATH", str(tmp_manifest)):
|
|
yield TestClient(sidecar_app)
|
|
|
|
|
|
class TestModelsAvailable:
|
|
"""Tests for GET /models/available."""
|
|
|
|
def test_returns_profiles_from_manifest(self, client):
|
|
response = client.get("/models/available")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert data[0]["id"] == "qwen-3-8b"
|
|
assert data[0]["name"] == "Qwen 3 8B"
|
|
assert data[0]["model_path"] == "/home/bigt/AI/llm/qwen/qwen3-8b-q4.gguf"
|
|
assert "flags" in data[0]
|
|
|
|
def test_empty_manifest_returns_empty_list(self, tmp_path):
|
|
manifest_file = tmp_path / "empty.yaml"
|
|
manifest_file.write_text("[]\n")
|
|
with patch("sidecar.app.MANIFEST_PATH", str(manifest_file)):
|
|
client = TestClient(sidecar_app)
|
|
response = client.get("/models/available")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
def test_invalid_yaml_returns_500(self, tmp_path):
|
|
manifest_file = tmp_path / "invalid.yaml"
|
|
manifest_file.write_text("{{{{bad yaml:::\n")
|
|
with patch("sidecar.app.MANIFEST_PATH", str(manifest_file)):
|
|
client = TestClient(sidecar_app)
|
|
response = client.get("/models/available")
|
|
assert response.status_code == 500
|
|
body = response.json()
|
|
assert "detail" in body
|
|
|
|
def test_missing_file_returns_500(self):
|
|
with patch("sidecar.app.MANIFEST_PATH", "/tmp/does_not_exist_12345.yaml"):
|
|
client = TestClient(sidecar_app)
|
|
response = client.get("/models/available")
|
|
assert response.status_code == 500
|
|
body = response.json()
|
|
assert "detail" in body
|
|
|
|
def test_each_profile_has_required_fields(self, client):
|
|
response = client.get("/models/available")
|
|
profiles = response.json()
|
|
for p in profiles:
|
|
assert "id" in p
|
|
assert "name" in p
|
|
assert "model_path" in p
|
|
assert "flags" in p
|
|
|
|
|
|
class TestModelsStatus:
|
|
"""Tests for GET /models/status."""
|
|
|
|
def test_returns_inactive_status(self, client):
|
|
response = client.get("/models/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["active_profile"] is None
|
|
assert data["llama_server_running"] is False
|