309 lines
15 KiB
Python
309 lines
15 KiB
Python
import threading
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from dashboard import plugin_api as api
|
|
|
|
|
|
class ValidationHarnessTests(unittest.TestCase):
|
|
def test_legacy_multiple_models_map_to_primary_and_validators(self):
|
|
body = api.ChatRequest(models=["primary", "validator-a", "validator-b"])
|
|
with patch.object(api, "_require_installed_model", side_effect=lambda name: name):
|
|
primary, validators, harness = api._harness_models(body)
|
|
self.assertEqual(primary, "primary")
|
|
self.assertEqual(validators, ["validator-a", "validator-b"])
|
|
self.assertTrue(harness)
|
|
|
|
def test_sqlite_is_default_even_when_postgres_is_installed(self):
|
|
with patch.object(api, "_storage_config", return_value={"backend": "sqlite"}), patch.object(
|
|
api, "_postgres_configured", return_value=True
|
|
):
|
|
self.assertEqual(api._storage_backend(), "sqlite")
|
|
self.assertFalse(api._postgres_enabled())
|
|
|
|
def test_harness_requires_one_distinct_validator(self):
|
|
body = api.ChatRequest(primary_model="primary", validator_models=[], harness=True)
|
|
with patch.object(api, "_require_installed_model", side_effect=lambda name: name):
|
|
with self.assertRaises(api.HTTPException) as context:
|
|
api._harness_models(body)
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
|
|
def test_harness_accepts_one_validator(self):
|
|
body = api.ChatRequest(primary_model="primary", validator_models=["validator-a"], harness=True)
|
|
with patch.object(api, "_require_installed_model", side_effect=lambda name: name):
|
|
primary, validators, harness = api._harness_models(body)
|
|
self.assertEqual(primary, "primary")
|
|
self.assertEqual(validators, ["validator-a"])
|
|
self.assertTrue(harness)
|
|
|
|
def test_chat_route_queues_server_owned_job(self):
|
|
body = api.ChatRequest(primary_model="primary", message="Queue this request")
|
|
fake_job = {
|
|
"request_id": "request",
|
|
"conversation_id": "conversation",
|
|
"status": "queued",
|
|
"mode": "direct",
|
|
"primary_model": "primary",
|
|
"validator_models_json": "[]",
|
|
"result_json": "{}",
|
|
"error": "",
|
|
"updated_at": 1.0,
|
|
}
|
|
with patch.object(api, "_harness_models", return_value=("primary", [], False)), patch.object(
|
|
api, "_get_chat_job", side_effect=[None, fake_job]
|
|
), patch.object(api, "_ensure_conversation"), patch.object(api, "_persist_message"), patch.object(
|
|
api, "_chat_state"
|
|
), patch.object(api, "_create_chat_job"), patch.object(api, "_submit_chat_job"):
|
|
response = api.chat(body)
|
|
self.assertEqual(response["status"], "queued")
|
|
self.assertFalse(response["done"])
|
|
self.assertEqual(response["mode"], "direct")
|
|
|
|
def test_job_status_includes_durable_heartbeat_fields(self):
|
|
job = {
|
|
"request_id": "request",
|
|
"conversation_id": "conversation",
|
|
"status": "running",
|
|
"mode": "direct",
|
|
"primary_model": "primary",
|
|
"validator_models_json": "[]",
|
|
"result_json": "{}",
|
|
"error": "",
|
|
"attempt": 2,
|
|
"started_at": 100.0,
|
|
"heartbeat_at": 110.0,
|
|
"finished_at": None,
|
|
"updated_at": 110.0,
|
|
}
|
|
with patch.object(api.time, "time", return_value=112.5):
|
|
response = api._job_status_response(job)
|
|
self.assertFalse(response["done"])
|
|
self.assertEqual(response["attempt"], 2)
|
|
self.assertEqual(response["heartbeat_at"], 110.0)
|
|
self.assertEqual(response["heartbeat_age"], 2.5)
|
|
|
|
def test_active_jobs_route_returns_server_owned_jobs(self):
|
|
active = [{"request_id": "request", "status": "running"}]
|
|
with patch.object(api, "_list_chat_jobs", return_value=active) as listed:
|
|
response = api.chat_jobs(active=True, conversation_id="conversation", limit=20)
|
|
self.assertEqual(response, {"jobs": active})
|
|
listed.assert_called_once_with(active_only=True, conversation_id="conversation", limit=20)
|
|
|
|
def test_primary_draft_validators_and_primary_compilation_produce_one_answer(self):
|
|
body = api.ChatRequest(
|
|
primary_model="primary",
|
|
validator_models=["validator-a", "validator-b"],
|
|
harness=True,
|
|
message="What is the verified answer?",
|
|
)
|
|
calls = []
|
|
|
|
def fake_stream(payload, request_id, cancel_event=None, parent_id=None):
|
|
model = payload["model"]
|
|
prompt = payload["messages"][-1]["content"]
|
|
calls.append((model, prompt, parent_id))
|
|
if model == "primary" and "VALIDATION REPORTS:" not in prompt:
|
|
content = "primary draft"
|
|
elif model.startswith("validator"):
|
|
content = f"{model} found no material issue"
|
|
else:
|
|
content = "one compiled final answer"
|
|
return {"message": {"role": "assistant", "content": content}, "done": True}
|
|
|
|
def fake_metric(conversation_id, request_id, model, state, status=None, error=""):
|
|
return {"request_id": request_id, "model": model, "status": status}
|
|
|
|
with patch.object(api, "_require_installed_model", side_effect=lambda name: name), patch.object(
|
|
api, "_stream_chat_request", side_effect=fake_stream
|
|
), patch.object(api, "_persist_metric", side_effect=fake_metric), patch.object(
|
|
api, "_persist_chat_stage"
|
|
), patch.object(api, "_persist_chat_event"):
|
|
final, metrics, reports = api._run_validation_harness(
|
|
body,
|
|
"root-request",
|
|
"conversation",
|
|
"primary",
|
|
["validator-a", "validator-b"],
|
|
threading.Event(),
|
|
[],
|
|
)
|
|
|
|
self.assertEqual(final, "one compiled final answer")
|
|
self.assertEqual([item["model"] for item in reports], ["validator-a", "validator-b"])
|
|
self.assertEqual(len(metrics), 4)
|
|
self.assertEqual(len(calls), 4)
|
|
self.assertEqual(calls[0][0], "primary")
|
|
self.assertTrue(all(call[2] == "root-request" for call in calls))
|
|
compiler_prompt = calls[-1][1]
|
|
self.assertIn("PRIMARY DRAFT:", compiler_prompt)
|
|
self.assertIn("VALIDATOR 1", compiler_prompt)
|
|
self.assertIn("VALIDATOR 2", compiler_prompt)
|
|
self.assertNotIn("validator-a found no material issue\n\nvalidator-b found no material issue", final)
|
|
def test_validation_report_is_retried_and_never_returned_as_final_answer(self):
|
|
body = api.ChatRequest(primary_model="primary", validator_models=["validator"], harness=True, message="Write the requested result")
|
|
calls = []
|
|
|
|
def fake_stream(payload, request_id, cancel_event=None, parent_id=None):
|
|
model = payload["model"]
|
|
prompt = payload["messages"][-1]["content"]
|
|
calls.append((model, prompt))
|
|
if model == "validator":
|
|
content = "The draft needs a stronger ending."
|
|
elif "IMPORTANT FINALIZATION RETRY" in prompt:
|
|
content = "refined final answer"
|
|
elif "VALIDATION REPORTS:" in prompt:
|
|
content = "# Story Validation Report\n## Narrative Structure\nRating: 8/10"
|
|
else:
|
|
content = "primary draft"
|
|
return {"message": {"role": "assistant", "content": content}, "done": True}
|
|
|
|
def fake_metric(conversation_id, request_id, model, state, status=None, error=""):
|
|
return {"request_id": request_id, "model": model, "status": status}
|
|
|
|
with patch.object(api, "_require_installed_model", side_effect=lambda name: name), patch.object(
|
|
api, "_stream_chat_request", side_effect=fake_stream
|
|
), patch.object(api, "_persist_metric", side_effect=fake_metric), patch.object(
|
|
api, "_persist_chat_stage"
|
|
), patch.object(api, "_persist_chat_event"), patch.object(api, "_chat_state"):
|
|
final, metrics, reports = api._run_validation_harness(
|
|
body, "root-request", "conversation", "primary", ["validator"], threading.Event(), []
|
|
)
|
|
|
|
self.assertEqual(final, "refined final answer")
|
|
self.assertNotIn("Validation Report", final)
|
|
self.assertEqual(len(reports), 1)
|
|
self.assertEqual(len(metrics), 4)
|
|
self.assertTrue(any("IMPORTANT FINALIZATION RETRY" in prompt for _, prompt in calls))
|
|
|
|
def test_status_omits_full_catalog_by_default(self):
|
|
with patch.object(api, "_local_tags", return_value=[]), patch.object(api, "_local_ps", return_value=[]), patch.object(
|
|
api, "_ensure_catalog", return_value={"models": [], "families": {}, "source": "test"}
|
|
), patch.object(api, "_ollama_version", return_value="test"), patch.object(
|
|
api, "_connection_snapshot", return_value=[]
|
|
), patch.object(api, "_disk_snapshot", return_value={}), patch.object(api, "_running_in_container", return_value=False), patch.object(
|
|
api, "_next_refresh", return_value=None
|
|
):
|
|
response = api.status()
|
|
self.assertNotIn("catalog", response)
|
|
self.assertNotIn("catalog_all", response)
|
|
self.assertEqual(response["catalog_count"], 0)
|
|
|
|
def test_conversation_history_excludes_current_request(self):
|
|
class Cursor:
|
|
def fetchall(self):
|
|
return [
|
|
{"role": "user", "content": "prior question", "request_id": "prior"},
|
|
{"role": "assistant", "content": "prior answer", "request_id": "prior"},
|
|
{"role": "user", "content": "current question", "request_id": "current"},
|
|
]
|
|
|
|
class Database:
|
|
def execute(self, statement, parameters=()):
|
|
self.parameters = parameters
|
|
return Cursor()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
database = Database()
|
|
with patch.object(api, "_chat_db", return_value=database):
|
|
history = api._conversation_history("conversation", exclude_request_id="current")
|
|
self.assertEqual(history, [{"role": "user", "content": "prior question"}, {"role": "assistant", "content": "prior answer"}])
|
|
|
|
def test_catalog_endpoint_returns_filtered_page(self):
|
|
catalog_rows = [
|
|
{"name": "qwen3:8b", "family": "qwen3", "capabilities": ["completion"], "is_moe": False},
|
|
{"name": "llama3:8b", "family": "llama3", "capabilities": ["completion"], "is_moe": False},
|
|
]
|
|
snapshot = {
|
|
"catalog": catalog_rows,
|
|
"catalog_all": catalog_rows,
|
|
"catalog_count": 2,
|
|
"catalog_filter_options": {"capabilities": ["completion"]},
|
|
"catalog_source": "test",
|
|
"catalog_updated_at": None,
|
|
"catalog_error": None,
|
|
}
|
|
with patch.object(api, "status", return_value=snapshot):
|
|
response = api.catalog(q="qwen", page=1, page_size=1, recent_only=False)
|
|
self.assertEqual(response["total"], 1)
|
|
self.assertEqual(response["catalog"][0]["name"], "qwen3:8b")
|
|
self.assertFalse(response["has_more"])
|
|
|
|
def test_chat_route_persists_empty_browser_history_in_job(self):
|
|
body = api.ChatRequest(primary_model="primary", message="Canonical only", history=[{"role": "user", "content": "stale"}])
|
|
fake_job = {
|
|
"request_id": "request",
|
|
"conversation_id": "conversation",
|
|
"status": "queued",
|
|
"mode": "direct",
|
|
"primary_model": "primary",
|
|
"validator_models_json": "[]",
|
|
"result_json": "{}",
|
|
"error": "",
|
|
"updated_at": 1.0,
|
|
}
|
|
created = []
|
|
with patch.object(api, "_harness_models", return_value=("primary", [], False)), patch.object(
|
|
api, "_get_chat_job", side_effect=[None, fake_job]
|
|
), patch.object(api, "_ensure_conversation"), patch.object(api, "_persist_message"), patch.object(
|
|
api, "_chat_state"
|
|
), patch.object(api, "_create_chat_job", side_effect=lambda request_id, conversation_id, value, primary, validators, harness: created.append(value)), patch.object(
|
|
api, "_submit_chat_job"
|
|
):
|
|
api.chat(body)
|
|
self.assertEqual(len(created), 1)
|
|
self.assertEqual(created[0].history, [])
|
|
|
|
def test_dashboard_restores_complete_bottom_performance_graph_set(self):
|
|
bundle = Path(__file__).parents[1].joinpath("dashboard", "dist", "index.js").read_text(encoding="utf-8")
|
|
for label in ("CPU usage", "CPU load", "System memory", "GPU usage", "GPU VRAM", "Disk usage", "Swap usage", "Ollama model weights", "Resident models"):
|
|
self.assertIn('title: "' + label + '"', bundle)
|
|
self.assertIn("function PerformanceGraphs", bundle)
|
|
self.assertIn("ollama-operations-section", bundle)
|
|
self.assertIn("historical performance graphs are shown in Operations / Performance below", bundle)
|
|
self.assertIn("setInterval(pollRuntime, 1000)", bundle)
|
|
|
|
def test_performance_sample_uses_minute_bucket_and_runtime_values(self):
|
|
sample = api._performance_sample({
|
|
"captured_at": 1700000061,
|
|
"memory_total_bytes": 200,
|
|
"memory_used_bytes": 100,
|
|
"swap_total_bytes": 100,
|
|
"swap_used_bytes": 25,
|
|
"ollama_model_bytes": 1024 ** 3,
|
|
"model_memory": [{"name": "model"}],
|
|
"cpu": {"usage_percent": 12.5, "load_average": [1.25]},
|
|
"gpu": {"utilization_percent": 40, "gpus": [{"total_bytes": 100, "used_bytes": 50}]},
|
|
"disk": {"used_percent": 33.3},
|
|
})
|
|
self.assertEqual(sample["captured_at"], 1700000040)
|
|
self.assertEqual(sample["memory_used_percent"], 50.0)
|
|
self.assertEqual(sample["gpu_vram_used_percent"], 50.0)
|
|
self.assertEqual(sample["resident_model_count"], 1)
|
|
|
|
def test_performance_history_accepts_only_requested_windows(self):
|
|
for hours in (1, 6, 9, 12, 24):
|
|
self.assertEqual(api._validate_performance_hours(hours), hours)
|
|
with self.assertRaises(api.HTTPException) as context:
|
|
api._validate_performance_hours(2)
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
|
|
def test_performance_history_deduplicates_minute_buckets(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(api, "_home", return_value=Path(directory)):
|
|
base = int(time.time()) - 120
|
|
runtime = {"captured_at": base + 1, "memory_total_bytes": 1, "memory_used_bytes": 1, "cpu": {}, "gpu": {}, "disk": {}, "model_memory": []}
|
|
first = api._record_performance_sample(runtime)
|
|
runtime["captured_at"] = base + 2
|
|
second = api._record_performance_sample(runtime)
|
|
self.assertEqual(len(first), 1)
|
|
self.assertEqual(len(second), 1)
|
|
self.assertEqual(second[0]["captured_at"], ((base + 2) // 60) * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|