enodia-sentinal/tests/test_gui_model.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""Tests for the GUI presentation model (GUI-free)."""
import unittest
from unittest.mock import patch

from enodia_sentinel.config import Config
from enodia_sentinel.gui.model import (
    alert_rows,
    collect,
    incident_rows,
    integrity_lines,
    plan_rows,
    plan_text,
    posture_rows,
    status_lines,
)
from enodia_sentinel.tray.state import parse_status


class TestStatusLines(unittest.TestCase):
    def test_empty_status_shows_help(self):
        lines = status_lines({})
        self.assertTrue(any("No status available" in line for line in lines))
        self.assertTrue(any("Start the daemon" in line for line in lines))

    def test_running_status_fields(self):
        lines = status_lines({
            "running": True,
            "host": "testhost",
            "version": "0.7.0",
            "total_alerts": 3,
            "counts": {"HIGH": 1, "MEDIUM": 2},
            "last_alert": "2026-07-10T12:00:00",
            "heartbeat_age": 12,
            "heartbeat_stale": False,
            "ebpf": "attached",
            "ebpf_exec": "on",
            "ebpf_syscall": "off",
        })
        text = "\n".join(lines)
        self.assertIn("testhost", text)
        self.assertIn("running", text)
        self.assertIn("3 total", text)
        self.assertIn("HIGH:1", text)
        self.assertIn("12s ago", text)

    def test_stale_heartbeat_flag(self):
        lines = status_lines({"heartbeat_age": 90, "heartbeat_stale": True})
        self.assertTrue(any("STALE" in line for line in lines))


class TestAlertRows(unittest.TestCase):
    def test_flattens_snapshots(self):
        rows = alert_rows([
            {"time": "2026-07-10T12:00:00", "snapshot": "alert-1.json",
             "alerts": [
                 {"severity": "HIGH", "sid": 100001,
                  "signature": "reverse_shell", "detail": "bash"},
             ]},
            {"time": "2026-07-10T12:01:00", "snapshot": "alert-2.json",
             "alerts": [
                 {"severity": "MEDIUM", "sid": 100002,
                  "signature": "ld_preload", "detail": "libevil"},
             ]},
        ])
        self.assertEqual(len(rows), 2)
        self.assertEqual(rows[0][1], "HIGH")
        self.assertEqual(rows[1][3], "ld_preload")

    def test_empty(self):
        self.assertEqual(alert_rows([]), ())
        self.assertEqual(alert_rows(None), ())


class TestIncidentRows(unittest.TestCase):
    def test_maps_fields(self):
        rows = incident_rows([
            {"id": "INC-1", "severity": "CRITICAL",
             "first_seen": "2026-07-10T12:00:00",
             "last_seen": "2026-07-10T12:05:00",
             "alert_count": 5,
             "signatures": ["reverse_shell", "egress"]},
        ])
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0][0], "INC-1")
        self.assertEqual(rows[0][1], "CRITICAL")
        self.assertEqual(rows[0][4], 5)
        self.assertIn("reverse_shell", rows[0][5])


class TestPostureRows(unittest.TestCase):
    def test_maps_findings(self):
        rows = posture_rows({
            "findings": [
                {"severity": "HIGH", "signature": "ssh_root_login",
                 "detail": "PermitRootLogin yes"},
            ],
        })
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0], ("HIGH", "ssh_root_login",
                                   "PermitRootLogin yes"))


class TestIntegrityLines(unittest.TestCase):
    def test_empty(self):
        self.assertEqual(integrity_lines({}), ("No integrity state available.",))

    def test_flattens_nested(self):
        lines = integrity_lines({
            "watchdog": {"ok": True, "last_check": "now"},
            "fim": ["/etc/passwd", "/etc/shadow"],
        })
        self.assertTrue(any("watchdog.ok: True" in line for line in lines))
        self.assertTrue(any("fim: 2 item(s)" in line for line in lines))


class TestPlanRows(unittest.TestCase):
    def test_normal_plan(self):
        plans = [{
            "incident_id": "INC-9",
            "summary": "freeze and contain",
            "actions": [
                {"title": "Freeze evidence", "reason": "preserve",
                 "commands": ["tar", "-czf", "evidence.tgz"]},
                {"title": "Stop process", "commands": ["kill", "-STOP", "1234"]},
            ],
        }]
        rows, details = plan_rows(plans)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0], ("INC-9", 2, "freeze and contain"))
        self.assertIn("Freeze evidence", details["INC-9"])
        self.assertIn("$ kill", details["INC-9"])

    def test_error_plan(self):
        plans = [{"incident_id": "INC-X", "error": "bundle missing"}]
        rows, details = plan_rows(plans)
        self.assertEqual(rows[0][2], "bundle missing")
        self.assertEqual(details["INC-X"], "bundle missing")


class TestPlanText(unittest.TestCase):
    def test_empty_actions(self):
        text = plan_text({"incident_id": "INC-1", "actions": []})
        self.assertIn("(no actions suggested)", text)


class TestCollect(unittest.TestCase):
    def test_collect_composes_model(self):
        cfg = Config.load(None)
        fake = {
            "status": {
                "running": True, "total_alerts": 1,
                "counts": {"HIGH": 1}, "host": "h", "version": "v",
            },
            "alerts": [{"time": "t", "alerts": [
                {"severity": "HIGH", "sid": 1, "signature": "s", "detail": "d"},
            ]}],
            "incidents": [{"id": "INC-1", "severity": "HIGH",
                           "first_seen": "t1", "last_seen": "t2",
                           "alert_count": 1, "signatures": ["s"]}],
            "posture": {"findings": []},
            "integrity": {"ok": True},
            "events": ["line1"],
            "response": [{"incident_id": "INC-1", "actions": []}],
        }
        with patch("enodia_sentinel.gui.model.collect_model", return_value=fake):
            model = collect(cfg)
        self.assertTrue(model.header.running)
        self.assertEqual(len(model.alerts), 1)
        self.assertEqual(len(model.incidents), 1)
        self.assertIn("ok: True", model.integrity_lines)
        self.assertEqual(model.events, ("line1",))
        self.assertEqual(parse_status(fake["status"]).summary, model.header.summary)