operator
206 lines
7.2 KiB
Python
206 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""test_kpi.py — Unit tests for NetVM Fleet KPI, spend monitor and runtime preservation engine."""
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
|
|
|
import kpi
|
|
from kpi import (
|
|
auto_spawn_workers,
|
|
calculate_efficiency,
|
|
check_node_routes,
|
|
find_pending_work_for_node,
|
|
generate_preservation_advisory,
|
|
get_agent_dm_metrics,
|
|
get_agent_kpi,
|
|
get_live_advisory_block,
|
|
spawn_tmux_worker,
|
|
)
|
|
|
|
|
|
class TestKPIMetrics(unittest.TestCase):
|
|
def test_calculate_efficiency_high(self):
|
|
# High work units relative to burn
|
|
idx, rating = calculate_efficiency(
|
|
jobs_done=5,
|
|
subagents_active=2,
|
|
tmux_workers=2,
|
|
calls_verified=20,
|
|
weekly_used_pct=30,
|
|
uptime_hours=10.0,
|
|
)
|
|
self.assertGreaterEqual(idx, 3.0)
|
|
self.assertEqual(rating, "HIGH_EFFICIENCY")
|
|
|
|
def test_calculate_efficiency_vanity_idle(self):
|
|
# Long uptime but zero workers and zero jobs
|
|
idx, rating = calculate_efficiency(
|
|
jobs_done=0,
|
|
subagents_active=0,
|
|
tmux_workers=0,
|
|
calls_verified=0,
|
|
weekly_used_pct=10,
|
|
uptime_hours=5.0,
|
|
)
|
|
self.assertEqual(rating, "VANITY_IDLE")
|
|
|
|
def test_generate_preservation_advisory_critical_quota(self):
|
|
adv = generate_preservation_advisory(
|
|
node="646",
|
|
weekly_used_pct=100,
|
|
extra_tokens_remaining="0 tokens left",
|
|
subagents_active=0,
|
|
tmux_workers=0,
|
|
rating="VANITY_IDLE",
|
|
)
|
|
self.assertIn("CRITICAL: Quota exhausted", adv)
|
|
self.assertIn("box onboard start", adv)
|
|
|
|
def test_generate_preservation_advisory_high_quota_idle(self):
|
|
adv = generate_preservation_advisory(
|
|
node="pip",
|
|
weekly_used_pct=75,
|
|
extra_tokens_remaining="500M tokens left",
|
|
subagents_active=0,
|
|
tmux_workers=0,
|
|
rating="VANITY_IDLE",
|
|
)
|
|
self.assertIn("Cease prose chatter", adv)
|
|
self.assertIn("Spawn subagents", adv)
|
|
|
|
@patch("kpi.Path.exists")
|
|
@patch("approvals.get_node_pages")
|
|
def test_check_node_routes_online(self, mock_pages, mock_exists):
|
|
mock_exists.return_value = True
|
|
mock_pages.return_value = [{"title": "Chat"}]
|
|
st = check_node_routes("646")
|
|
self.assertEqual(st, "ONLINE")
|
|
|
|
@patch("kpi.Path.exists")
|
|
def test_check_node_routes_no_netns(self, mock_exists):
|
|
mock_exists.return_value = False
|
|
st = check_node_routes("646")
|
|
self.assertEqual(st, "NO_NETNS")
|
|
|
|
@patch("kpi.get_agent_dm_metrics")
|
|
@patch("kpi.get_agent_job_metrics")
|
|
@patch("kpi.get_agent_subagent_count")
|
|
@patch("kpi.get_agent_tmux_workers")
|
|
@patch("kpi.get_agent_uptime_hours")
|
|
@patch("kpi.check_node_routes")
|
|
def test_get_agent_kpi(
|
|
self,
|
|
mock_routes,
|
|
mock_uptime,
|
|
mock_tmux,
|
|
mock_sub,
|
|
mock_jobs,
|
|
mock_dm,
|
|
):
|
|
mock_dm.return_value = {"sent": 50, "verified": 40, "total_events": 100}
|
|
mock_jobs.return_value = {"assigned": 10, "completed": 4}
|
|
mock_sub.return_value = 2
|
|
mock_tmux.return_value = ["worker-1"]
|
|
mock_uptime.return_value = 8.5
|
|
mock_routes.return_value = "ONLINE"
|
|
|
|
usage_cache = {
|
|
"weekly_used_pct": 25,
|
|
"extra_tokens_remaining": "900M tokens left",
|
|
"is_blocked": False,
|
|
}
|
|
|
|
res = get_agent_kpi("dev", usage_cache=usage_cache)
|
|
self.assertEqual(res.node, "dev")
|
|
self.assertEqual(res.weekly_used_pct, 25)
|
|
self.assertEqual(res.calls_sent, 50)
|
|
self.assertEqual(res.calls_verified, 40)
|
|
self.assertEqual(res.jobs_completed, 4)
|
|
self.assertEqual(res.subagents_active, 2)
|
|
self.assertEqual(res.tmux_workers_active, 1)
|
|
self.assertEqual(res.route_status, "ONLINE")
|
|
|
|
@patch("kpi.get_agent_kpi")
|
|
def test_get_live_advisory_block(self, mock_kpi):
|
|
mock_kpi.return_value = MagicMock(
|
|
node="646",
|
|
weekly_used_pct=30,
|
|
extra_tokens_remaining="1B tokens left",
|
|
uptime_hours=12.0,
|
|
jobs_completed=5,
|
|
subagents_active=2,
|
|
tmux_workers_active=1,
|
|
efficiency_rating="HIGH_EFFICIENCY",
|
|
efficiency_index=4.5,
|
|
route_status="ONLINE",
|
|
preservation_advisory="Runtime healthy.",
|
|
)
|
|
block = get_live_advisory_block("646")
|
|
self.assertIn("BOX PERFORMANCE & RUNTIME ADVISORY", block)
|
|
self.assertIn("AGENT: @646", block)
|
|
self.assertIn("30% weekly limit used", block)
|
|
self.assertIn("HIGH_EFFICIENCY", block)
|
|
self.assertIn("Runtime healthy.", block)
|
|
|
|
@patch("subprocess.run")
|
|
@patch("subagent_tracker.register_session")
|
|
def test_spawn_tmux_worker(self, mock_register, mock_subproc):
|
|
mock_subproc.return_value = MagicMock(returncode=0)
|
|
res = spawn_tmux_worker("dev", "audit-sub", "python3 audit.py")
|
|
self.assertTrue(res["ok"])
|
|
self.assertEqual(res["node"], "dev")
|
|
self.assertEqual(res["session"], "dev-audit-sub")
|
|
mock_register.assert_called_once()
|
|
call_kwargs = mock_register.call_args[1]
|
|
self.assertEqual(call_kwargs["parent"], "dev")
|
|
self.assertIn("dev-audit-sub", call_kwargs["session_id"])
|
|
|
|
def test_find_pending_work_for_node(self):
|
|
work = find_pending_work_for_node("dev")
|
|
self.assertIsNotNone(work)
|
|
self.assertIn(work["type"], ("job", "swarm"))
|
|
if work["type"] == "job":
|
|
self.assertIn("dev", work["name"])
|
|
self.assertIn("job-dispatch.py", work["cmd"])
|
|
|
|
@patch("kpi.get_agent_tmux_workers")
|
|
def test_auto_spawn_workers_skip_active(self, mock_tmux):
|
|
mock_tmux.return_value = ["active-worker-1"]
|
|
res = auto_spawn_workers(nodes=["dev"], dry_run=True)
|
|
self.assertEqual(len(res), 1)
|
|
self.assertEqual(res[0]["action"], "skip")
|
|
self.assertIn("Active tmux worker already running", res[0]["reason"])
|
|
|
|
@patch("kpi.get_agent_tmux_workers")
|
|
def test_auto_spawn_workers_dry_run(self, mock_tmux):
|
|
mock_tmux.return_value = []
|
|
res = auto_spawn_workers(nodes=["dev"], dry_run=True)
|
|
self.assertEqual(len(res), 1)
|
|
self.assertEqual(res[0]["action"], "would_spawn")
|
|
self.assertIn("session", res[0])
|
|
self.assertIn("command", res[0])
|
|
|
|
@patch("kpi.spawn_tmux_worker")
|
|
@patch("invite_handler.send_loopback_notice")
|
|
@patch("kpi.get_agent_tmux_workers")
|
|
def test_auto_spawn_workers_live(self, mock_tmux, mock_notice, mock_spawn):
|
|
mock_tmux.return_value = []
|
|
mock_spawn.return_value = {"ok": True, "session": "worker-1", "node": "dev"}
|
|
mock_notice.return_value = True
|
|
|
|
res = auto_spawn_workers(nodes=["dev"], dry_run=False)
|
|
self.assertEqual(len(res), 1)
|
|
self.assertEqual(res[0]["action"], "spawned")
|
|
mock_spawn.assert_called_once()
|
|
mock_notice.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|