76 lines
3.8 KiB
Python
76 lines
3.8 KiB
Python
import datetime as dt
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"monitor_status", Path(__file__).resolve().parents[1] / "deploy" / "monitor_status.py")
|
|
monitor = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(monitor)
|
|
|
|
|
|
class MonitorStatusTests(unittest.TestCase):
|
|
def test_compose_json_is_reduced_to_safe_state(self):
|
|
value = json.dumps([
|
|
{"Service": "api", "State": "running", "Health": "healthy", "Publishers": "secret"},
|
|
{"Service": "worker", "State": "running", "Health": ""},
|
|
{"Service": "mongo", "State": "running", "Health": "healthy"},
|
|
]).encode()
|
|
self.assertEqual(monitor.parse_compose(value)["api"], {"state": "running", "health": "healthy"})
|
|
self.assertNotIn("Publishers", monitor.parse_compose(value)["api"])
|
|
|
|
def test_age_ignores_symlinks_and_unexpected_files(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
directory = Path(folder)
|
|
backup = directory / "guestops-20261001T021700Z.tar.gpg"
|
|
backup.write_bytes(b"fixture")
|
|
moment = dt.datetime(2026, 10, 1, 3, 17, tzinfo=dt.timezone.utc)
|
|
os.utime(backup, (moment.timestamp() - 3600, moment.timestamp() - 3600))
|
|
(directory / "secret.txt").write_text("ignored", encoding="utf-8")
|
|
self.assertEqual(monitor.age_seconds(directory, monitor.BACKUP_NAME, moment), 3600)
|
|
|
|
def test_build_status_contains_only_aggregate_health(self):
|
|
now = dt.datetime(2026, 10, 1, 12, tzinfo=dt.timezone.utc)
|
|
compose = json.dumps([
|
|
{"Service": name, "State": "running", "Health": "healthy"}
|
|
for name in ("api", "worker", "mongo")
|
|
]).encode()
|
|
usage = type("Usage", (), {"total": 1000, "used": 500, "free": 500})()
|
|
with tempfile.TemporaryDirectory() as folder, \
|
|
patch.object(monitor, "https_status", return_value={"ready": True, "certificateDaysRemaining": 60}), \
|
|
patch.object(monitor, "run", return_value=compose), \
|
|
patch.object(monitor, "worker_heartbeat_age", return_value=30), \
|
|
patch.object(monitor.shutil, "disk_usage", return_value=usage):
|
|
status, errors = monitor.build_status(Path(folder), Path(folder), "https://example.invalid", now)
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(status["workerHeartbeatAgeSeconds"], 30)
|
|
self.assertEqual(status["diskFreePercent"], 50)
|
|
self.assertNotIn("credentials", json.dumps(status).lower())
|
|
|
|
def test_failed_probes_write_categories_not_exception_details(self):
|
|
now = dt.datetime(2026, 10, 1, 12, tzinfo=dt.timezone.utc)
|
|
usage = type("Usage", (), {"total": 1000, "used": 500, "free": 500})()
|
|
with tempfile.TemporaryDirectory() as folder, \
|
|
patch.object(monitor, "https_status", side_effect=RuntimeError("secret value")), \
|
|
patch.object(monitor, "run", side_effect=RuntimeError("secret value")), \
|
|
patch.object(monitor, "worker_heartbeat_age", side_effect=RuntimeError("secret value")), \
|
|
patch.object(monitor.shutil, "disk_usage", return_value=usage):
|
|
status, errors = monitor.build_status(Path(folder), Path(folder), "https://example.invalid", now)
|
|
self.assertGreaterEqual(len(errors), 3)
|
|
self.assertNotIn("secret value", json.dumps(status))
|
|
|
|
def test_status_output_is_atomic_json(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "status.json"
|
|
monitor.write_status(path, {"schemaVersion": 1, "errors": []})
|
|
self.assertEqual(json.loads(path.read_text(encoding="utf-8"))["schemaVersion"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|