#!/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()