131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Tests for health check module."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||
|
|
from tools.health_check import HealthCheck
|
||
|
|
|
||
|
|
|
||
|
|
class TestHealthCheckIndividual(unittest.TestCase):
|
||
|
|
"""Test individual health checks."""
|
||
|
|
|
||
|
|
def test_check_disk_space(self):
|
||
|
|
ok, detail = HealthCheck.check_disk_space()
|
||
|
|
self.assertIsInstance(ok, bool)
|
||
|
|
self.assertIn("GB", detail)
|
||
|
|
self.assertIn("free", detail)
|
||
|
|
|
||
|
|
def test_check_memory_file_exists(self):
|
||
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
|
||
|
|
f.write("# Memory\nTest content\n")
|
||
|
|
f.flush()
|
||
|
|
with patch.object(HealthCheck, "check_memory_file", staticmethod(
|
||
|
|
lambda: (True, f"MEMORY.md: 2 lines, {os.path.getsize(f.name)} bytes")
|
||
|
|
)):
|
||
|
|
ok, detail = HealthCheck.check_memory_file()
|
||
|
|
self.assertTrue(ok)
|
||
|
|
os.unlink(f.name)
|
||
|
|
|
||
|
|
def test_check_skills_count(self):
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
# Create a fake skill
|
||
|
|
skill_dir = Path(tmp) / "test-skill"
|
||
|
|
skill_dir.mkdir()
|
||
|
|
(skill_dir / "SKILL.md").write_text("---\nname: test\n---\n# Test")
|
||
|
|
|
||
|
|
with patch.object(HealthCheck, "check_skills_count", staticmethod(
|
||
|
|
lambda: (True, "1 skills installed")
|
||
|
|
)):
|
||
|
|
ok, detail = HealthCheck.check_skills_count()
|
||
|
|
self.assertTrue(ok)
|
||
|
|
self.assertIn("1", detail)
|
||
|
|
|
||
|
|
def test_check_cron_jobs_valid(self):
|
||
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||
|
|
json.dump([
|
||
|
|
{"id": "1", "status": "active"},
|
||
|
|
{"id": "2", "status": "paused"},
|
||
|
|
], f)
|
||
|
|
f.flush()
|
||
|
|
|
||
|
|
# Test the logic directly
|
||
|
|
jobs = json.loads(Path(f.name).read_text())
|
||
|
|
active = sum(1 for j in jobs if j.get("status") == "active")
|
||
|
|
self.assertEqual(active, 1)
|
||
|
|
os.unlink(f.name)
|
||
|
|
|
||
|
|
|
||
|
|
class TestHealthCheckRunner(unittest.TestCase):
|
||
|
|
"""Test the health check runner."""
|
||
|
|
|
||
|
|
def test_check_method(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
result = hc.check("test_pass", lambda: (True, "all good"))
|
||
|
|
self.assertEqual(result["status"], "PASS")
|
||
|
|
self.assertEqual(result["detail"], "all good")
|
||
|
|
|
||
|
|
def test_check_failure(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
result = hc.check("test_fail", lambda: (False, "broken"))
|
||
|
|
self.assertEqual(result["status"], "FAIL")
|
||
|
|
|
||
|
|
def test_check_exception(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
def boom():
|
||
|
|
raise RuntimeError("kaboom")
|
||
|
|
result = hc.check("test_error", boom)
|
||
|
|
self.assertEqual(result["status"], "ERROR")
|
||
|
|
self.assertIn("kaboom", result["detail"])
|
||
|
|
|
||
|
|
def test_check_critical_flag(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
result = hc.check("test_crit", lambda: (False, "bad"), critical=True)
|
||
|
|
self.assertTrue(result["critical"])
|
||
|
|
|
||
|
|
def test_run_all_returns_structure(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
result = hc.run_all()
|
||
|
|
self.assertIn("timestamp", result)
|
||
|
|
self.assertIn("total", result)
|
||
|
|
self.assertIn("passed", result)
|
||
|
|
self.assertIn("failed", result)
|
||
|
|
self.assertIn("healthy", result)
|
||
|
|
self.assertIn("checks", result)
|
||
|
|
self.assertIsInstance(result["checks"], list)
|
||
|
|
self.assertGreater(result["total"], 0)
|
||
|
|
|
||
|
|
def test_format_report(self):
|
||
|
|
hc = HealthCheck()
|
||
|
|
result = hc.run_all()
|
||
|
|
report = hc.format_report(result)
|
||
|
|
self.assertIn("Ezra Health Check", report)
|
||
|
|
self.assertIn("HEALTHY", report.upper())
|
||
|
|
self.assertIn("|", report) # Table format
|
||
|
|
|
||
|
|
|
||
|
|
class TestHealthCheckLive(unittest.TestCase):
|
||
|
|
"""Live checks against actual infrastructure (may fail in CI)."""
|
||
|
|
|
||
|
|
def test_disk_space_live(self):
|
||
|
|
ok, detail = HealthCheck.check_disk_space()
|
||
|
|
# Should always work on a real system
|
||
|
|
self.assertIsInstance(ok, bool)
|
||
|
|
self.assertRegex(detail, r'\d+\.\d+GB free')
|
||
|
|
|
||
|
|
def test_hermes_gateway_live(self):
|
||
|
|
ok, detail = HealthCheck.check_hermes_gateway()
|
||
|
|
# Just verify it runs without error
|
||
|
|
self.assertIsInstance(ok, bool)
|
||
|
|
self.assertIsInstance(detail, str)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|