test(routes): cover settings library endpoints

This commit is contained in:
pixelpaws
2025-10-04 08:38:59 +08:00
parent 65a0c00e33
commit b69c24ae14
4 changed files with 221 additions and 0 deletions

View File

@@ -376,5 +376,21 @@ class Config:
len(self.embeddings_roots or []),
)
def get_library_registry_snapshot(self) -> Dict[str, object]:
"""Return the current library registry and active library name."""
try:
from .services.settings_manager import settings as settings_service
libraries = settings_service.get_libraries()
active_library = settings_service.get_active_library_name()
return {
"active_library": active_library,
"libraries": libraries,
}
except Exception as exc: # pragma: no cover - defensive logging
logger.debug("Failed to collect library registry snapshot: %s", exc)
return {"active_library": "", "libraries": {}}
# Global config instance
config = Config()

View File

@@ -166,6 +166,24 @@ class SettingsHandler:
self._metadata_provider_updater = metadata_provider_updater
self._downloader_factory = downloader_factory
async def get_libraries(self, request: web.Request) -> web.Response:
"""Return the registered libraries and the active selection."""
try:
snapshot = config.get_library_registry_snapshot()
libraries = snapshot.get("libraries", {})
active_library = snapshot.get("active_library", "")
return web.json_response(
{
"success": True,
"libraries": libraries,
"active_library": active_library,
}
)
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Error getting library registry: %s", exc, exc_info=True)
return web.json_response({"success": False, "error": str(exc)}, status=500)
async def get_settings(self, request: web.Request) -> web.Response:
try:
response_data = {}
@@ -178,6 +196,41 @@ class SettingsHandler:
logger.error("Error getting settings: %s", exc, exc_info=True)
return web.json_response({"success": False, "error": str(exc)}, status=500)
async def activate_library(self, request: web.Request) -> web.Response:
"""Activate the selected library."""
try:
data = await request.json()
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Error parsing activate library request: %s", exc, exc_info=True)
return web.json_response({"success": False, "error": "Invalid JSON payload"}, status=400)
library_name = data.get("library") or data.get("library_name")
if not isinstance(library_name, str) or not library_name.strip():
return web.json_response(
{"success": False, "error": "Library name is required"}, status=400
)
try:
normalized_name = library_name.strip()
self._settings.activate_library(normalized_name)
snapshot = config.get_library_registry_snapshot()
libraries = snapshot.get("libraries", {})
active_library = snapshot.get("active_library", "")
return web.json_response(
{
"success": True,
"active_library": active_library,
"libraries": libraries,
}
)
except KeyError as exc:
logger.debug("Attempted to activate unknown library '%s'", library_name)
return web.json_response({"success": False, "error": str(exc)}, status=404)
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Error activating library '%s': %s", library_name, exc, exc_info=True)
return web.json_response({"success": False, "error": str(exc)}, status=500)
async def update_settings(self, request: web.Request) -> web.Response:
try:
data = await request.json()
@@ -731,6 +784,8 @@ class MiscHandlerSet:
"health_check": self.health.health_check,
"get_settings": self.settings.get_settings,
"update_settings": self.settings.update_settings,
"get_settings_libraries": self.settings.get_libraries,
"activate_library": self.settings.activate_library,
"update_usage_stats": self.usage_stats.update_usage_stats,
"get_usage_stats": self.usage_stats.get_usage_stats,
"update_lora_code": self.lora_code.update_lora_code,

View File

@@ -22,6 +22,8 @@ class RouteDefinition:
MISC_ROUTE_DEFINITIONS: tuple[RouteDefinition, ...] = (
RouteDefinition("GET", "/api/lm/settings", "get_settings"),
RouteDefinition("POST", "/api/lm/settings", "update_settings"),
RouteDefinition("GET", "/api/lm/settings/libraries", "get_settings_libraries"),
RouteDefinition("POST", "/api/lm/settings/libraries/activate", "activate_library"),
RouteDefinition("GET", "/api/lm/health-check", "health_check"),
RouteDefinition("POST", "/api/lm/open-file-location", "open_file_location"),
RouteDefinition("POST", "/api/lm/update-usage-stats", "update_usage_stats"),

View File

@@ -0,0 +1,148 @@
import json
import pytest
from py.config import config
from py.routes.handlers.misc_handlers import SettingsHandler
class FakeRequest:
def __init__(self, *, json_data=None):
self._json_data = json_data or {}
async def json(self):
return self._json_data
class DummySettings:
def __init__(self):
self.activated = None
self.should_raise = None
def activate_library(self, name):
if self.should_raise:
raise self.should_raise
self.activated = name
class DummyDownloader:
async def refresh_session(self): # pragma: no cover - helper
return None
async def dummy_downloader_factory(): # pragma: no cover - helper
return DummyDownloader()
async def noop_async(*_args, **_kwargs): # pragma: no cover - helper
return None
@pytest.fixture
def handler():
return SettingsHandler(
settings_service=DummySettings(),
metadata_provider_updater=noop_async,
downloader_factory=dummy_downloader_factory,
)
@pytest.mark.asyncio
async def test_get_libraries_returns_registry(monkeypatch, handler):
registry = {"libraries": {"default": {"name": "Default"}}, "active_library": "default"}
monkeypatch.setattr(config, "get_library_registry_snapshot", lambda: registry)
response = await handler.get_libraries(FakeRequest())
payload = json.loads(response.text)
assert response.status == 200
assert payload == {
"success": True,
"libraries": registry["libraries"],
"active_library": "default",
}
@pytest.mark.asyncio
async def test_get_libraries_handles_errors(monkeypatch, handler):
def boom():
raise RuntimeError("exploded")
monkeypatch.setattr(config, "get_library_registry_snapshot", boom)
response = await handler.get_libraries(FakeRequest())
payload = json.loads(response.text)
assert response.status == 500
assert payload["success"] is False
assert payload["error"] == "exploded"
@pytest.mark.asyncio
async def test_activate_library_success(monkeypatch):
dummy_settings = DummySettings()
handler = SettingsHandler(
settings_service=dummy_settings,
metadata_provider_updater=noop_async,
downloader_factory=dummy_downloader_factory,
)
registry = {"libraries": {"alpha": {"name": "Alpha"}}, "active_library": "alpha"}
monkeypatch.setattr(config, "get_library_registry_snapshot", lambda: registry)
response = await handler.activate_library(FakeRequest(json_data={"library": "alpha"}))
payload = json.loads(response.text)
assert response.status == 200
assert payload == {
"success": True,
"active_library": "alpha",
"libraries": registry["libraries"],
}
assert dummy_settings.activated == "alpha"
@pytest.mark.asyncio
async def test_activate_library_requires_name(handler):
response = await handler.activate_library(FakeRequest(json_data={}))
payload = json.loads(response.text)
assert response.status == 400
assert payload["success"] is False
assert payload["error"] == "Library name is required"
@pytest.mark.asyncio
async def test_activate_library_unknown_returns_404(monkeypatch):
dummy_settings = DummySettings()
dummy_settings.should_raise = KeyError("Unknown library")
handler = SettingsHandler(
settings_service=dummy_settings,
metadata_provider_updater=noop_async,
downloader_factory=dummy_downloader_factory,
)
response = await handler.activate_library(FakeRequest(json_data={"library": "ghost"}))
payload = json.loads(response.text)
assert response.status == 404
assert payload["success"] is False
assert payload["error"] == "'Unknown library'"
@pytest.mark.asyncio
async def test_activate_library_unexpected_error_returns_500(monkeypatch):
dummy_settings = DummySettings()
dummy_settings.should_raise = ValueError("bad things")
handler = SettingsHandler(
settings_service=dummy_settings,
metadata_provider_updater=noop_async,
downloader_factory=dummy_downloader_factory,
)
response = await handler.activate_library(FakeRequest(json_data={"library": "broken"}))
payload = json.loads(response.text)
assert response.status == 500
assert payload["success"] is False
assert payload["error"] == "bad things"