home_control/tests/test_service.py

import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch

from lib._zyxel import ZyxelRouter, normalize_mac
from lib._arrival_detection import (
    ArrivalDetector,
    ArrivalMonitor,
    _configured_devices,
    _record_arrival,
)
from lib._notify import accepts_event
from service import Controller


class ArrivalDetectorTests(unittest.TestCase):
    def test_starting_present_does_not_emit_arrival(self):
        detector = ArrivalDetector(absent_after=2, present_after=2)
        self.assertFalse(detector.sample(True))
        self.assertFalse(detector.sample(True))

    def test_emits_only_after_absence_and_debounced_presence(self):
        detector = ArrivalDetector(absent_after=2, present_after=2)
        for value in (False, False, True):
            self.assertFalse(detector.sample(value))
        self.assertTrue(detector.sample(True))
        self.assertFalse(detector.sample(True))

    def test_single_missed_poll_does_not_mark_absent(self):
        detector = ArrivalDetector(absent_after=2, present_after=1)
        detector.sample(True)
        self.assertFalse(detector.sample(False))
        self.assertFalse(detector.sample(True))

    def test_accepts_up_to_three_devices(self):
        devices = _configured_devices(
            {
                "devices": [
                    {"id": "one", "mac": "00:00:00:00:00:01"},
                    {"id": "two", "mac": "00:00:00:00:00:02"},
                    {"id": "three", "mac": "00:00:00:00:00:03"},
                ]
            }
        )
        self.assertEqual(len(devices), 3)

    def test_rejects_more_than_three_devices(self):
        config = {
            "devices": [
                {"id": str(index), "mac": f"00:00:00:00:00:0{index}"}
                for index in range(1, 5)
            ]
        }
        with self.assertRaisesRegex(ValueError, "at most 3"):
            _configured_devices(config)

    @patch("lib._arrival_detection._build_router")
    def test_monitor_emits_generic_arrived_event(self, _build_router):
        events = []
        monitor = ArrivalMonitor(
            {
                "devices": [
                    {"id": "ignace", "mac": "00:00:00:00:00:01"}
                ],
                "polling": {"absent_after": 1, "present_after": 1},
            },
            events.append,
        )
        mac = "00:00:00:00:00:01"
        monitor._process_states({mac: False})
        monitor._process_states({mac: True})
        self.assertEqual(
            events,
            [
                {
                    "sender": "_arrival_detection",
                    "event": "arrived",
                    "id": "ignace",
                    "text": "",
                },
                {
                    "sender": "_arrival_detection",
                    "event": "first_arrival",
                    "id": "ignace",
                    "text": "",
                },
            ],
        )

    @patch("lib._arrival_detection._build_router")
    def test_first_arrival_fires_only_when_empty_house_becomes_occupied(
        self, _build_router
    ):
        events = []
        monitor = ArrivalMonitor(
            {
                "devices": [
                    {"id": "ignace", "mac": "00:00:00:00:00:01"},
                    {"id": "janine", "mac": "00:00:00:00:00:02"},
                ],
                "polling": {"absent_after": 1, "present_after": 1},
            },
            events.append,
        )
        first = "00:00:00:00:00:01"
        second = "00:00:00:00:00:02"
        monitor._process_states({first: False, second: False})
        monitor._process_states({first: True, second: False})
        monitor._process_states({first: True, second: True})

        self.assertEqual(
            [(event["event"], event["id"]) for event in events],
            [
                ("arrived", "ignace"),
                ("first_arrival", "ignace"),
                ("arrived", "janine"),
            ],
        )

    @patch("lib._arrival_detection._build_router")
    def test_monitor_emits_departure_and_empty_after_last_device_leaves(self, _build_router):
        events = []
        monitor = ArrivalMonitor(
            {
                "devices": [
                    {"id": "ignace", "mac": "00:00:00:00:00:01"},
                    {"id": "janine", "mac": "00:00:00:00:00:02"},
                ],
                "polling": {"absent_after": 1, "present_after": 1},
            },
            events.append,
        )
        first = "00:00:00:00:00:01"
        second = "00:00:00:00:00:02"
        monitor._process_states({first: True, second: True})
        self.assertEqual(events, [])
        monitor._process_states({first: False, second: True})
        self.assertEqual([event["event"] for event in events], ["departed"])
        monitor._process_states({first: False, second: False})
        self.assertEqual(
            [(event["event"], event["id"]) for event in events],
            [("departed", "ignace"), ("departed", "janine"), ("empty", "house")],
        )

    @patch("lib._arrival_detection._build_router")
    def test_initially_empty_house_emits_no_events(self, _build_router):
        events = []
        monitor = ArrivalMonitor(
            {
                "devices": [{"id": "ignace", "mac": "00:00:00:00:00:01"}],
                "polling": {"absent_after": 1, "present_after": 1},
            },
            events.append,
        )
        monitor._process_states({"00:00:00:00:00:01": False})
        self.assertEqual(events, [])


class MacTests(unittest.TestCase):
    def test_normalizes_common_formats(self):
        self.assertEqual(normalize_mac("AA-BB-CC-DD-EE-FF"), "aa:bb:cc:dd:ee:ff")

    def test_rejects_invalid_mac(self):
        with self.assertRaises(ValueError):
            normalize_mac("not-a-mac")

    def test_parses_ex5601_nested_lanhosts_response(self):
        router = ZyxelRouter("192.0.2.1", "user", "password")
        router.dal_get = lambda _oid: {
            "result": "ZCFG_SUCCESS",
            "Object": [
                {
                    "lanhosts": [
                        {
                            "PhysAddress": "2C:DB:07:50:2B:5C",
                            "Active": True,
                            "HostName": "workstation",
                            "IPAddress": "192.168.178.10",
                        }
                    ]
                }
            ],
        }
        hosts = router.get_lan_hosts()
        self.assertEqual(len(hosts), 1)
        self.assertEqual(hosts[0].mac, "2c:db:07:50:2b:5c")
        self.assertTrue(router.is_connected("2c-db-07-50-2b-5c"))

    def test_logs_only_the_first_successful_zyxel_connection(self):
        router = ZyxelRouter("192.0.2.1", "user", "password")
        with self.assertLogs("home_control.zyxel", level="INFO") as logs:
            router._log_first_connection()
            router._log_first_connection()
        self.assertEqual(len(logs.output), 1)
        self.assertIn("Connected successfully", logs.output[0])

    def test_debug_logs_presence_result(self):
        router = ZyxelRouter("192.0.2.1", "user", "password", debug=True)
        router.get_lan_hosts = lambda: []
        with self.assertLogs("home_control.zyxel", level="INFO") as logs:
            self.assertFalse(router.is_connected("2c:db:07:50:2b:5c"))
        self.assertIn("2c:db:07:50:2b:5c -> absent", logs.output[0])


class NotificationTests(unittest.TestCase):
    @patch("lib._cloud_logger.send_event")
    @patch("lib._notify.send_notification")
    def test_controller_routes_event_to_multiple_actions(
        self, send_notification, send_event
    ):
        config = {
            "controller": {
                "plugins": {
                    "source": {
                        "module": "lib.example",
                        "config": "source",
                        "on_event": ["_notify", "_cloud_logger"],
                    }
                },
                "actions": {
                    "_notify": {
                        "module": "lib._notify",
                        "config": "notify",
                    },
                    "_cloud_logger": {
                        "module": "lib._cloud_logger",
                        "config": "cloud_logger",
                    },
                },
            },
            "source": {},
            "notify": {"topic_url": "https://ntfy.example/home"},
            "cloud_logger": {"event_url": "https://logger.example/event"},
        }
        message = {
            "sender": "source",
            "event": "arrived",
            "id": "ignace",
            "text": "hello",
        }

        Controller(config).handle_event(message)

        send_notification.assert_called_once()
        send_event.assert_called_once()
        payload = send_event.call_args.args[0]
        self.assertEqual(payload["sender"], "source")
        self.assertEqual(payload["event"], "arrived")
        self.assertEqual(payload["id"], "ignace")
        self.assertEqual(payload["text"], "hello")
        self.assertIn("datetime", payload)
        self.assertEqual(
            send_event.call_args.kwargs,
            {"event_url": "https://logger.example/event", "timeout": 10.0},
        )

    @patch("lib._notify.send_notification")
    def test_controller_routes_generic_event_to_notify(self, send_notification):
        config = {
            "controller": {
                "plugins": {
                    "_arrival_detection": {
                        "module": "lib._arrival_detection",
                        "config": "arrival_detection",
                        "on_event": "_notify",
                    }
                },
                "actions": {
                    "_notify": {
                        "module": "lib._notify",
                        "handler": "on_event",
                        "config": "notify",
                    }
                },
            },
            "arrival_detection": {},
            "notify": {
                "enabled": True,
                "message": "Welcome home, {id}",
                "topic_url": "https://ntfy.example/home",
                "timeout": 4,
            },
        }
        Controller(config).handle_event(
            {
                "sender": "_arrival_detection",
                "event": "arrived",
                "id": "ignace",
                "text": "",
            }
        )
        send_notification.assert_called_once_with(
            "Welcome home, ignace",
            topic_url="https://ntfy.example/home",
            timeout=4.0,
        )

    def test_controller_has_no_plugin_specific_setup(self):
        config = {
            "controller": {
                "plugins": {
                    "example": {
                        "module": "lib.example",
                        "factory": "build",
                        "config": "example_settings",
                        "on_event": "example_action",
                    }
                },
                "actions": {},
            },
            "example_settings": {},
        }
        controller = Controller(config)
        self.assertEqual(controller.plugin_specs["example"]["factory"], "build")

    def test_notify_accept_filter(self):
        config = {
            "filter": {
                "accept": {"events": ["arrived"], "ids": ["ignace"]},
                "ignore": {"events": [], "ids": []},
            }
        }
        self.assertTrue(
            accepts_event(config, {"event": "arrived", "id": "ignace"})
        )
        self.assertFalse(
            accepts_event(config, {"event": "departed", "id": "ignace"})
        )
        self.assertFalse(
            accepts_event(config, {"event": "arrived", "id": "janine"})
        )

    def test_notify_ignore_filter_takes_precedence(self):
        config = {
            "filter": {
                "accept": {"events": ["departed"], "ids": ["ignace"]},
                "ignore": {"events": ["departed"], "ids": []},
            }
        }
        self.assertFalse(
            accepts_event(config, {"event": "departed", "id": "ignace"})
        )

    def test_monitor_appends_arrival_to_event_log(self):
        with TemporaryDirectory() as directory:
            event_log = Path(directory) / "arrivals.txt"
            config = {
                "event_log": str(event_log),
            }
            _record_arrival(
                config, {"id": "ignace", "mac": "aa:bb:cc:dd:ee:ff"}
            )
            self.assertTrue(
                event_log.read_text().endswith(
                    " ARRIVED id=ignace mac=aa:bb:cc:dd:ee:ff\n"
                )
            )


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