Media
c2544c44c7
publish_router_mode registered its self-echo correlation marker before publishing but removed it in a finally block the instant the QoS 1 publish() call observed its PUBACK. A broker may legally deliver a Controller's own subscription copy of that same publish after the PUBACK, since MQTT does not order the two; when that happened, the Controller mistook its own late echo for live operator intent and relinquished automatic restoration ownership. Keep the marker claimable for a short, bounded mqtt.router_echo_grace_seconds window (default 2s) after the PUBACK instead of clearing it immediately. Genuine external/operator commands are unaffected: once the grace window elapses, a matching payload is still correctly treated as operator intent and ownership is relinquished as before. Verified with deterministic fake-clock unit tests plus a real Mosquitto broker integration test (tests/test_router_echo_mosquitto_integration.py, skipped automatically when Docker is unavailable).
505 lines
18 KiB
Python
505 lines
18 KiB
Python
import json
|
|
import queue
|
|
import threading
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
from controller.config import ControllerConfig, MqttConfig
|
|
from controller.mqtt_adapter import MqttAdapter
|
|
|
|
|
|
class MutableClock:
|
|
def __init__(self, value=0.0):
|
|
self.value = value
|
|
|
|
def __call__(self):
|
|
return self.value
|
|
|
|
|
|
class FakeMessageInfo:
|
|
def __init__(self, mid, published):
|
|
self.rc = mqtt.MQTT_ERR_SUCCESS
|
|
self.mid = mid
|
|
self.published = published
|
|
self.waited = []
|
|
|
|
def wait_for_publish(self, timeout=None):
|
|
self.waited.append(timeout)
|
|
|
|
def is_published(self):
|
|
return self.published
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.published = []
|
|
self.subscriptions = []
|
|
self.acks = []
|
|
self.suppress_exceptions = False
|
|
self.connect_error = None
|
|
self.connect_result = None
|
|
self.loop_result = mqtt.MQTT_ERR_SUCCESS
|
|
self.publish_acknowledged = None
|
|
self.publish_hook = None
|
|
|
|
def username_pw_set(self, username, password):
|
|
self.calls.append(("username_pw_set", username, password))
|
|
|
|
def manual_ack_set(self, enabled):
|
|
self.calls.append(("manual_ack_set", enabled))
|
|
|
|
def max_queued_messages_set(self, count):
|
|
self.calls.append(("max_queued_messages_set", count))
|
|
|
|
def reconnect_delay_set(self, **kwargs):
|
|
self.calls.append(("reconnect_delay_set", kwargs))
|
|
|
|
def connect_async(self, *args, **kwargs):
|
|
self.calls.append(("connect_async", args, kwargs))
|
|
if self.connect_error is not None:
|
|
raise self.connect_error
|
|
return self.connect_result
|
|
|
|
def connect(self, *args, **kwargs):
|
|
self.calls.append(("connect", args, kwargs))
|
|
raise AssertionError("synchronous connect must never be used")
|
|
|
|
def loop_start(self):
|
|
self.calls.append(("loop_start",))
|
|
return self.loop_result
|
|
|
|
def loop_stop(self):
|
|
self.calls.append(("loop_stop",))
|
|
|
|
def disconnect(self):
|
|
self.calls.append(("disconnect",))
|
|
|
|
def subscribe(self, topic, qos=0):
|
|
self.subscriptions.append((topic, qos))
|
|
return mqtt.MQTT_ERR_SUCCESS, len(self.subscriptions)
|
|
|
|
def publish(self, topic, payload, qos=0, retain=False):
|
|
self.published.append((topic, payload, qos, retain))
|
|
if self.publish_hook is not None:
|
|
self.publish_hook(topic, payload, qos, retain)
|
|
if self.publish_acknowledged is not None:
|
|
return FakeMessageInfo(
|
|
len(self.published), self.publish_acknowledged
|
|
)
|
|
return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=len(self.published))
|
|
|
|
def ack(self, message_id, qos):
|
|
self.acks.append((message_id, qos))
|
|
return mqtt.MQTT_ERR_SUCCESS
|
|
|
|
|
|
class MqttAdapterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = FakeClient()
|
|
self.events = queue.Queue()
|
|
self.config = ControllerConfig(
|
|
mqtt=MqttConfig(
|
|
host="broker",
|
|
port=1884,
|
|
username="user",
|
|
password="pass",
|
|
client_id="controller-test",
|
|
reconnect_min_seconds=1,
|
|
reconnect_max_seconds=2,
|
|
command_queue_size=8,
|
|
)
|
|
)
|
|
self.adapter = MqttAdapter(
|
|
self.config,
|
|
client=self.client,
|
|
events=self.events,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.adapter.stop()
|
|
|
|
def test_start_is_asynchronous_and_configures_manual_ack(self):
|
|
self.adapter.start()
|
|
names = [call[0] for call in self.client.calls]
|
|
self.assertIn("manual_ack_set", names)
|
|
self.assertIn("connect_async", names)
|
|
self.assertIn("loop_start", names)
|
|
self.assertNotIn("connect", names)
|
|
connect = next(call for call in self.client.calls if call[0] == "connect_async")
|
|
self.assertEqual(connect[1], ("broker", 1884))
|
|
self.assertEqual(connect[2], {"keepalive": 30})
|
|
|
|
def test_paho_two_connect_async_none_starts_network_loop(self):
|
|
self.client.connect_result = None
|
|
self.adapter.start()
|
|
|
|
self.assertTrue(self.adapter.started)
|
|
self.assertIn("loop_start", [call[0] for call in self.client.calls])
|
|
self.assertIsNone(self.adapter.last_error)
|
|
|
|
def test_initial_setup_oserror_does_not_escape_startup(self):
|
|
self.client.connect_error = OSError("network unreachable")
|
|
self.adapter.start()
|
|
self.assertTrue(self.adapter.started)
|
|
self.assertIn("network unreachable", self.adapter.last_error)
|
|
self.assertNotIn("loop_start", [call[0] for call in self.client.calls])
|
|
|
|
def test_loop_start_failure_does_not_escape_startup(self):
|
|
self.client.loop_result = mqtt.MQTT_ERR_UNKNOWN
|
|
self.adapter.start()
|
|
self.assertTrue(self.adapter.started)
|
|
self.assertIn("failed to start MQTT network loop", self.adapter.last_error)
|
|
|
|
def test_stop_during_blocked_connect_never_starts_network_loop_afterward(self):
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
original_connect = self.client.connect_async
|
|
|
|
def blocked_connect(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(timeout=2)
|
|
return original_connect(*args, **kwargs)
|
|
|
|
self.client.connect_async = blocked_connect
|
|
starter = threading.Thread(target=self.adapter.start)
|
|
starter.start()
|
|
self.assertTrue(entered.wait(timeout=1))
|
|
self.adapter.stop()
|
|
release.set()
|
|
starter.join(timeout=2)
|
|
self.assertFalse(starter.is_alive())
|
|
self.assertNotIn("loop_start", [call[0] for call in self.client.calls])
|
|
self.assertFalse(self.adapter.started)
|
|
|
|
def test_stop_wins_against_connect_callback_blocked_during_subscribe(self):
|
|
self.adapter.start()
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
original_subscribe = self.client.subscribe
|
|
|
|
def blocked_subscribe(topic, qos=0):
|
|
entered.set()
|
|
release.wait(timeout=2)
|
|
return original_subscribe(topic, qos=qos)
|
|
|
|
self.client.subscribe = blocked_subscribe
|
|
callback = threading.Thread(
|
|
target=self.adapter._on_connect,
|
|
args=(self.client, None, None, 0),
|
|
)
|
|
callback.start()
|
|
self.assertTrue(entered.wait(timeout=1))
|
|
stopper = threading.Thread(target=self.adapter.stop)
|
|
stopper.start()
|
|
release.set()
|
|
callback.join(timeout=2)
|
|
stopper.join(timeout=2)
|
|
self.assertFalse(callback.is_alive())
|
|
self.assertFalse(stopper.is_alive())
|
|
self.assertFalse(self.adapter.started)
|
|
self.assertFalse(self.adapter.connected)
|
|
self.assertLess(len(self.client.subscriptions), 6)
|
|
self.assertTrue(self.events.empty())
|
|
self.assertFalse(
|
|
self.adapter.publish_player_command({"action": "stop"}).accepted
|
|
)
|
|
|
|
def test_connect_subscribes_to_all_authoritative_inputs(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.assertTrue(self.adapter.connected)
|
|
self.assertEqual(
|
|
self.client.subscriptions,
|
|
[
|
|
("youtube/player/status", 1),
|
|
("youtube/player/telemetry", 0),
|
|
("youtube/player/state", 0),
|
|
("youtube/controller/cmd", 1),
|
|
("audio/player/mode/set", 1),
|
|
("audio/player/mode/state", 1),
|
|
],
|
|
)
|
|
event = self.events.get_nowait()
|
|
self.assertEqual(event.kind, "connected")
|
|
self.assertEqual(event.connection_generation, 1)
|
|
|
|
def test_player_commands_are_exact_qos_one_and_nonretained(self):
|
|
disconnected = self.adapter.publish_player_command({"action": "stop"})
|
|
self.assertFalse(disconnected.accepted)
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
receipt = self.adapter.publish_player_command(
|
|
{"action": "play", "slot": 3, "label": "Opening"}
|
|
)
|
|
self.assertTrue(receipt.accepted)
|
|
topic, payload, qos, retain = self.client.published[-1]
|
|
self.assertEqual(topic, "youtube/player/cmd")
|
|
self.assertEqual(json.loads(payload), {"action": "play", "slot": 3, "label": "Opening"})
|
|
self.assertEqual(qos, 1)
|
|
self.assertIs(retain, False)
|
|
|
|
def test_router_commands_are_qos_one_nonretained_plain_strings(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
receipt = self.adapter.publish_router_mode("video")
|
|
self.assertTrue(receipt.accepted)
|
|
self.assertEqual(
|
|
self.client.published[-1],
|
|
("audio/player/mode/set", "video", 1, False),
|
|
)
|
|
|
|
def test_router_self_echo_is_marked_while_publish_is_in_flight(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
|
|
def echo(topic, payload, qos, retain):
|
|
self.adapter._on_message(
|
|
self.client,
|
|
None,
|
|
SimpleNamespace(
|
|
topic=topic,
|
|
payload=payload.encode("utf-8"),
|
|
qos=qos,
|
|
retain=retain,
|
|
dup=False,
|
|
mid=31,
|
|
),
|
|
)
|
|
|
|
self.client.publish_hook = echo
|
|
self.assertTrue(self.adapter.publish_router_mode("video").accepted)
|
|
self.assertTrue(self.events.get_nowait().controller_echo)
|
|
|
|
def test_router_self_echo_delivered_after_puback_is_still_claimed_within_grace(
|
|
self,
|
|
):
|
|
# Regression: a broker may legally deliver our own subscription echo
|
|
# after the PUBACK that already acknowledged the same publish. If the
|
|
# echo marker were cleared the instant publish() observes that
|
|
# PUBACK, this late echo would be mistaken for live operator intent
|
|
# and cause automatic Router restoration ownership to be relinquished.
|
|
clock = MutableClock(1000.0)
|
|
client = FakeClient()
|
|
events = queue.Queue()
|
|
adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
|
|
try:
|
|
adapter.start()
|
|
adapter._on_connect(client, None, None, 0)
|
|
events.get_nowait()
|
|
|
|
self.assertTrue(adapter.publish_router_mode("video").accepted)
|
|
clock.value += self.config.mqtt.router_echo_grace_seconds - 0.5
|
|
|
|
adapter._on_message(
|
|
client,
|
|
None,
|
|
SimpleNamespace(
|
|
topic="audio/player/mode/set",
|
|
payload=b"video",
|
|
qos=1,
|
|
retain=False,
|
|
dup=False,
|
|
mid=32,
|
|
),
|
|
)
|
|
self.assertTrue(events.get_nowait().controller_echo)
|
|
finally:
|
|
adapter.stop()
|
|
|
|
def test_router_self_echo_marker_expires_after_grace_window(self):
|
|
# A genuine external/operator command that happens to repeat the same
|
|
# value must still be recognized as operator intent once the bounded
|
|
# grace window has elapsed; the marker must not linger forever.
|
|
clock = MutableClock(1000.0)
|
|
client = FakeClient()
|
|
events = queue.Queue()
|
|
adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
|
|
try:
|
|
adapter.start()
|
|
adapter._on_connect(client, None, None, 0)
|
|
events.get_nowait()
|
|
|
|
self.assertTrue(adapter.publish_router_mode("video").accepted)
|
|
clock.value += self.config.mqtt.router_echo_grace_seconds + 0.5
|
|
|
|
adapter._on_message(
|
|
client,
|
|
None,
|
|
SimpleNamespace(
|
|
topic="audio/player/mode/set",
|
|
payload=b"video",
|
|
qos=1,
|
|
retain=False,
|
|
dup=False,
|
|
mid=32,
|
|
),
|
|
)
|
|
self.assertFalse(events.get_nowait().controller_echo)
|
|
finally:
|
|
adapter.stop()
|
|
|
|
def test_retained_router_set_never_consumes_controller_echo_marker(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
|
|
def retained_then_live_echo(topic, payload, qos, retain):
|
|
for message_retain, mid in ((True, 41), (False, 42)):
|
|
self.adapter._on_message(
|
|
self.client,
|
|
None,
|
|
SimpleNamespace(
|
|
topic=topic,
|
|
payload=payload.encode("utf-8"),
|
|
qos=qos,
|
|
retain=message_retain,
|
|
dup=False,
|
|
mid=mid,
|
|
),
|
|
)
|
|
|
|
self.client.publish_hook = retained_then_live_echo
|
|
self.assertTrue(self.adapter.publish_router_mode("video").accepted)
|
|
retained_event = self.events.get_nowait()
|
|
live_event = self.events.get_nowait()
|
|
self.assertTrue(retained_event.retain)
|
|
self.assertFalse(retained_event.controller_echo)
|
|
self.assertFalse(live_event.retain)
|
|
self.assertTrue(live_event.controller_echo)
|
|
|
|
def test_qos_message_is_acked_only_after_consumer_calls_acknowledge(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
message = SimpleNamespace(
|
|
topic="youtube/controller/cmd",
|
|
payload=b'{"action":"stop"}',
|
|
qos=1,
|
|
retain=False,
|
|
mid=42,
|
|
)
|
|
self.adapter._on_message(self.client, None, message)
|
|
event = self.events.get_nowait()
|
|
self.assertEqual(self.client.acks, [])
|
|
self.assertEqual(event.message_id, 42)
|
|
self.assertTrue(self.adapter.acknowledge(event))
|
|
self.assertEqual(self.client.acks, [(42, 1)])
|
|
|
|
def test_stale_mid_from_previous_connection_is_not_acked(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
message = SimpleNamespace(
|
|
topic="youtube/controller/cmd",
|
|
payload=b'{"action":"stop"}',
|
|
qos=1,
|
|
retain=False,
|
|
mid=7,
|
|
)
|
|
self.adapter._on_message(self.client, None, message)
|
|
old_event = self.events.get_nowait()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
self.assertFalse(self.adapter.acknowledge(old_event))
|
|
self.assertEqual(self.client.acks, [])
|
|
|
|
def test_retained_flag_is_preserved_for_coordinator_rejection(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
message = SimpleNamespace(
|
|
topic="youtube/controller/cmd",
|
|
payload=b'{"action":"delete_slot","slot":1}',
|
|
qos=1,
|
|
retain=True,
|
|
mid=8,
|
|
)
|
|
self.adapter._on_message(self.client, None, message)
|
|
self.assertIs(self.events.get_nowait().retain, True)
|
|
|
|
def test_duplicate_flag_is_preserved_for_coordinator_deduplication(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
message = SimpleNamespace(
|
|
topic="youtube/controller/cmd",
|
|
payload=b'{"action":"toggle"}',
|
|
qos=1,
|
|
retain=False,
|
|
dup=True,
|
|
mid=9,
|
|
)
|
|
self.adapter._on_message(self.client, None, message)
|
|
self.assertIs(self.events.get_nowait().duplicate, True)
|
|
|
|
def test_missing_puback_is_uncertain_and_invalidates_connection(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
self.client.publish_acknowledged = False
|
|
receipt = self.adapter.publish_player_command({"action": "stop"})
|
|
self.assertFalse(receipt.accepted)
|
|
self.assertTrue(receipt.uncertain)
|
|
self.assertIn("no broker PUBACK", receipt.reason)
|
|
self.assertFalse(self.adapter.connected)
|
|
|
|
def test_optional_qos_event_puback_failure_does_not_invalidate_transport(self):
|
|
self.adapter.start()
|
|
self.adapter._on_connect(self.client, None, None, 0)
|
|
self.events.get_nowait()
|
|
self.client.publish_acknowledged = False
|
|
receipt = self.adapter.publish_json(
|
|
"youtube/controller/v2/events",
|
|
{"version": 2, "type": "warning"},
|
|
qos=1,
|
|
retain=False,
|
|
)
|
|
self.assertFalse(receipt.accepted)
|
|
self.assertTrue(receipt.uncertain)
|
|
self.assertTrue(self.adapter.connected)
|
|
|
|
def test_discarded_transport_releases_callback_blocked_on_full_queue(self):
|
|
events = queue.Queue(maxsize=1)
|
|
client = FakeClient()
|
|
adapter = MqttAdapter(self.config, client=client, events=events)
|
|
try:
|
|
adapter.start()
|
|
adapter._on_connect(client, None, None, 0)
|
|
self.assertTrue(events.full())
|
|
callback = threading.Thread(
|
|
target=adapter._on_message,
|
|
args=(
|
|
client,
|
|
None,
|
|
SimpleNamespace(
|
|
topic="youtube/controller/cmd",
|
|
payload=b'{"action":"stop"}',
|
|
qos=1,
|
|
retain=False,
|
|
dup=False,
|
|
mid=99,
|
|
),
|
|
),
|
|
)
|
|
callback.start()
|
|
callback.join(timeout=0.05)
|
|
self.assertTrue(callback.is_alive())
|
|
|
|
adapter._discard_unacknowledged_transport("test transport discard")
|
|
callback.join(timeout=1)
|
|
self.assertFalse(callback.is_alive())
|
|
self.assertEqual(events.qsize(), 1)
|
|
self.assertEqual(events.get_nowait().kind, "connected")
|
|
finally:
|
|
adapter.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|