homelab/kubernetes/apps/hermes/skills/tests/test_alerts.py

"""Offline tests for the Hermes skill scripts (stdlib unittest, no network).

Run: python3 -m unittest discover -s kubernetes/apps/hermes/skills/tests
"""
import datetime
import hashlib
import importlib.util
import os
import tempfile
import unittest
from unittest import mock

HERE = os.path.dirname(os.path.abspath(__file__))


def load(name, rel):
    spec = importlib.util.spec_from_file_location(name, os.path.join(HERE, "..", rel))
    assert spec and spec.loader
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


alerts = load("alerts", "alert-monitor/scripts/alerts.py")


class TitleContract(unittest.TestCase):
    """The dedup gate and the [alert] card instructions depend on this exact title shape."""

    def test_namespace_first(self):
        self.assertEqual(alerts.title_of({"alertname": "A", "namespace": "ns", "job": "j"}), "[alert] A (ns)")

    def test_job_fallback(self):
        self.assertEqual(alerts.title_of({"alertname": "TargetDown", "job": "pve"}), "[alert] TargetDown (pve)")

    def test_cluster_fallback(self):
        self.assertEqual(alerts.title_of({"alertname": "A"}), "[alert] A (cluster)")


class NoiseLog(unittest.TestCase):
    def issues(self, *lines):
        return [{"title": alerts.NOISE_TITLE, "body": "header\n\n" + "\n".join(lines)}]

    def test_recent_entry_suppresses(self):
        today = datetime.date.today().isoformat()
        fps = alerts.noise_fingerprints(self.issues(f"- 0123456789abcdef KubeJobFailed — {today}: benign"))
        self.assertEqual(fps, {"0123456789abcdef"})

    def test_expired_entry_does_not_suppress(self):
        old = (datetime.date.today() - datetime.timedelta(days=alerts.NOISE_TTL_DAYS + 1)).isoformat()
        self.assertEqual(alerts.noise_fingerprints(self.issues(f"- 0123456789abcdef A — {old}: x")), set())

    def test_invalid_date_is_dropped_not_fatal(self):
        today = datetime.date.today().isoformat()
        fps = alerts.noise_fingerprints(self.issues(
            "- aaaaaaaaaaaaaaaa A — 2026-02-30: impossible date",
            f"- bbbbbbbbbbbbbbbb B — {today}: fine",
        ))
        self.assertEqual(fps, {"bbbbbbbbbbbbbbbb"})

    def test_fingerprint_in_prose_does_not_suppress(self):
        self.assertEqual(alerts.noise_fingerprints(self.issues("see cccccccccccccccc for details")), set())

    def test_no_noise_issue(self):
        self.assertEqual(alerts.noise_fingerprints([{"title": "other", "body": "x"}]), set())


class BlipFingerprint(unittest.TestCase):
    def test_matches_old_pipeline(self):
        # printf '%s|%s' "$NAME" "$SCOPE" | sha256sum | cut -c1-16
        want = hashlib.sha256(b"TargetDown|pve").hexdigest()[:16]
        series = [{"metric": {"alertname": "TargetDown", "job": "pve"}}]
        with mock.patch.object(alerts, "http", side_effect=[[], {"data": {"result": series}}]), \
                mock.patch.object(alerts, "open_issues", return_value=[]), \
                mock.patch("builtins.print") as out:
            alerts.cmd_gate(None)
        printed = "\n".join(str(c.args[0]) for c in out.call_args_list)
        self.assertIn(want, printed)
        self.assertIn('"[alert] TargetDown (pve)"', printed)


class Gate(unittest.TestCase):
    def run_gate(self, am, prom, issues):
        with mock.patch.object(alerts, "http", side_effect=[am, prom]), \
                mock.patch.object(alerts, "open_issues", return_value=issues), \
                mock.patch("builtins.print") as out:
            alerts.cmd_gate(None)
        return [str(c.args[0]) for c in out.call_args_list]

    def alert(self, name, ns, fp, receivers=("email",)):
        return {"labels": {"alertname": name, "namespace": ns}, "fingerprint": fp,
                "annotations": {}, "receivers": [{"name": r} for r in receivers]}

    def test_nothing_new_sleeps(self):
        am = [self.alert("Watchdog", "monitoring", "1" * 16),
              self.alert("Muted", "x", "2" * 16, receivers=("null",)),
              self.alert("Known", "ns", "3" * 16)]
        lines = self.run_gate(am, {"data": {"result": []}}, [{"title": "[alert] Known (ns)", "body": ""}])
        self.assertEqual(lines[-1], '{"wakeAgent": false}')

    def test_new_alert_wakes(self):
        lines = self.run_gate([self.alert("New", "ns", "4" * 16)], {"data": {"result": []}}, [])
        self.assertNotIn('"wakeAgent": false', lines[-1])
        self.assertIn('"[alert] New (ns)"', lines[-1])

    def test_prometheus_error_keeps_live_alerts(self):
        with mock.patch.object(alerts, "http",
                               side_effect=[[self.alert("New", "ns", "5" * 16)], alerts.ApiError("HTTP 503")]), \
                mock.patch.object(alerts, "open_issues", return_value=[]), \
                mock.patch("builtins.print") as out:
            alerts.cmd_gate(None)
        lines = [str(c.args[0]) for c in out.call_args_list]
        self.assertTrue(any(line.startswith("WARNING: blip lookback unavailable") for line in lines))
        self.assertIn('"[alert] New (ns)"', lines[-1])


class SecretGuard(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.TemporaryDirectory()
        os.makedirs(os.path.join(self.dir.name, "gitea"))
        with open(os.path.join(self.dir.name, "gitea", "api-token"), "w") as f:
            f.write("0123456789abcdef0123456789abcdef01234567\n")
        os.makedirs(os.path.join(self.dir.name, "talos.dev"))
        with open(os.path.join(self.dir.name, "talos.dev", "config"), "w") as f:
            f.write("context: x\nendpoints:\n  - talos.default\nkey: LS0tLS1CRUdJTiBFRDI1NTE5IFBSSVZBVEUgS0VZLS0tLS0K\n")
        self.patch = mock.patch.object(alerts, "SECRETS_DIR", self.dir.name)
        self.patch.start()

    def tearDown(self):
        self.patch.stop()
        self.dir.cleanup()

    def test_whole_token_refused(self):
        self.assertTrue(alerts.leaks_secret("token is 0123456789abcdef0123456789abcdef01234567 here"))

    def test_piece_of_key_refused(self):
        self.assertTrue(alerts.leaks_secret("leak: RUdJTiBFRDI1NTE5IFBSSVZB"))

    def test_env_secret_refused(self):
        with mock.patch.dict(os.environ, {"API_SERVER_KEY": "abcDEF123456ghiJKL789012mno"}):
            self.assertTrue(alerts.leaks_secret("x abcDEF123456ghiJKL789012mno y"))

    def test_unaligned_piece_refused(self):
        # 24 characters starting at offset 3 of the 40-char token
        self.assertTrue(alerts.leaks_secret("x 3456789abcdef0123456789a y"))

    def test_certificates_are_not_secrets(self):
        import base64
        pem = "-----BEGIN CERTIFICATE-----\nMIIBszCCAVmgAwIBAgIRAKs1v0abcdefghijklmnop123456\n-----END CERTIFICATE-----\n"
        with open(os.path.join(self.dir.name, "ca.crt"), "w") as f:
            f.write(pem)
        b64 = base64.b64encode(pem.encode()).decode()
        with open(os.path.join(self.dir.name, "talos.dev", "config"), "a") as f:
            f.write(f"ca: {b64}\n")
        self.assertFalse(alerts.leaks_secret("caBundle: " + b64))
        self.assertFalse(alerts.leaks_secret("MIIBszCCAVmgAwIBAgIRAKs1v0abcdefghijklmnop123456"))
        # the private key in the same talosconfig is still guarded
        self.assertTrue(alerts.leaks_secret("LS0tLS1CRUdJTiBFRDI1NTE5IFBSSVZBVEUgS0VZLS0tLS0K"))

    def test_normal_body_passes(self):
        body = ("**What is firing**: KubeJobFailed in renovate since 2026-10-20T10:00:00Z, "
                "talos.default endpoint, kube-prometheus-stack-alertmanager.")
        self.assertFalse(alerts.leaks_secret(body))


class FileWithLabel(unittest.TestCase):
    BODY = "**What is firing**: KubeJobFailed in renovate since today, evidence below."

    def run_file(self, labels_by_url, label="daedalus"):
        calls = []

        def fake(method, url, body=None, auth=False, timeout=30):
            calls.append((method, url, body))
            if method == "GET" and "/labels" in url:
                return labels_by_url.get(url.split("?")[0], [])
            if method == "GET":
                return []  # open issues
            return {"number": 9, "html_url": "https://gitea.renner.uno/konairius/homelab/issues/9"}

        args = mock.Mock(title="[alert] KubeJobFailed (renovate)", fingerprint="0" * 16, label=label)
        with mock.patch.object(alerts, "http", side_effect=fake), \
                mock.patch.object(alerts, "leaks_secret", return_value=False), \
                mock.patch("sys.stdin", mock.Mock(read=lambda: self.BODY)):
            alerts.cmd_file(args)
        return [c for c in calls if c[0] == "POST"][0][2]

    def test_org_label_id_set_at_creation(self):
        org = "https://gitea.renner.uno/api/v1/orgs/konairius/labels"
        posted = self.run_file({org: [{"id": 41, "name": "atlas"}, {"id": 42, "name": "daedalus"}]})
        self.assertEqual(posted["labels"], [42])
        self.assertIn("`daedalus` expert", posted["body"])
        self.assertNotIn("issue-fix", posted["body"])

    def test_repo_label_wins(self):
        repo = alerts.GITEA + "/labels"
        posted = self.run_file({repo: [{"id": 7, "name": "Daedalus"}]})
        self.assertEqual(posted["labels"], [7])

    def test_missing_label_files_without_it(self):
        posted = self.run_file({})
        self.assertNotIn("labels", posted)

    def test_label_is_required_and_validated(self):
        with mock.patch("sys.argv", ["alerts.py", "file", "--title", "[alert] X (y)", "--fingerprint", "0" * 16,
                                     "--label", "bogus"]), self.assertRaises(SystemExit):
            alerts.main()


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