zorruno
593f86e6dd
Use configured topics for commands, state, availability and Last Will. Replace shared installation prefixes with generic examples. Validate configuration before stopping an existing service; add tests for custom and invalid topics.
238 lines
10 KiB
Python
238 lines
10 KiB
Python
"""Behaviour tests with mocked MQTT and Docker; no live services are changed."""
|
|
import json
|
|
import configparser
|
|
import tempfile
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace, ModuleType
|
|
from unittest.mock import Mock, patch
|
|
|
|
from controller import (
|
|
CONTAINER, Topics, Controller, DockerControl, phone_mode, load_config, main,
|
|
)
|
|
|
|
|
|
TEST_TOPICS = Topics("custom/command", "custom/state", "custom/availability")
|
|
|
|
|
|
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, TEST_TOPICS)
|
|
|
|
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()
|
|
self.client.subscribe.assert_called_with(TEST_TOPICS.command, qos=1)
|
|
|
|
def test_shutdown_publishes_to_configured_availability_topic(self):
|
|
self.ready()
|
|
self.control.stopping.set()
|
|
self.client.publish.reset_mock()
|
|
self.control.run()
|
|
self.client.publish.assert_called_once_with(
|
|
TEST_TOPICS.availability, "offline", qos=1, retain=True,
|
|
)
|
|
|
|
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 [
|
|
(TEST_TOPICS.command, b"ON", True),
|
|
(TEST_TOPICS.command, b"ON; docker rm mqtt5", False),
|
|
(TEST_TOPICS.command, b"off", False),
|
|
(TEST_TOPICS.command, 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=TEST_TOPICS.command, 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] == TEST_TOPICS.state]
|
|
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] == TEST_TOPICS.state]
|
|
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(TEST_TOPICS.availability, "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] == TEST_TOPICS.state]
|
|
self.assertEqual(states, ["OFF", "ON"])
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def config(self, values):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "controller.ini"
|
|
path.write_text("[mqtt]\n" + values)
|
|
return load_config(path)
|
|
|
|
def test_full_custom_topics_are_loaded_without_prefix_or_interpolation(self):
|
|
_, _, _, _, topics = self.config(
|
|
"command_topic = house%20/radio/set\n"
|
|
"state_topic = another/tree/state\n"
|
|
"availability_topic = separate/health\n"
|
|
)
|
|
self.assertEqual(topics, Topics("house%20/radio/set", "another/tree/state", "separate/health"))
|
|
|
|
def test_missing_topics_do_not_fall_back_to_another_installation(self):
|
|
with self.assertRaisesRegex(ValueError, "Add full topics"):
|
|
self.config("host = 127.0.0.1\n")
|
|
|
|
def test_invalid_topics_are_rejected_for_each_setting(self):
|
|
for index in range(3):
|
|
for invalid in ("", "radio/#", "radio/+/state", "radio\0state", "x" * 65536, "é" * 32768):
|
|
with self.subTest(index=index, invalid=invalid[:20]):
|
|
values = ["radio/set", "radio/state", "radio/availability"]
|
|
values[index] = invalid
|
|
with self.assertRaises(ValueError):
|
|
Topics(*values)
|
|
|
|
def test_duplicate_topics_are_rejected(self):
|
|
for values in (("same", "same", "other"), ("same", "other", "same"), ("other", "same", "same")):
|
|
with self.assertRaisesRegex(ValueError, "must be different"):
|
|
Topics(*values)
|
|
|
|
def test_blank_topic_in_ini_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
self.config("command_topic = \nstate_topic = radio/state\navailability_topic = radio/health\n")
|
|
|
|
def test_shared_examples_use_matching_generic_topics(self):
|
|
root = Path(__file__).parent
|
|
_, _, _, _, topics = load_config(root / "meshcore-phone-mode.ini.example")
|
|
yaml = (root / "meshcore_phone_app_mode.yaml").read_text()
|
|
for key, topic in (("command_topic", topics.command), ("state_topic", topics.state),
|
|
("json_attributes_topic", topics.state), ("availability_topic", topics.availability)):
|
|
self.assertIn(f'{key}: "{topic}"', yaml)
|
|
|
|
def test_main_sets_last_will_and_controller_from_same_private_config(self):
|
|
mqtt = Mock()
|
|
paho_package = ModuleType("paho")
|
|
mqtt_package = ModuleType("paho.mqtt")
|
|
paho_package.mqtt = mqtt_package
|
|
mqtt_package.client = mqtt
|
|
modules = {"paho": paho_package, "paho.mqtt": mqtt_package, "paho.mqtt.client": mqtt}
|
|
with patch.dict(sys.modules, modules), patch("controller.load_config", return_value=(
|
|
"127.0.0.1", 1883, 15, configparser.ConfigParser(), TEST_TOPICS,
|
|
)), patch("controller.DockerControl") as docker, patch("controller.Controller") as control, \
|
|
patch("controller.signal.signal"), patch("controller.logging.basicConfig"), \
|
|
patch.object(sys, "argv", ["controller.py"]):
|
|
main()
|
|
mqtt.Client.return_value.will_set.assert_called_once_with(
|
|
TEST_TOPICS.availability, "offline", qos=1, retain=True,
|
|
)
|
|
control.assert_called_once_with(mqtt.Client.return_value, docker.return_value, TEST_TOPICS, 15)
|
|
control.return_value.run.assert_called_once_with()
|
|
|
|
|
|
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()
|