ServerMonitor/tests/test_checks.py

"""Unit tests for ServerMonitor checks and notification logic."""

import http.server
import json
import os
import socket
import tempfile
import threading
import unittest
from io import BytesIO
from unittest.mock import patch

from checks import http_check, netdata_alarms, netdata_metrics, tcp_check
from notify import AlertEngine, StateStore, escape_html


class TestHttpCheck(unittest.TestCase):
    """Tests for ``http_check``."""

    def _serve(self, handler_class, port=0):
        """Start a threaded HTTP server and return its URL."""
        server = http.server.HTTPServer(("127.0.0.1", port), handler_class)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()

        def cleanup():
            server.shutdown()
            server.server_close()

        self.addCleanup(cleanup)
        host, port = server.server_address
        return f"http://{host}:{port}/"

    def test_ok_on_200(self):
        """http_check returns ok when the status is in expected_codes."""
        url = self._serve(http.server.SimpleHTTPRequestHandler)
        ok, message = http_check("test", url, [200], 5)
        self.assertTrue(ok)
        self.assertIn("HTTP 200", message)

    def test_fail_on_wrong_status(self):
        """http_check fails when the server returns an unexpected status."""
        class NotFoundHandler(http.server.BaseHTTPRequestHandler):
            def do_GET(self):
                self.send_response(404)
                self.end_headers()

            def log_message(self, fmt, *args):
                pass

        url = self._serve(NotFoundHandler)
        ok, message = http_check("test", url, [200], 5)
        self.assertFalse(ok)
        self.assertIn("HTTP 404", message)
        self.assertIn("expected [200]", message)

    def test_fail_on_connection_refused(self):
        """http_check reports failure when the port is closed."""
        # Find a closed port by binding to 0 and then immediately closing it.
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 0))
        _, port = sock.getsockname()
        sock.close()
        url = f"http://127.0.0.1:{port}/"
        ok, message = http_check("test", url, [200], 2)
        self.assertFalse(ok)
        self.assertIn("connection error", message)


class TestTcpCheck(unittest.TestCase):
    """Tests for ``tcp_check``."""

    def test_ok_on_listening_socket(self):
        """tcp_check succeeds when something is listening."""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("127.0.0.1", 0))
        sock.listen(1)
        _, port = sock.getsockname()
        try:
            ok, message = tcp_check("test", "127.0.0.1", port, 2)
            self.assertTrue(ok)
            self.assertIn("TCP ok", message)
        finally:
            sock.close()

    def test_fail_on_closed_port(self):
        """tcp_check fails when the port is closed."""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 0))
        _, port = sock.getsockname()
        sock.close()

        ok, message = tcp_check("test", "127.0.0.1", port, 2)
        self.assertFalse(ok)
        self.assertIn("TCP error", message)


class FakeResponse:
    """Minimal urllib response stand-in for monkeypatching."""

    def __init__(self, payload_dict):
        self._body = json.dumps(payload_dict).encode("utf-8")

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return False


class TestNetdataAlarms(unittest.TestCase):
    """Tests for ``netdata_alarms``."""

    def test_active_alarms_present(self):
        """netdata_alarms is not ok and names the alarm when alarms exist."""
        payload = {
            "hostname": "unraid",
            "status": True,
            "alarms": {
                "CPU_USAGE": {
                    "status": "CRITICAL",
                    "info": "CPU is high",
                    "value": 95,
                }
            },
        }
        with patch("checks.urllib.request.urlopen", return_value=FakeResponse(payload)):
            ok, message = netdata_alarms("http://netdata.test", 5)
        self.assertFalse(ok)
        self.assertIn("CPU_USAGE", message)
        self.assertIn("CRITICAL", message)
        self.assertIn("95", message)

    def test_empty_alarms_ok(self):
        """netdata_alarms is ok when alarms are empty."""
        payload = {
            "hostname": "unraid",
            "status": True,
            "alarms": {},
        }
        with patch("checks.urllib.request.urlopen", return_value=FakeResponse(payload)):
            ok, message = netdata_alarms("http://netdata.test", 5)
        self.assertTrue(ok)
        self.assertIn("empty", message)

    def test_status_false_not_ok(self):
        """netdata_alarms is not ok when the top-level status is false."""
        payload = {
            "hostname": "unraid",
            "status": False,
            "alarms": {},
        }
        with patch("checks.urllib.request.urlopen", return_value=FakeResponse(payload)):
            ok, message = netdata_alarms("http://netdata.test", 5)
        self.assertFalse(ok)
        self.assertIn("status false", message)


class TestNetdataMetrics(unittest.TestCase):
    """Tests for ``netdata_metrics`` parsing (Netdata v2 shapes)."""

    def test_cpu_without_idle_uses_busy_sum(self):
        """Netdata v2 has no idle dim: utilization is the sum of busy dims."""
        cpu_response = FakeResponse(
            {
                "labels": ["time", "user", "system", "iowait"],
                "data": [[1785957961, 50.0, 30.0, 20.0]],
            }
        )
        with patch(
            "checks.urllib.request.urlopen", return_value=cpu_response
        ) as mocked:
            metrics = netdata_metrics("http://netdata.test", 5)
        self.assertEqual(metrics["cpu_percent"], 100.0)

    def test_cpu_with_idle(self):
        """Older Netdata with idle dim: utilization = 100 - idle."""
        cpu_response = FakeResponse(
            {
                "labels": ["time", "idle", "user"],
                "data": [[1785957961, 25.0, 75.0]],
            }
        )
        with patch(
            "checks.urllib.request.urlopen", return_value=cpu_response
        ) as mocked:
            metrics = netdata_metrics("http://netdata.test", 5)
        self.assertAlmostEqual(metrics["cpu_percent"], 75.0)

    def test_disk_discovery_and_max_used(self):
        """disk_space.* charts are discovered and max used-percent is taken."""
        responses = [
            FakeResponse({"labels": ["time", "user"], "data": [[1, 10.0]]}),  # cpu
            FakeResponse(
                {
                    "labels": ["time", "used", "free"],
                    "data": [[1, 2000.0, 1000.0]],
                }
            ),  # ram
            FakeResponse(
                {"charts": {"disk_space./": {}, "disk_space./var": {}}}
            ),  # charts
            FakeResponse(
                {"labels": ["time", "avail", "used"], "data": [[1, 50.0, 45.3]]}
            ),  # disk / -> 45.3/(50+45.3) = 47.5%
            FakeResponse(
                {"labels": ["time", "avail", "used"], "data": [[1, 10.0, 88.0]]}
            ),  # disk /var -> 88/(10+88) = 89.8%
        ]
        with patch(
            "checks.urllib.request.urlopen",
            side_effect=responses,
        ):
            metrics = netdata_metrics("http://netdata.test", 5)
        self.assertAlmostEqual(metrics["disk_used_percent"], 89.8, delta=0.01)

    def test_ram_totals(self):
        """ram_used_mb and ram_total_mb are parsed from the ram chart."""
        responses = [
            FakeResponse({"labels": ["time", "user"], "data": [[1, 10.0]]}),  # cpu
            FakeResponse(
                {
                    "labels": ["time", "free", "used", "cached", "buffers"],
                    "data": [[1, 1000.0, 2000.0, 3000.0, 100.0]],
                }
            ),  # ram
            FakeResponse({"charts": {}}),  # no disk charts
        ]
        with patch(
            "checks.urllib.request.urlopen",
            side_effect=responses,
        ):
            metrics = netdata_metrics("http://netdata.test", 5)
        self.assertEqual(metrics["ram_used_mb"], 2000.0)
        self.assertEqual(metrics["ram_total_mb"], 6100.0)


class TestAlertEngine(unittest.TestCase):
    """Tests for ``AlertEngine`` deduplication logic."""

    def test_first_crit_alerts(self):
        """The first CRIT transition triggers an alert."""
        store = StateStore()
        store.load()
        engine = AlertEngine()
        engine.set_state("svc", "CRIT")
        should, recovery = engine.should_alert("svc", 1000, store, 600)
        self.assertTrue(should)
        self.assertFalse(recovery)

    def test_second_crit_within_window_does_not_alert(self):
        """A repeated CRIT inside the reminder window is suppressed."""
        store = StateStore()
        store.load()
        engine = AlertEngine()
        engine.set_state("svc", "CRIT")
        engine.should_alert("svc", 1000, store, 600)
        engine.set_state("svc", "CRIT")
        should, recovery = engine.should_alert("svc", 1200, store, 600)
        self.assertFalse(should)
        self.assertFalse(recovery)

    def test_crit_after_window_alerts_again(self):
        """A CRIT after the reminder window triggers a reminder."""
        store = StateStore()
        store.load()
        engine = AlertEngine()
        engine.set_state("svc", "CRIT")
        engine.should_alert("svc", 1000, store, 600)
        engine.set_state("svc", "CRIT")
        should, recovery = engine.should_alert("svc", 1700, store, 600)
        self.assertTrue(should)
        self.assertFalse(recovery)

    def test_recovery_sent_once(self):
        """A CRIT → OK transition triggers exactly one recovery message."""
        store = StateStore()
        store.load()
        engine = AlertEngine()
        engine.set_state("svc", "CRIT")
        engine.should_alert("svc", 1000, store, 600)
        engine.set_state("svc", "OK")
        should, recovery = engine.should_alert("svc", 1100, store, 600)
        self.assertTrue(should)
        self.assertTrue(recovery)
        # Repeated OK must not alert again.
        engine.set_state("svc", "OK")
        should, recovery = engine.should_alert("svc", 1200, store, 600)
        self.assertFalse(should)
        self.assertFalse(recovery)


    def test_no_recovery_without_prior_alert(self):
        """A CRIT observed at baseline (never alerted) must not send recovery."""
        store = StateStore()
        store.load()
        # Simulate baseline: CRIT observed but alert_count stays 0.
        store.set("svc", "CRIT", 0.0, 0)
        engine = AlertEngine()
        engine.set_state("svc", "OK")
        should, recovery = engine.should_alert("svc", 1100, store, 600)
        self.assertFalse(should)
        self.assertFalse(recovery)
        # State must still be updated to OK.
        self.assertEqual(store.get("svc")["state"], "OK")


class TestStateStore(unittest.TestCase):
    """Tests for ``StateStore`` persistence."""

    def test_round_trip(self):
        """State written to disk can be loaded back."""
        with tempfile.NamedTemporaryFile(delete=False) as tmp:
            path = tmp.name
        try:
            store = StateStore(path)
            store.load()
            store.set("svc", "CRIT", 123.4, 7)
            store.save()

            store2 = StateStore(path)
            store2.load()
            self.assertEqual(store2.get("svc"), {"state": "CRIT", "last_alert_ts": 123.4, "alert_count": 7})
        finally:
            os.unlink(path)

    def test_corrupt_file_starts_fresh(self):
        """A corrupt state file is handled gracefully and yields an empty state."""
        with tempfile.NamedTemporaryFile(delete=False, mode="w", encoding="utf-8") as tmp:
            tmp.write("not valid json {{}}")
            path = tmp.name
        try:
            store = StateStore(path)
            store.load()
            self.assertEqual(store.get("svc"), {"state": "OK", "last_alert_ts": 0.0, "alert_count": 0})
        finally:
            os.unlink(path)


class TestEscapeHtml(unittest.TestCase):
    """Tests for ``escape_html``."""

    def test_escapes_special_chars(self):
        """escape_html escapes &, < and >."""
        self.assertEqual(escape_html("a & b < c > d"), "a &amp; b &lt; c &gt; d")


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