box/tests/test_kpi.py

#!/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 (
    calculate_efficiency,
    check_node_routes,
    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"])


if __name__ == "__main__":
    unittest.main()