From 25cacd4894b8d6977cfc2be61b9a9ef24dd49a30 Mon Sep 17 00:00:00 2001 From: zorruno <1+zorruno@noreply.home.fox.co.nz> Date: Sat, 3 Oct 2026 15:22:31 +1300 Subject: [PATCH] Add Home Assistant switch, regression tests and ignore rules --- .gitignore | 28 +++++++ meshcore_phone_app_mode.yaml | 42 ++++++++++ test_controller.py | 157 +++++++++++++++++++++++++++++++++++ 3 files changed, 227 insertions(+) create mode 100644 .gitignore create mode 100644 meshcore_phone_app_mode.yaml create mode 100644 test_controller.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d3900a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,28 @@ +# Local credentials and configuration +.env +.env.* +!.env.example +meshcore-phone-mode.ini +*.local.ini +*.local.yaml +*.local.yml +secrets/ + +# Python runtime and editor files +venv/ +.venv/ +__pycache__/ +*.py[cod] +.pytest_cache/ +.coverage +.DS_Store +*.swp +*.swo + +# Runtime data and downloaded distributions +*.log +*.db +*.db-* +*.sqlite* +*.tar.gz +*.zip diff --git a/meshcore_phone_app_mode.yaml b/meshcore_phone_app_mode.yaml new file mode 100644 index 0000000..806a372 --- /dev/null +++ b/meshcore_phone_app_mode.yaml @@ -0,0 +1,42 @@ +############################################################################### +# MeshCore Phone App Mode — Zorruno v1.0 +# Date: 2026-10-03 +# Home Assistant package: /config/packages/meshcore_phone_app_mode.yaml +# +# Purpose: +# Hand over the Wi-Fi companion between meshcore-webui on beaver and the +# native MeshCore phone app. The independent MQTT controller on beaver manages +# ONLY the meshcore-webui container. No SSH or Docker API access from HA. +# +# ON = WebUI stopped; disconnect other direct clients, then connect phone app. +# OFF = WebUI running; disconnect phone app BEFORE turning this switch OFF. +# +# MQTT commands: plain ON/OFF, QoS 1, NOT retained. +# MQTT state: retained JSON, derived from Docker inspection every 15 seconds. +# Availability: retained online/offline with an MQTT Last Will. +# +# No automatic action on HA/controller/broker restart. A stopped WebUI stays +# stopped until an explicit OFF command. The switch does not guarantee that +# another client is not connected, or that the remote repeater is reachable. +# +# Prerequisite: HA's MQTT integration must connect to this same broker or an +# existing bridge must carry these command/state/availability topics both ways. +############################################################################### + +mqtt: + - switch: + name: "MeshCore Phone App Mode" + unique_id: meshcore_phone_app_mode_beaver + icon: mdi:cellphone-wireless + command_topic: "viewroad-commands/meshcore/phone_app_mode" + state_topic: "viewroad-status/meshcore/phone_app_mode" + value_template: "{{ value_json.phone_app_mode }}" + json_attributes_topic: "viewroad-status/meshcore/phone_app_mode" + availability_topic: "viewroad-status/meshcore/controller/availability" + payload_on: "ON" + payload_off: "OFF" + payload_available: "online" + payload_not_available: "offline" + optimistic: false + qos: 1 + retain: false diff --git a/test_controller.py b/test_controller.py new file mode 100644 index 0000000..e26a617 --- /dev/null +++ b/test_controller.py @@ -0,0 +1,157 @@ +"""Behaviour tests with mocked MQTT and Docker; no live services are changed.""" +import json +import queue +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from controller import ( + AVAILABILITY_TOPIC, COMMAND_TOPIC, STATE_TOPIC, CONTAINER, + Controller, DockerControl, phone_mode, +) + + +class ControllerTests(unittest.TestCase): + def setUp(self): + self.client = Mock() + self.client.subscribe.return_value = (0, 42) + publication = Mock(rc=0) + publication.is_published.return_value = True + self.client.publish.return_value = publication + self.docker = Mock() + self.docker.status.return_value = "running" + self.control = Controller(self.client, self.docker) + + def ready(self): + self.control.on_connect(self.client, None, None, 0, None) + self.control.on_subscribe(self.client, None, 42, [1], None) + + def test_connect_and_reconnect_never_change_container(self): + self.ready() + self.control.report() + self.control.on_disconnect(self.client, None, None, 1, None) + self.ready() + self.control.report() + self.docker.set_mode.assert_not_called() + + def test_subscription_denial_keeps_controller_unavailable(self): + self.control.on_connect(self.client, None, None, 0, None) + self.control.on_subscribe(self.client, None, 42, [128], None) + self.assertFalse(self.control.connected.is_set()) + self.assertNotIn("online", [call.args[1] for call in self.client.publish.call_args_list]) + + def test_initial_connection_does_not_assume_subscription_succeeded(self): + self.control.on_connect(self.client, None, None, 0, None) + self.assertFalse(self.control.connected.is_set()) + + def test_retained_and_invalid_commands_do_not_run_docker(self): + self.ready() + for topic, payload, retained in [ + (COMMAND_TOPIC, b"ON", True), + (COMMAND_TOPIC, b"ON; docker rm mqtt5", False), + (COMMAND_TOPIC, b"off", False), + (COMMAND_TOPIC, b"\xff", False), + ("another/topic", b"ON", False), + ]: + self.control.on_message(self.client, None, SimpleNamespace( + topic=topic, payload=payload, retain=retained, + )) + self.assertTrue(self.control.commands.empty()) + self.docker.set_mode.assert_not_called() + + def test_live_command_waits_for_confirmed_docker_state(self): + self.ready() + self.control.on_message(self.client, None, SimpleNamespace( + topic=COMMAND_TOPIC, payload=b"ON", retain=False, + )) + self.docker.set_mode.return_value = "exited" + self.control.process_command(self.control.commands.get_nowait()) + self.docker.set_mode.assert_called_once_with("ON") + states = [json.loads(c.args[1]) for c in self.client.publish.call_args_list if c.args[0] == STATE_TOPIC] + self.assertEqual(states[-1]["phone_app_mode"], "ON") + self.assertEqual(states[-1]["container_status"], "exited") + + def test_failed_stop_reports_actual_running_state(self): + self.ready() + self.docker.set_mode.side_effect = RuntimeError("stop failed") + self.control.process_command("ON") + states = [json.loads(c.args[1]) for c in self.client.publish.call_args_list if c.args[0] == STATE_TOPIC] + self.assertEqual(states[-1]["phone_app_mode"], "OFF") + + def test_missing_container_reports_offline(self): + self.ready() + self.docker.status.side_effect = RuntimeError("missing container") + self.assertFalse(self.control.report()) + self.client.publish.assert_called_with(AVAILABILITY_TOPIC, "offline", qos=1, retain=True) + + def test_unacknowledged_state_does_not_report_online(self): + self.ready() + self.client.publish.reset_mock() + self.client.publish.return_value.is_published.return_value = False + self.assertFalse(self.control.report()) + self.assertNotIn("online", [call.args[1] for call in self.client.publish.call_args_list]) + + def test_disconnect_drops_queued_commands(self): + self.ready() + self.control.commands.put_nowait("OFF") + self.control.on_disconnect(self.client, None, None, 1, None) + self.assertTrue(self.control.commands.empty()) + self.control.process_command("OFF") + self.docker.set_mode.assert_not_called() + + def test_manual_container_changes_are_reported(self): + self.ready() + self.control.report("running") + self.control.report("exited") + states = [json.loads(c.args[1])["phone_app_mode"] for c in self.client.publish.call_args_list if c.args[0] == STATE_TOPIC] + self.assertEqual(states, ["OFF", "ON"]) + + +class DockerTests(unittest.TestCase): + def control(self): + with patch("controller.shutil.which", return_value="/usr/bin/docker"): + return DockerControl() + + def test_stop_only_targets_webui_with_separate_arguments(self): + control = self.control() + control.status = Mock(side_effect=["running", "exited"]) + control.run = Mock() + self.assertEqual(control.set_mode("ON"), "exited") + control.run.assert_called_once_with("stop", "--time", "10", CONTAINER, timeout=35) + + def test_start_only_targets_webui(self): + control = self.control() + control.status = Mock(side_effect=["exited", "running"]) + control.run = Mock() + self.assertEqual(control.set_mode("OFF"), "running") + control.run.assert_called_once_with("start", CONTAINER, timeout=35) + + def test_duplicate_commands_are_idempotent(self): + control = self.control() + control.status = Mock(return_value="exited") + control.run = Mock() + self.assertEqual(control.set_mode("ON"), "exited") + control.run.assert_not_called() + + def test_invalid_request_is_rejected_before_docker(self): + control = self.control() + control.status = Mock() + with self.assertRaises(ValueError): + control.set_mode("ON; shutdown") + control.status.assert_not_called() + + def test_paused_or_unknown_is_not_claimed_as_free(self): + for status in ["paused", "dead", "removing", "unknown"]: + with self.assertRaises(RuntimeError): + phone_mode(status) + + def test_successful_command_exit_without_state_change_is_failure(self): + control = self.control() + control.status = Mock(return_value="running") + control.run = Mock() + with self.assertRaises(RuntimeError): + control.set_mode("ON") + + +if __name__ == "__main__": + unittest.main()