included cloud logging with filters
This commit is contained in:
+248
-17
@@ -4,7 +4,14 @@ from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
|
||||
from lib._zyxel import ZyxelRouter, normalize_mac
|
||||
from lib._arrival_detection import ArrivalDetector, _notify
|
||||
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):
|
||||
@@ -26,6 +33,93 @@ class ArrivalDetectorTests(unittest.TestCase):
|
||||
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": "",
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
@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):
|
||||
@@ -57,38 +151,175 @@ class MacTests(unittest.TestCase):
|
||||
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._arrival_detection.send_notification")
|
||||
def test_uses_notify_module(self, send_notification):
|
||||
@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 = {
|
||||
"notification": {
|
||||
"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, {name} ({mac})",
|
||||
"message": "Welcome home, {id}",
|
||||
"topic_url": "https://ntfy.example/home",
|
||||
"timeout": 4,
|
||||
}
|
||||
},
|
||||
}
|
||||
_notify(config, {"name": "Ignace", "mac": "aa:bb:cc:dd:ee:ff"})
|
||||
Controller(config).handle_event(
|
||||
{
|
||||
"sender": "_arrival_detection",
|
||||
"event": "arrived",
|
||||
"id": "ignace",
|
||||
"text": "",
|
||||
}
|
||||
)
|
||||
send_notification.assert_called_once_with(
|
||||
"Welcome home, Ignace (aa:bb:cc:dd:ee:ff)",
|
||||
"Welcome home, ignace",
|
||||
topic_url="https://ntfy.example/home",
|
||||
timeout=4.0,
|
||||
)
|
||||
|
||||
@patch("lib._arrival_detection.send_notification")
|
||||
def test_appends_notification_to_event_log(self, send_notification):
|
||||
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 = {
|
||||
"notification": {
|
||||
"enabled": True,
|
||||
"message": "{name} arrived home",
|
||||
"event_log": str(event_log),
|
||||
}
|
||||
"event_log": str(event_log),
|
||||
}
|
||||
_notify(config, {"name": "Ignace", "mac": "aa:bb:cc:dd:ee:ff"})
|
||||
self.assertTrue(event_log.read_text().endswith(" Ignace arrived home\n"))
|
||||
_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__":
|
||||
|
||||
Reference in New Issue
Block a user