166 lines
6.5 KiB
Python
166 lines
6.5 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MODULE_PATH = ROOT / "platform/llama-dashboard/app.py"
|
|
|
|
|
|
class _Response:
|
|
status = 202
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self):
|
|
return b'{"status":"accepted"}'
|
|
|
|
|
|
class DashboardModeTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tempdir = tempfile.TemporaryDirectory()
|
|
with patch.dict(os.environ, {
|
|
"DASHBOARD_HISTORY_DB": str(Path(cls.tempdir.name) / "history.sqlite3"),
|
|
"DASHBOARD_BACKUP_DIR": str(Path(cls.tempdir.name) / "backups"),
|
|
"ROUTER_URL": "http://router.test:8081",
|
|
"ROUTER_API_KEY": "test-key",
|
|
}):
|
|
spec = importlib.util.spec_from_file_location("dashboard_app_test", MODULE_PATH)
|
|
cls.dashboard = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(cls.dashboard)
|
|
cls.backup_dir = Path(cls.tempdir.name) / "backups"
|
|
cls.backup_dir.mkdir()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.tempdir.cleanup()
|
|
|
|
def test_separation_mode_is_forwarded_to_router(self):
|
|
with patch.object(self.dashboard.urllib.request, "urlopen", return_value=_Response()) as urlopen:
|
|
status, body = self.dashboard.change_mode("separation")
|
|
|
|
self.assertEqual(status, 202)
|
|
self.assertEqual(body, {"status": "accepted"})
|
|
request = urlopen.call_args.args[0]
|
|
self.assertEqual(json.loads(request.data), {"mode": "separation"})
|
|
self.assertEqual(request.get_header("Authorization"), "Bearer test-key")
|
|
|
|
def test_voice_mode_is_forwarded_to_router(self):
|
|
with patch.object(self.dashboard.urllib.request, "urlopen", return_value=_Response()) as urlopen:
|
|
status, body = self.dashboard.change_mode("voice")
|
|
|
|
self.assertEqual(status, 202)
|
|
self.assertEqual(body, {"status": "accepted"})
|
|
request = urlopen.call_args.args[0]
|
|
self.assertEqual(json.loads(request.data), {"mode": "voice"})
|
|
|
|
def test_voice_change_mode_is_forwarded_to_router(self):
|
|
with patch.object(self.dashboard.urllib.request, "urlopen", return_value=_Response()) as urlopen:
|
|
status, body = self.dashboard.change_mode("voicechange")
|
|
|
|
self.assertEqual(status, 202)
|
|
self.assertEqual(body, {"status": "accepted"})
|
|
request = urlopen.call_args.args[0]
|
|
self.assertEqual(json.loads(request.data), {"mode": "voicechange"})
|
|
|
|
def test_applio_mode_is_forwarded_to_router(self):
|
|
with patch.object(self.dashboard.urllib.request, "urlopen", return_value=_Response()) as urlopen:
|
|
status, body = self.dashboard.change_mode("applio")
|
|
self.assertEqual(status, 202)
|
|
self.assertEqual(body, {"status": "accepted"})
|
|
self.assertEqual(json.loads(urlopen.call_args.args[0].data), {"mode": "applio"})
|
|
|
|
def test_unknown_mode_is_rejected_without_router_request(self):
|
|
with patch.object(self.dashboard.urllib.request, "urlopen") as urlopen:
|
|
status, body = self.dashboard.change_mode("unknown")
|
|
|
|
self.assertEqual(status, 400)
|
|
self.assertEqual(body, {"error": "invalid mode"})
|
|
urlopen.assert_not_called()
|
|
|
|
def test_dashboard_uses_one_status_poll_for_all_mode_labels(self):
|
|
html = self.dashboard.HTML
|
|
|
|
self.assertEqual(html.count("fetch('/api/status'"), 1)
|
|
self.assertNotIn("refreshVoiceChange", html)
|
|
self.assertIn("voicechange:'X-VC Voice Changer'", html)
|
|
self.assertIn("applio:'Applio / RVC'", html)
|
|
|
|
def test_dashboard_offers_both_applio_frontends(self):
|
|
html = self.dashboard.HTML
|
|
|
|
self.assertIn("Original Applio UI", html)
|
|
self.assertIn("Mikes Applio UI", html)
|
|
self.assertIn("http://192.168.1.212:8011/", html)
|
|
self.assertIn("http://192.168.1.212:8012/", html)
|
|
|
|
def test_dashboard_lists_only_portable_encrypted_backups(self):
|
|
valid = self.backup_dir / "athena-portable-2026-09-10T10-00-00Z.tar.zst.age"
|
|
valid.write_bytes(b"encrypted")
|
|
valid.with_name(valid.name + ".sha256").write_text(
|
|
"a" * 64 + " " + valid.name + "\n", encoding="utf-8"
|
|
)
|
|
(self.backup_dir / "unrelated.txt").write_text("ignore", encoding="utf-8")
|
|
|
|
backups = self.dashboard.backup_inventory()
|
|
|
|
self.assertEqual([item["name"] for item in backups], [valid.name])
|
|
self.assertEqual(backups[0]["sha256"], "a" * 64)
|
|
self.assertTrue(backups[0]["download_url"].startswith("/api/backups/download/"))
|
|
|
|
def test_slot_context_history_is_recorded_and_exposed(self):
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
store = self.dashboard.HistoryStore(Path(tempdir) / "slots.sqlite3")
|
|
store.record({
|
|
"timestamp": 1_789_000_000,
|
|
"router": {
|
|
"current_profile": "medium",
|
|
"upstream": {"model": "qwen"},
|
|
"llama_telemetry": {
|
|
"slots": [{
|
|
"id": 0,
|
|
"context_used": 40_000,
|
|
"n_ctx": 160_000,
|
|
"processing": True,
|
|
}],
|
|
},
|
|
},
|
|
"llama_runtime": {"pid": 123, "model_file": "qwen.gguf"},
|
|
"gpus": [],
|
|
})
|
|
|
|
result = store.query("all")
|
|
store._db.close()
|
|
|
|
self.assertEqual(len(result["slot_points"]), 1)
|
|
self.assertEqual(result["slot_points"][0]["slot_id"], 0)
|
|
self.assertEqual(result["slot_points"][0]["context_percent"], 25.0)
|
|
self.assertEqual(result["slot_points"][0]["busy_ratio"], 1.0)
|
|
|
|
def test_dashboard_renders_slot_context_history_controls(self):
|
|
self.assertIn('id="slotHistoryChart"', self.dashboard.HTML)
|
|
self.assertIn('id="slotHistoryRanges"', self.dashboard.HTML)
|
|
self.assertIn("drawSlotHistory(d.slot_points||[])", self.dashboard.HISTORY_JS)
|
|
|
|
def test_unified_slots_show_current_shared_capacity(self):
|
|
script = self.dashboard.FULL_JS
|
|
|
|
self.assertIn("function sharedSlotPool(slots,unified)", script)
|
|
self.assertIn("used+pool.free", script)
|
|
self.assertIn("lr.kv_unified===true", script)
|
|
self.assertIn("Gemeinsamer Pool:", script)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|