#!/usr/bin/env python3 """Tests for health_scanner.py and roadmap.py — 30+ tests.""" import json import os import sys import tempfile import unittest from datetime import datetime, timedelta from pathlib import Path from unittest.mock import patch, MagicMock # Add parent to path sys.path.insert(0, str(Path(__file__).parent)) from cortex.health_scanner import HealthScanner, Finding, _severity_rank, format_human, DEPRECATED_MODELS import cortex.roadmap as roadmap class TestSeverityRank(unittest.TestCase): def test_ordering(self): self.assertLess(_severity_rank("INFO"), _severity_rank("WARN")) self.assertLess(_severity_rank("WARN"), _severity_rank("CRITICAL")) def test_unknown(self): self.assertEqual(_severity_rank("UNKNOWN"), -1) class TestFinding(unittest.TestCase): def test_to_dict(self): f = Finding("agents", "WARN", "test title", "detail") d = f.to_dict() self.assertEqual(d["section"], "agents") self.assertEqual(d["severity"], "WARN") self.assertEqual(d["title"], "test title") def test_to_dict_no_detail(self): f = Finding("deps", "INFO", "ok") self.assertEqual(f.to_dict()["detail"], "") class TestHealthScanner(unittest.TestCase): def _make_scanner(self, config=None): with patch.object(HealthScanner, '_load_config') as mock_load: scanner = HealthScanner() scanner.config = config or {"agents": {"list": [], "defaults": {}}} scanner.findings = [] return scanner def test_no_agents(self): scanner = self._make_scanner({"agents": {"list": []}}) scanner.check_agents() self.assertTrue(any("No agents" in f.title for f in scanner.findings)) def test_missing_workspace(self): scanner = self._make_scanner({"agents": {"list": [ {"id": "test", "workspace": "/nonexistent/path/xyz", "model": {"primary": "anthropic/test"}} ]}}) scanner.check_agents() self.assertTrue(any("missing" in f.title.lower() for f in scanner.findings)) def test_valid_workspace(self): with tempfile.TemporaryDirectory() as td: scanner = self._make_scanner({"agents": {"list": [ {"id": "test", "workspace": td, "model": {"primary": "anthropic/test"}} ]}}) scanner.check_agents() self.assertTrue(any("OK" in f.title for f in scanner.findings)) @patch('cortex.health_scanner.urllib.request.urlopen') def test_ollama_check_reachable(self, mock_urlopen): mock_resp = MagicMock() mock_resp.read.return_value = json.dumps({"models": [{"name": "mymodel:latest"}]}).encode() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_urlopen.return_value = mock_resp scanner = self._make_scanner() scanner._check_ollama_model("test", "ollama-desktop/mymodel:latest") self.assertTrue(any("available" in f.title for f in scanner.findings)) @patch('cortex.health_scanner.urllib.request.urlopen') def test_ollama_check_unreachable(self, mock_urlopen): mock_urlopen.side_effect = Exception("connection refused") scanner = self._make_scanner() scanner._check_ollama_model("test", "ollama-desktop/model:latest") self.assertTrue(any("unreachable" in f.title.lower() for f in scanner.findings)) def test_memory_check(self): scanner = self._make_scanner() scanner._check_memory() self.assertTrue(any("Memory" in f.title for f in scanner.findings)) def test_disk_check(self): scanner = self._make_scanner() scanner._check_disk() self.assertTrue(any("Disk" in f.title for f in scanner.findings)) @patch('cortex.health_scanner.subprocess.run') def test_nats_active(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout="active") scanner = self._make_scanner() scanner._check_nats() self.assertTrue(any("NATS" in f.title and f.severity == "INFO" for f in scanner.findings)) @patch('cortex.health_scanner.subprocess.run') def test_nats_inactive(self, mock_run): mock_run.return_value = MagicMock(returncode=3, stdout="inactive") scanner = self._make_scanner() scanner._check_nats() self.assertTrue(any("NATS" in f.title and f.severity == "CRITICAL" for f in scanner.findings)) def test_deprecated_model_detection(self): scanner = self._make_scanner({"agents": { "list": [{"id": "test", "model": {"primary": "anthropic/claude-3-5-haiku-latest", "fallbacks": []}}], "defaults": {"model": {}, "models": {}, "heartbeat": {}} }}) scanner.check_config() self.assertTrue(any("EOL" in f.title for f in scanner.findings)) def test_run_all_sections(self): scanner = self._make_scanner({"agents": {"list": [], "defaults": {"model": {}, "models": {}, "heartbeat": {}}}}) report = scanner.run() self.assertIn("timestamp", report) self.assertIn("overall", report) self.assertIn("findings", report) def test_run_single_section(self): scanner = self._make_scanner({"agents": {"list": []}}) report = scanner.run(["agents"]) self.assertTrue(all(f["section"] == "agents" for f in report["findings"])) def test_format_human(self): report = { "timestamp": "2026-02-08T20:00:00", "overall": "WARN", "findings_count": {"INFO": 1, "WARN": 1, "CRITICAL": 0}, "findings": [ {"section": "agents", "severity": "WARN", "title": "test warn", "detail": ""}, {"section": "deps", "severity": "INFO", "title": "test info", "detail": ""}, ] } output = format_human(report) self.assertIn("WARN", output) self.assertIn("test warn", output) class TestRoadmap(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp() self.roadmap_file = Path(self.tmpdir) / "roadmap.json" self.seed_data = { "items": [ { "id": "item-001", "title": "Test task 1", "status": "open", "priority": "P0", "deadline": (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d"), "depends_on": [], "tags": ["infra"], "created": "2026-01-01T00:00:00", "updated": "2026-01-01T00:00:00", "notes": "" }, { "id": "item-002", "title": "Test task 2", "status": "in_progress", "priority": "P1", "deadline": (datetime.now() + timedelta(days=3)).strftime("%Y-%m-%d"), "depends_on": ["item-001"], "tags": ["business"], "created": "2026-01-01T00:00:00", "updated": "2026-01-01T00:00:00", "notes": "" }, { "id": "item-003", "title": "Done task", "status": "done", "priority": "P2", "deadline": "2026-01-01", "depends_on": [], "tags": ["infra"], "created": "2026-01-01T00:00:00", "updated": "2026-01-01T00:00:00", "notes": "" }, { "id": "item-004", "title": "Stale task", "status": "in_progress", "priority": "P2", "deadline": null if False else None, "depends_on": ["item-999"], "tags": [], "created": "2026-01-01T00:00:00", "updated": (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%dT%H:%M:%S"), "notes": "" } ] } self.roadmap_file.write_text(json.dumps(self.seed_data)) def tearDown(self): import shutil shutil.rmtree(self.tmpdir) def test_load(self): data = roadmap.load_roadmap(self.roadmap_file) self.assertEqual(len(data["items"]), 4) def test_load_nonexistent(self): data = roadmap.load_roadmap(Path(self.tmpdir) / "nope.json") self.assertEqual(data, {"items": []}) def test_save_and_reload(self): data = roadmap.load_roadmap(self.roadmap_file) data["items"].append({"id": "new", "title": "new"}) roadmap.save_roadmap(data, self.roadmap_file) reloaded = roadmap.load_roadmap(self.roadmap_file) self.assertEqual(len(reloaded["items"]), 5) def test_find_item_full_id(self): data = roadmap.load_roadmap(self.roadmap_file) item = roadmap.find_item(data, "item-001") self.assertIsNotNone(item) self.assertEqual(item["title"], "Test task 1") def test_find_item_prefix(self): data = roadmap.load_roadmap(self.roadmap_file) item = roadmap.find_item(data, "item-00") self.assertIsNotNone(item) def test_find_item_missing(self): data = roadmap.load_roadmap(self.roadmap_file) self.assertIsNone(roadmap.find_item(data, "nonexistent")) def test_parse_date_iso(self): d = roadmap.parse_date("2026-02-08") self.assertEqual(d.year, 2026) self.assertEqual(d.month, 2) def test_parse_date_with_time(self): d = roadmap.parse_date("2026-02-08 09:00") self.assertEqual(d.hour, 9) def test_parse_date_none(self): self.assertIsNone(roadmap.parse_date(None)) self.assertIsNone(roadmap.parse_date("not-a-date")) def test_overdue_detection(self, ): """Item-001 has deadline in the past → overdue.""" data = roadmap.load_roadmap(self.roadmap_file) now = datetime.now() overdue = [i for i in data["items"] if i["status"] != "done" and roadmap.parse_date(i.get("deadline")) and roadmap.parse_date(i["deadline"]) < now] self.assertEqual(len(overdue), 1) self.assertEqual(overdue[0]["id"], "item-001") def test_upcoming_detection(self): data = roadmap.load_roadmap(self.roadmap_file) now = datetime.now() cutoff = now + timedelta(days=7) upcoming = [i for i in data["items"] if i["status"] != "done" and roadmap.parse_date(i.get("deadline")) and now <= roadmap.parse_date(i["deadline"]) <= cutoff] self.assertEqual(len(upcoming), 1) self.assertEqual(upcoming[0]["id"], "item-002") def test_dep_check_missing(self): data = roadmap.load_roadmap(self.roadmap_file) id_map = {i["id"]: i for i in data["items"]} # item-004 depends on item-999 which doesn't exist item = id_map["item-004"] for dep_id in item["depends_on"]: self.assertNotIn(dep_id, id_map) def test_dep_check_incomplete(self): data = roadmap.load_roadmap(self.roadmap_file) id_map = {i["id"]: i for i in data["items"]} # item-002 depends on item-001 which is not done item = id_map["item-002"] for dep_id in item["depends_on"]: dep = id_map.get(dep_id) self.assertIsNotNone(dep) self.assertNotEqual(dep["status"], "done") def test_stale_detection(self): data = roadmap.load_roadmap(self.roadmap_file) now = datetime.now() stale = [i for i in data["items"] if i["status"] == "in_progress" and roadmap.parse_date(i.get("updated")) and (now - roadmap.parse_date(i["updated"])).days > 7] self.assertTrue(len(stale) >= 1) stale_ids = [s["id"] for s in stale] self.assertIn("item-004", stale_ids) def test_valid_statuses(self): for s in ["open", "in_progress", "blocked", "done"]: self.assertIn(s, roadmap.VALID_STATUSES) def test_valid_priorities(self): for p in ["P0", "P1", "P2", "P3"]: self.assertIn(p, roadmap.VALID_PRIORITIES) def test_now_iso(self): ts = roadmap.now_iso() # Should be parseable dt = datetime.fromisoformat(ts) self.assertIsInstance(dt, datetime) class TestReportGeneration(unittest.TestCase): def test_report_contains_sections(self): """Capture report output and verify structure.""" import io from contextlib import redirect_stdout data = { "items": [ {"id": "r1", "title": "Overdue thing", "status": "open", "priority": "P0", "deadline": "2020-01-01", "depends_on": [], "tags": [], "created": "2020-01-01T00:00:00", "updated": "2020-01-01T00:00:00", "notes": ""}, {"id": "r2", "title": "Active thing", "status": "in_progress", "priority": "P1", "deadline": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"), "depends_on": [], "tags": [], "created": "2020-01-01T00:00:00", "updated": "2020-01-01T00:00:00", "notes": ""}, ] } f = io.StringIO() args = type('Args', (), {})() with redirect_stdout(f): roadmap.cmd_report(args, data) output = f.getvalue() self.assertIn("Roadmap Status Report", output) self.assertIn("Overdue", output) self.assertIn("Overdue thing", output) if __name__ == "__main__": unittest.main()