nearxos
6185b9b85a
Self-hosted, Docker-based agentic troubleshooting platform: FastAPI backend + LangGraph agent, Next.js UI, tiered LLM routing (local Ollama -> Gemini -> DeepSeek -> OpenRouter), MCP server manager, encrypted device credentials, RBAC, audit log, project-memory + Obsidian integrations, and editable troubleshooting decision rules tuned for the GeneseasX vessel stack. Co-authored-by: Cursor <cursoragent@cursor.com>
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""Tests for agent helpers that need no external services."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-unit-tests-only")
|
|
|
|
from app.agent.graph import _parse_json, _report_findings, _scope_checked # noqa: E402
|
|
from app.agent.playbooks import playbook_for # noqa: E402
|
|
from app.models.enums import DeviceType # noqa: E402
|
|
from app.services.llm_usage import estimate_cost, merge_run_history # noqa: E402
|
|
from app.services.model_router import plan_reasoning_tiers # noqa: E402
|
|
from app.services.model_config import LlmRoutingConfig # noqa: E402
|
|
|
|
|
|
def test_parse_json_plain():
|
|
assert _parse_json('{"a": 1}') == {"a": 1}
|
|
|
|
|
|
def test_parse_json_fenced():
|
|
text = """Here is the result:\n```json\n{"severity": "high", "plan": ["a"]}\n```"""
|
|
out = _parse_json(text)
|
|
assert out["severity"] == "high"
|
|
assert out["plan"] == ["a"]
|
|
|
|
|
|
def test_parse_json_garbage():
|
|
assert _parse_json("no json here") == {}
|
|
|
|
|
|
def test_playbook_lookup():
|
|
pb = playbook_for(DeviceType.pfsense, None)
|
|
assert pb is not None
|
|
assert pb["mcp_server"] == "pfsense-mcp"
|
|
assert pb["diagnostics"]
|
|
|
|
|
|
def test_playbook_override():
|
|
pb = playbook_for(DeviceType.debian, "custom-ssh-mcp")
|
|
assert pb["mcp_server"] == "custom-ssh-mcp"
|
|
|
|
|
|
def test_estimate_cost_local():
|
|
assert estimate_cost("qwen2.5:7b-instruct", "ollama", 1000, 500) == 0.0
|
|
|
|
|
|
def test_merge_run_history():
|
|
prior = [{"run_number": 1, "run_cost_usd": 0.001, "llm_usage": []}]
|
|
report = {"summary": "still broken", "resolved": False, "devices_checked": ["proxmox"]}
|
|
llm = {"llm_usage": [{"step": "triage", "cost_usd": 0.0}], "run_cost_usd": 0.002}
|
|
merged = merge_run_history(prior, 2, report, llm)
|
|
assert merged["run_number"] == 2
|
|
assert len(merged["runs"]) == 2
|
|
assert merged["total_cost_usd"] == 0.003
|
|
|
|
|
|
def test_plan_reasoning_local_first():
|
|
tiers, reason = plan_reasoning_tiers(
|
|
{"severity": "low", "needs_cloud_reasoning": False, "recommended_tier": "local"},
|
|
[],
|
|
cfg=LlmRoutingConfig(),
|
|
)
|
|
assert tiers == ["local"]
|
|
assert "local" in reason.lower()
|
|
|
|
|
|
def test_plan_reasoning_escalates_critical():
|
|
tiers, _ = plan_reasoning_tiers(
|
|
{"severity": "critical", "needs_cloud_reasoning": True, "recommended_tier": "premium"},
|
|
[{"ok": False}, {"ok": False}],
|
|
cfg=LlmRoutingConfig(),
|
|
)
|
|
assert tiers == ["local", "economy", "premium"]
|
|
|
|
|
|
def test_scope_checked_tracks_tools_per_device():
|
|
devices = [{"catalog_key": "proxmox", "name": "PVE"}, {"catalog_key": "pfsense", "name": "Firewall"}]
|
|
diagnostics = [
|
|
{"device": "proxmox", "tool": "proxmox_list_qemu", "ok": True},
|
|
{"device": "pfsense", "tool": "pfsense_get_system_status", "ok": True},
|
|
]
|
|
scope = _scope_checked(devices, diagnostics)
|
|
assert scope[0]["device"] == "proxmox"
|
|
assert scope[0]["checks"] == ["proxmox_list_qemu"]
|
|
assert scope[1]["device"] == "pfsense"
|
|
|
|
|
|
def test_report_findings_does_not_invent_unchecked_devices():
|
|
findings = _report_findings(
|
|
{"summary": "done", "resolution": "No critical issue found."},
|
|
[{"device": "proxmox", "tool": "proxmox_list_qemu", "ok": True}],
|
|
["proxmox"],
|
|
)
|
|
assert "proxmox" in findings[0]["summary"]
|
|
assert "pfsense" not in findings[0]["summary"].lower()
|
|
|