Model upgrade: - qwen2.5:14b → qwen3.5:latest across config, tools, and docs - Added qwen3.5 to multimodal model registry Self-hosted Gitea CI: - .gitea/workflows/tests.yml: lint + test jobs via act_runner - Unified Dockerfile: pre-baked deps from poetry.lock for fast CI - sitepackages=true in tox for ~2s dep resolution (was ~40s) - OLLAMA_URL set to dead port in CI to prevent real LLM calls Test isolation fixes: - Smoke test fixture mocks create_timmy (was hitting real Ollama) - WebSocket sends initial_state before joining broadcast pool (race fix) - Tests use settings.ollama_model/url instead of hardcoded values - skip_ci marker for Ollama-dependent tests, excluded in CI tox envs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
186 lines
6.4 KiB
Python
186 lines
6.4 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
# ── Index ─────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_index_returns_200(client):
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_index_contains_title(client):
|
|
response = client.get("/")
|
|
assert "MISSION CONTROL" in response.text
|
|
|
|
|
|
def test_index_contains_chat_interface(client):
|
|
response = client.get("/")
|
|
# Agent panel loads dynamically via HTMX; verify the trigger attribute is present
|
|
assert 'hx-get="/agents/default/panel"' in response.text
|
|
|
|
|
|
# ── Health ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_health_endpoint_ok(client):
|
|
with patch(
|
|
"dashboard.routes.health.check_ollama",
|
|
new_callable=AsyncMock,
|
|
return_value=True,
|
|
):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["services"]["ollama"] == "up"
|
|
assert "agents" in data
|
|
|
|
|
|
def test_health_endpoint_ollama_down(client):
|
|
with patch(
|
|
"dashboard.routes.health.check_ollama",
|
|
new_callable=AsyncMock,
|
|
return_value=False,
|
|
):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["services"]["ollama"] == "down"
|
|
|
|
|
|
def test_health_status_panel_ollama_up(client):
|
|
with patch(
|
|
"dashboard.routes.health.check_ollama",
|
|
new_callable=AsyncMock,
|
|
return_value=True,
|
|
):
|
|
response = client.get("/health/status")
|
|
assert response.status_code == 200
|
|
assert "UP" in response.text
|
|
|
|
|
|
def test_health_status_panel_ollama_down(client):
|
|
with patch(
|
|
"dashboard.routes.health.check_ollama",
|
|
new_callable=AsyncMock,
|
|
return_value=False,
|
|
):
|
|
response = client.get("/health/status")
|
|
assert response.status_code == 200
|
|
assert "DOWN" in response.text
|
|
|
|
|
|
# ── Agents ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_agents_list(client):
|
|
response = client.get("/agents")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "agents" in data
|
|
ids = [a["id"] for a in data["agents"]]
|
|
assert "default" in ids
|
|
|
|
|
|
def test_agents_list_metadata(client):
|
|
from config import settings
|
|
|
|
response = client.get("/agents")
|
|
agent = next(a for a in response.json()["agents"] if a["id"] == "default")
|
|
assert agent["name"] == "Agent"
|
|
assert agent["model"] == settings.ollama_model
|
|
assert agent["type"] == "local"
|
|
|
|
|
|
# ── Chat ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _mock_run(content="Operational and ready."):
|
|
"""Create a mock RunOutput for a completed run."""
|
|
run = MagicMock()
|
|
run.content = content
|
|
run.status = "COMPLETED"
|
|
run.active_requirements = []
|
|
return run
|
|
|
|
|
|
def test_chat_agent_success(client):
|
|
with patch("dashboard.routes.agents.chat_with_tools", return_value=_mock_run()):
|
|
response = client.post("/agents/default/chat", data={"message": "status?"})
|
|
|
|
assert response.status_code == 200
|
|
assert "status?" in response.text
|
|
assert "Operational" in response.text
|
|
|
|
|
|
def test_chat_agent_shows_user_message(client):
|
|
with patch("dashboard.routes.agents.chat_with_tools", return_value=_mock_run("Acknowledged.")):
|
|
response = client.post("/agents/default/chat", data={"message": "hello there"})
|
|
|
|
assert "hello there" in response.text
|
|
|
|
|
|
def test_chat_agent_ollama_offline(client):
|
|
# When Ollama is unreachable, chat shows the user message + error.
|
|
with patch(
|
|
"dashboard.routes.agents.chat_with_tools",
|
|
side_effect=Exception("Ollama unreachable"),
|
|
):
|
|
response = client.post("/agents/default/chat", data={"message": "ping"})
|
|
|
|
assert response.status_code == 200
|
|
assert "ping" in response.text
|
|
|
|
|
|
def test_chat_agent_requires_message(client):
|
|
response = client.post("/agents/default/chat", data={})
|
|
assert response.status_code == 422
|
|
|
|
|
|
# ── History ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_history_empty_shows_init_message(client):
|
|
response = client.get("/agents/default/history")
|
|
assert response.status_code == 200
|
|
assert "Mission Control initialized" in response.text
|
|
|
|
|
|
def test_history_records_user_and_agent_messages(client):
|
|
with patch(
|
|
"dashboard.routes.agents.chat_with_tools", return_value=_mock_run("I am operational.")
|
|
):
|
|
client.post("/agents/default/chat", data={"message": "status check"})
|
|
|
|
response = client.get("/agents/default/history")
|
|
assert "status check" in response.text
|
|
|
|
|
|
def test_history_records_error_when_offline(client):
|
|
with patch(
|
|
"dashboard.routes.agents.chat_with_tools",
|
|
side_effect=Exception("Ollama unreachable"),
|
|
):
|
|
client.post("/agents/default/chat", data={"message": "ping"})
|
|
|
|
response = client.get("/agents/default/history")
|
|
assert "ping" in response.text
|
|
|
|
|
|
def test_history_clear_resets_to_init_message(client):
|
|
with patch("dashboard.routes.agents.chat_with_tools", return_value=_mock_run("Acknowledged.")):
|
|
client.post("/agents/default/chat", data={"message": "hello"})
|
|
|
|
response = client.delete("/agents/default/history")
|
|
assert response.status_code == 200
|
|
assert "Mission Control initialized" in response.text
|
|
|
|
|
|
def test_history_empty_after_clear(client):
|
|
with patch("dashboard.routes.agents.chat_with_tools", return_value=_mock_run("OK.")):
|
|
client.post("/agents/default/chat", data={"message": "test"})
|
|
|
|
client.delete("/agents/default/history")
|
|
response = client.get("/agents/default/history")
|
|
assert "test" not in response.text
|
|
assert "Mission Control initialized" in response.text
|