158 lines
6.5 KiB
Python
158 lines
6.5 KiB
Python
"""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()
|