diff --git a/frigate/app.py b/frigate/app.py index 9f8192a31d..896bc7f307 100644 --- a/frigate/app.py +++ b/frigate/app.py @@ -335,6 +335,7 @@ class FrigateApp: self.ptz_metrics, comms, ) + self.dispatcher.start_communicators() def init_profile_manager(self) -> None: self.profile_manager = ProfileManager( diff --git a/frigate/comms/base_communicator.py b/frigate/comms/base_communicator.py index 090552a5bd..abb84fecf5 100644 --- a/frigate/comms/base_communicator.py +++ b/frigate/comms/base_communicator.py @@ -1,11 +1,27 @@ from abc import ABC, abstractmethod from collections.abc import Callable -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from frigate.comms.dispatcher import Dispatcher class Communicator(ABC): """pub/sub model via specific protocol.""" + def attach_dispatcher(self, dispatcher: "Dispatcher") -> None: + """Receive the owning dispatcher. + + Transports that need more than the receiver callback (the command topic + surface, the snapshot API) take it here rather than reaching through the + bound receiver. + """ + return None + + def start(self) -> None: + """Start background I/O after receiver wiring is complete.""" + return None + @abstractmethod def publish(self, topic: str, payload: Any, retain: bool = False) -> None: """Send data via specific protocol.""" diff --git a/frigate/comms/dispatcher.py b/frigate/comms/dispatcher.py index 1aedaa6e61..901d77b279 100644 --- a/frigate/comms/dispatcher.py +++ b/frigate/comms/dispatcher.py @@ -49,6 +49,11 @@ from frigate.util.services import restart_frigate logger = logging.getLogger(__name__) +# ///set, one segment longer than the rest +SUB_COMMAND_TOPICS = frozenset({"motion_mask", "object_mask", "zone"}) + +BARE_COMMAND_TOPICS = frozenset({"onConnect", "restart"}) + class Dispatcher: """Handle communication between Frigate and communicators.""" @@ -103,12 +108,114 @@ class Dispatcher: } self.profile_manager: ProfileManager | None = None - for comm in self.comms: - comm.subscribe(self._receive) - self.web_push_client = next( (comm for comm in communicators if isinstance(comm, WebPushClient)), None ) + + for comm in self.comms: + comm.subscribe(self._receive) + comm.attach_dispatcher(self) + + def start_communicators(self) -> None: + """Start communicators after dispatcher wiring is fully initialized.""" + for comm in self.comms: + comm.start() + + def is_command_topic(self, topic: str) -> bool: + """Whether a prefix-stripped topic maps to a command handler. + + Transports that fan a whole topic tree in must filter on this: + _receive() republishes anything it does not recognize, so forwarding + unfiltered would echo Frigate's own publishes back. + """ + parts = topic.split("/") + + if topic in BARE_COMMAND_TOPICS: + return True + + if len(parts) == 2 and parts[1] == "ptz": + return True + + if len(parts) == 2 and parts[1] == "set": + return parts[0] in self._global_settings_handlers + + if len(parts) == 3 and parts[2] == "set": + return ( + parts[1] in self._camera_settings_handlers + and parts[1] not in SUB_COMMAND_TOPICS + ) + + if len(parts) == 3 and parts[2] == "suspend": + return parts[1] == "notifications" + + if len(parts) == 4 and parts[3] == "set": + return parts[1] in SUB_COMMAND_TOPICS + + return False + + def _build_camera_activity_snapshot(self) -> tuple[dict[str, Any], dict[str, Any]]: + """Build the current runtime activity snapshot for reconnect consumers.""" + camera_status = { + camera: status + for camera, status in self.camera_activity.last_camera_activity.copy().items() + if camera in self.config.cameras + } + audio_detections = self.audio_activity.current_audio_detections.copy() + cameras_with_status = camera_status.keys() + + for camera in self.config.cameras.keys(): + if camera not in cameras_with_status: + camera_status[camera] = {} + + camera_status[camera]["config"] = { + "detect": self.config.cameras[camera].detect.enabled, + "enabled": self.config.cameras[camera].enabled, + "snapshots": self.config.cameras[camera].snapshots.enabled, + "record": self.config.cameras[camera].record.enabled, + "audio": self.config.cameras[camera].audio.enabled, + "audio_transcription": self.config.cameras[ + camera + ].audio_transcription.live_enabled, + "notifications": self.config.cameras[camera].notifications.enabled, + "notifications_suspended": int( + self.web_push_client.suspended_cameras.get(camera, 0) + ) + if self.web_push_client + and camera in self.web_push_client.suspended_cameras + else 0, + "autotracking": self.config.cameras[camera].onvif.autotracking.enabled, + "alerts": self.config.cameras[camera].review.alerts.enabled, + "detections": self.config.cameras[camera].review.detections.enabled, + "object_descriptions": self.config.cameras[ + camera + ].objects.genai.enabled, + "review_descriptions": self.config.cameras[camera].review.genai.enabled, + } + + return camera_status, audio_detections + + def publish_runtime_snapshot( + self, + publisher: Callable[[str, Any, bool], None] | None = None, + ) -> None: + """Publish the runtime snapshot for newly connected listeners.""" + publish = publisher or self.publish + camera_status, audio_detections = self._build_camera_activity_snapshot() + + publish("camera_activity", json.dumps(camera_status), False) + publish("model_state", json.dumps(self.model_state.copy()), False) + publish( + "embeddings_reindex_progress", + json.dumps(self.embeddings_reindex.copy()), + False, + ) + publish("birdseye_layout", json.dumps(self.birdseye_layout.copy()), False) + publish("audio_detections", json.dumps(audio_detections), False) + publish( + "profile/state", + self.config.active_profile or "none", + True, + ) if self.web_push_client is not None: self.web_push_client.set_suspension_broadcaster(self.publish) @@ -127,17 +234,11 @@ class Dispatcher: try: if command_type == "set": - # Commands that require a sub-command (mask/zone name) - sub_command_required = { - "motion_mask", - "object_mask", - "zone", - } if sub_command: self._camera_settings_handlers[command]( camera_name, sub_command, payload ) - elif command in sub_command_required: + elif command in SUB_COMMAND_TOPICS: logger.error( "Command %s requires a sub-command (mask/zone name)", command, @@ -271,67 +372,11 @@ class Dispatcher: def handle_birdseye_layout() -> None: self.publish("birdseye_layout", json.dumps(self.birdseye_layout.copy())) - def handle_on_connect() -> None: - camera_status = { - camera: status - for camera, status in self.camera_activity.last_camera_activity.copy().items() - if camera in self.config.cameras - } - audio_detections = self.audio_activity.current_audio_detections.copy() - cameras_with_status = camera_status.keys() - - for camera in self.config.cameras.keys(): - if camera not in cameras_with_status: - camera_status[camera] = {} - - camera_status[camera]["config"] = { - "detect": self.config.cameras[camera].detect.enabled, - "enabled": self.config.cameras[camera].enabled, - "snapshots": self.config.cameras[camera].snapshots.enabled, - "record": self.config.cameras[camera].record.enabled, - "audio": self.config.cameras[camera].audio.enabled, - "audio_transcription": self.config.cameras[ - camera - ].audio_transcription.live_enabled, - "notifications": self.config.cameras[camera].notifications.enabled, - "notifications_suspended": int( - self.web_push_client.suspended_cameras.get(camera, 0) - ) - if self.web_push_client - and camera in self.web_push_client.suspended_cameras - else 0, - "autotracking": self.config.cameras[ - camera - ].onvif.autotracking.enabled, - "alerts": self.config.cameras[camera].review.alerts.enabled, - "detections": self.config.cameras[camera].review.detections.enabled, - "object_descriptions": self.config.cameras[ - camera - ].objects.genai.enabled, - "review_descriptions": self.config.cameras[ - camera - ].review.genai.enabled, - } - - self.publish("camera_activity", json.dumps(camera_status)) - self.publish("model_state", json.dumps(self.model_state.copy())) - self.publish( - "embeddings_reindex_progress", - json.dumps(self.embeddings_reindex.copy()), - ) - self.publish("birdseye_layout", json.dumps(self.birdseye_layout.copy())) - self.publish("audio_detections", json.dumps(audio_detections)) - self.publish( - "profile/state", - self.config.active_profile or "none", - retain=True, - ) - def handle_notification_test() -> None: self.publish("notification_test", "Test notification") # Dictionary mapping topic to handlers - topic_handlers = { + topic_handlers: dict[str, Callable[[], Any]] = { INSERT_MANY_RECORDINGS: handle_insert_many_recordings, REQUEST_REGION_GRID: handle_request_region_grid, INSERT_PREVIEW: handle_insert_preview, @@ -354,7 +399,7 @@ class Dispatcher: "jobState": handle_job_state, "audioTranscriptionState": handle_audio_transcription_state, "birdseyeLayout": handle_birdseye_layout, - "onConnect": handle_on_connect, + "onConnect": self.publish_runtime_snapshot, } if topic.endswith("set") or topic.endswith("ptz") or topic.endswith("suspend"): diff --git a/frigate/comms/inter_process.py b/frigate/comms/inter_process.py index 6897b12ebf..5c663ef666 100644 --- a/frigate/comms/inter_process.py +++ b/frigate/comms/inter_process.py @@ -18,10 +18,13 @@ SOCKET_REP_REQ = "ipc:///tmp/cache/comms" class InterProcessCommunicator(Communicator): def __init__(self) -> None: + # bound eagerly so subprocesses starting before start_communicators() + # can still connect; their requests queue in zmq until the reader runs self.context = zmq.Context() self.socket = self.context.socket(zmq.REP) self.socket.bind(SOCKET_REP_REQ) self.stop_event: MpEvent = mp.Event() + self.reader_thread: threading.Thread | None = None def publish(self, topic: str, payload: Any, retain: bool = False) -> None: """There is no communication back to the processes.""" @@ -29,6 +32,8 @@ class InterProcessCommunicator(Communicator): def subscribe(self, receiver: Callable) -> None: self._dispatcher = receiver + + def start(self) -> None: self.reader_thread = threading.Thread(target=self.read) self.reader_thread.start() @@ -61,7 +66,8 @@ class InterProcessCommunicator(Communicator): def stop(self) -> None: self.stop_event.set() - self.reader_thread.join() + if self.reader_thread is not None: + self.reader_thread.join() self.socket.close(linger=0) self.context.destroy(linger=0) diff --git a/frigate/comms/mqtt.py b/frigate/comms/mqtt.py index c7ef3e6441..f2ce4f7c6d 100644 --- a/frigate/comms/mqtt.py +++ b/frigate/comms/mqtt.py @@ -1,7 +1,12 @@ +from __future__ import annotations + import logging +import queue import threading +import time from collections.abc import Callable -from typing import Any +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any import paho.mqtt.client as mqtt from paho.mqtt.enums import CallbackAPIVersion @@ -9,8 +14,25 @@ from paho.mqtt.enums import CallbackAPIVersion from frigate.comms.base_communicator import Communicator from frigate.config import FrigateConfig, birdseye_modes_to_mqtt_payload +if TYPE_CHECKING: + from frigate.comms.dispatcher import Dispatcher + logger = logging.getLogger(__name__) +MQTT_LOOP_TIMEOUT = 1.0 +MQTT_RECONNECT_INTERVAL = 10.0 +MQTT_SHUTDOWN_FLUSH_TIMEOUT = 5.0 +MQTT_ON_CONNECT_RATE_LIMIT = 1.0 +MQTT_PUBLISH_WAIT_INTERVAL = 0.1 + + +@dataclass(slots=True) +class QueuedPublish: + topic: str + payload: Any + retain: bool + done: threading.Event | None = None + class MqttClient(Communicator): """Frigate wrapper for mqtt client.""" @@ -19,28 +41,80 @@ class MqttClient(Communicator): self.config = config self.mqtt_config = config.mqtt self.connected = False + self.client: mqtt.Client | None = None + self._dispatcher: Callable[[str, Any], Any] | None = None + self._command_router: Dispatcher | None = None + self._worker: threading.Thread | None = None + self._stop_event = threading.Event() + self._publish_queue: queue.Queue[QueuedPublish] = queue.Queue() + self._callback_queue: queue.Queue[tuple[Any, ...]] = queue.Queue() + self._retained_lock = threading.Lock() + self._pending_retained: dict[str, tuple[Any, bool]] = {} + self._inflight_retained: dict[int, tuple[str, Any]] = {} + self._subscription_mid: int | None = None + self._subscription_ready = False + self._next_connect_time = 0.0 + self._last_on_connect_dispatch = 0.0 def subscribe(self, receiver: Callable) -> None: """Wrapper for allowing dispatcher to subscribe.""" self._dispatcher = receiver - self._start() + + def attach_dispatcher(self, dispatcher: Dispatcher) -> None: + """Take Dispatcher's command surface and snapshot API.""" + self._command_router = dispatcher + + def start(self) -> None: + """Start the MQTT worker after all receiver wiring is complete.""" + + if self._worker and self._worker.is_alive(): + return + + self._stop_event.clear() + self._start_worker() def publish(self, topic: str, payload: Any, retain: bool = False) -> None: """Wrapper for publishing when client is in valid state.""" + full_topic = f"{self.mqtt_config.topic_prefix}/{topic}" + if not self.connected: - logger.debug(f"Unable to publish to {topic}: client is not connected") + if retain: + self._queue_retained(full_topic, payload, retain) + else: + logger.debug("Unable to publish to %s: client is not connected", topic) return - self.client.publish( - f"{self.mqtt_config.topic_prefix}/{topic}", - payload, - qos=self.config.mqtt.qos, - retain=retain, - ) + self._publish_queue.put(QueuedPublish(full_topic, payload, retain)) def stop(self) -> None: - self.publish("available", "stopped", retain=True) - self.client.disconnect() + if self._worker is None: + return + + if self.connected and self._subscription_ready: + publish_done = threading.Event() + self._publish_queue.put( + QueuedPublish( + f"{self.mqtt_config.topic_prefix}/available", + "stopped", + True, + publish_done, + ) + ) + publish_done.wait(MQTT_SHUTDOWN_FLUSH_TIMEOUT) + + self._stop_event.set() + + if self.client is not None: + try: + self.client.disconnect() + except Exception: + logger.debug("MQTT disconnect raised during shutdown", exc_info=True) + + if self._worker.is_alive(): + self._worker.join(MQTT_SHUTDOWN_FLUSH_TIMEOUT + MQTT_LOOP_TIMEOUT) + + self._cleanup_client() + self._worker = None def _notifications_enabled_in_config(self) -> bool: """Whether notifications are configured globally or on any camera. @@ -54,17 +128,17 @@ class MqttClient(Communicator): for cam in self.config.cameras.values() ) - def _set_initial_topics(self) -> None: - """Set initial state topics.""" + def _publish_retained_state(self) -> None: + """Publish retained MQTT state after a successful subscribe.""" for camera_name, camera in self.config.cameras.items(): self.publish( f"{camera_name}/enabled/state", - "ON" if camera.enabled_in_config else "OFF", + "ON" if camera.enabled else "OFF", retain=True, ) self.publish( f"{camera_name}/recordings/state", - "ON" if camera.record.enabled_in_config else "OFF", + "ON" if camera.record.enabled else "OFF", retain=True, ) self.publish( @@ -74,7 +148,7 @@ class MqttClient(Communicator): ) self.publish( f"{camera_name}/audio/state", - "ON" if camera.audio.enabled_in_config else "OFF", + "ON" if camera.audio.enabled else "OFF", retain=True, ) self.publish( @@ -89,7 +163,7 @@ class MqttClient(Communicator): ) self.publish( f"{camera_name}/motion/state", - "ON", + "ON" if camera.motion.enabled else "OFF", retain=True, ) self.publish( @@ -99,7 +173,7 @@ class MqttClient(Communicator): ) self.publish( f"{camera_name}/ptz_autotracker/state", - "ON" if camera.onvif.autotracking.enabled_in_config else "OFF", + "ON" if camera.onvif.autotracking.enabled else "OFF", retain=True, ) self.publish( @@ -133,22 +207,22 @@ class MqttClient(Communicator): ) self.publish( f"{camera_name}/review_alerts/state", - "ON" if camera.review.alerts.enabled_in_config else "OFF", + "ON" if camera.review.alerts.enabled else "OFF", retain=True, ) self.publish( f"{camera_name}/review_detections/state", - "ON" if camera.review.detections.enabled_in_config else "OFF", + "ON" if camera.review.detections.enabled else "OFF", retain=True, ) self.publish( f"{camera_name}/object_descriptions/state", - "ON" if camera.objects.genai.enabled_in_config else "OFF", + "ON" if camera.objects.genai.enabled else "OFF", retain=True, ) self.publish( f"{camera_name}/review_descriptions/state", - "ON" if camera.review.genai.enabled_in_config else "OFF", + "ON" if camera.review.genai.enabled else "OFF", retain=True, ) @@ -189,13 +263,521 @@ class MqttClient(Communicator): ) self.publish("available", "online", retain=True) - def on_mqtt_command( - self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage - ) -> None: - self._dispatcher( - message.topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1), - message.payload.decode(), + def _create_client(self) -> mqtt.Client: + """Build a fresh paho client for a single connect attempt.""" + client = mqtt.Client( + callback_api_version=CallbackAPIVersion.VERSION2, + client_id=self.mqtt_config.client_id, + reconnect_on_failure=False, ) + client.on_connect = self._on_connect + client.on_disconnect = self._on_disconnect + client.on_message = self._on_message + client.on_subscribe = self._on_subscribe + client.on_publish = self._on_publish + client.will_set( + self.mqtt_config.topic_prefix + "/available", + payload="offline", + qos=1, + retain=True, + ) + + if self.mqtt_config.tls_ca_certs is not None: + if ( + self.mqtt_config.tls_client_cert is not None + and self.mqtt_config.tls_client_key is not None + ): + client.tls_set( + self.mqtt_config.tls_ca_certs, + self.mqtt_config.tls_client_cert, + self.mqtt_config.tls_client_key, + ) + else: + client.tls_set(self.mqtt_config.tls_ca_certs) + + if self.mqtt_config.tls_insecure is not None: + client.tls_insecure_set(self.mqtt_config.tls_insecure) + + if self.mqtt_config.user is not None: + client.username_pw_set( + self.mqtt_config.user, + password=self.mqtt_config.password, + ) + + return client + + def _start_worker(self) -> None: + self._worker = threading.Thread( + target=self._worker_main, name="mqtt", daemon=True + ) + self._worker.start() + logger.info("MQTT worker started") + + def _worker_main(self) -> None: + """Run the worker loop. + + An unexpected crash disables MQTT for this session rather than taking + Frigate down with it, so it has to announce itself: without the offline + publish, consumers keep the last retained values and see a healthy + Frigate that has simply stopped updating. + """ + try: + self._mqtt_loop_worker() + except Exception: + if not self._stop_event.is_set(): + logger.exception("MQTT worker crashed, disabling MQTT for this session") + self._stop_event.set() + self._subscription_ready = False + self._publish_offline_availability() + self.connected = False + finally: + # nothing drains the queue once the loop is gone, so release any + # waiter here or stop() blocks for the full flush timeout + self._requeue_disconnected_publishes() + self._cleanup_client() + + def _publish_offline_availability(self) -> None: + """Announce that MQTT is going away after a worker crash. + + _cleanup_client() disconnects cleanly, which tells the broker to + suppress the will, so the retained topic would otherwise stay "online". + """ + if self.client is None: + return + + try: + message_info = self.client.publish( + f"{self.mqtt_config.topic_prefix}/available", + "offline", + qos=self.config.mqtt.qos, + retain=True, + ) + + # pumped here rather than through _wait_for_publish() so the drain + # that may have just crashed is not re-entered + deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT + while not message_info.is_published() and time.monotonic() < deadline: + if ( + self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL) + != mqtt.MQTT_ERR_SUCCESS + ): + break + except Exception: + logger.warning( + "MQTT is dormant and the broker could not be told Frigate is offline", + exc_info=True, + ) + + def _mqtt_loop_worker(self) -> None: + # The worker owns all socket I/O so reconnect, subscribe, and publish + # ordering stays serialized in one place. + while not self._stop_event.is_set(): + if self.client is None: + wait_time = self._next_connect_time - time.monotonic() + if wait_time > 0: + self._stop_event.wait(min(wait_time, MQTT_LOOP_TIMEOUT)) + continue + + if not self._connect_client(): + self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL + continue + + assert self.client is not None + try: + result = self.client.loop(timeout=MQTT_LOOP_TIMEOUT) + except (OSError, mqtt.WebsocketConnectionError) as err: + logger.warning("MQTT loop error: %s", err) + self._schedule_reconnect() + continue + + self._drain_callback_queue() + self._drain_publish_queue() + + if self._stop_event.is_set(): + break + + if result != mqtt.MQTT_ERR_SUCCESS and self.client is not None: + logger.error("MQTT loop returned error code: %s", result) + self._schedule_reconnect() + + def _connect_client(self) -> bool: + """Create and connect a new client instance owned by the worker thread.""" + try: + self.client = self._create_client() + self.client.connect(self.mqtt_config.host, self.mqtt_config.port, 60) + except Exception as err: + logger.error("Unable to connect to MQTT server: %s", err) + self._cleanup_client() + return False + + return True + + def _cleanup_client(self) -> None: + """Drop session-specific state and release the current paho client.""" + self.connected = False + self._subscription_ready = False + self._subscription_mid = None + self._requeue_inflight_retained() + + client = self.client + self.client = None + + if client is None: + return + + try: + client.disconnect() + except Exception: + logger.debug("MQTT client cleanup raised disconnect error", exc_info=True) + + def _schedule_reconnect(self) -> None: + """Tear down the current session and arm the next reconnect attempt.""" + if self._stop_event.is_set(): + return + + self.connected = False + self._subscription_ready = False + self._subscription_mid = None + self._requeue_disconnected_publishes() + self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL + logger.info("MQTT reconnect scheduled in %.1fs", MQTT_RECONNECT_INTERVAL) + self._cleanup_client() + + def _requeue_inflight_retained(self) -> None: + """Rebuffer retained publishes paho took but the broker never acked. + + Dropping the client drops paho's outbound queue with it, and the session + is clean, so the broker will not resume delivery on the new one. + """ + with self._retained_lock: + # mids are insertion ordered, so collapsing by topic keeps the + # newest value when several updates to one topic were in flight + latest = { + topic: payload for topic, payload in self._inflight_retained.values() + } + self._inflight_retained.clear() + + for topic, payload in latest.items(): + self._queue_retained(topic, payload, True, overwrite=False) + + def _buffer_undelivered( + self, queued_publish: QueuedPublish, overwrite: bool = True + ) -> None: + """Handle a publish that never reached the broker. + + Releasing the waiter matters on every path: stop() blocks on it, so a + broker error would otherwise stall shutdown for the full flush timeout. + """ + if queued_publish.retain: + self._queue_retained( + queued_publish.topic, + queued_publish.payload, + queued_publish.retain, + overwrite=overwrite, + ) + + if queued_publish.done is not None: + queued_publish.done.set() + + def _requeue_disconnected_publishes(self) -> None: + while True: + try: + queued_publish = self._publish_queue.get_nowait() + except queue.Empty: + break + + self._buffer_undelivered(queued_publish) + + def _drain_callback_queue(self) -> None: + # Paho callbacks only enqueue transport events; state transitions run + # here on the worker thread. + while True: + try: + event = self._callback_queue.get_nowait() + except queue.Empty: + break + + event_type = event[0] + + if event_type == "connect": + self._handle_connect_event(event[1]) + elif event_type == "connect_failure": + self._handle_connect_failure(event[1]) + elif event_type == "disconnect": + self._handle_disconnect_event(event[1]) + elif event_type == "subscribed": + self._handle_subscribe_event(event[1], event[2]) + elif event_type == "message": + self._handle_inbound_message(event[1], event[2]) + elif event_type == "published": + self._handle_publish_event(event[1]) + + def _drain_publish_queue(self) -> None: + """Publish queued work only after the session is fully subscribed. + + Oldest first: the outage buffer replays before the queue, so a topic + that changed since the reconnect ends up on its newest value rather + than being reverted by the replay. + """ + if self.connected and not self._subscription_ready: + return + + self._flush_pending_retained() + + while True: + try: + queued_publish = self._publish_queue.get_nowait() + except queue.Empty: + break + + if not self.connected: + self._buffer_undelivered(queued_publish) + continue + + self._publish_direct(queued_publish) + + def _flush_pending_retained(self) -> None: + """Replay the latest retained state once the broker session is ready.""" + if not self.connected or not self._subscription_ready: + return + + with self._retained_lock: + pending = list(self._pending_retained.items()) + self._pending_retained.clear() + + for topic, (payload, retain) in pending: + self._publish_direct(QueuedPublish(topic, payload, retain)) + + def _publish_direct(self, queued_publish: QueuedPublish) -> None: + """Publish a queued message from the worker thread's serialized context. + + The waiter is released however this exits. The message is already off + the queue by now, so nothing else can recover it for a stop() that is + blocked waiting on it. + """ + try: + if self.client is None: + # never attempted, so anything already buffered for this topic + # was written later and has to survive + self._buffer_undelivered(queued_publish, overwrite=False) + return + + try: + message_info = self.client.publish( + queued_publish.topic, + queued_publish.payload, + qos=self.config.mqtt.qos, + retain=queued_publish.retain, + ) + except (OSError, mqtt.WebsocketConnectionError) as err: + logger.warning( + "MQTT publish failed for %s: %s", queued_publish.topic, err + ) + # a newer buffered value for this topic wins over the failed one + self._buffer_undelivered(queued_publish, overwrite=False) + self._schedule_reconnect() + return + + if message_info.rc != mqtt.MQTT_ERR_SUCCESS: + logger.error( + "Unable to publish to %s: mqtt error %s", + queued_publish.topic, + message_info.rc, + ) + self._buffer_undelivered(queued_publish, overwrite=False) + self._schedule_reconnect() + return + + # a successful rc only means paho accepted the message; above qos 0 + # it is not durable until the broker acks, so keep a copy for replay + if queued_publish.retain and not message_info.is_published(): + with self._retained_lock: + self._inflight_retained[message_info.mid] = ( + queued_publish.topic, + queued_publish.payload, + ) + + if queued_publish.done is not None: + self._wait_for_publish(message_info) + finally: + if queued_publish.done is not None: + queued_publish.done.set() + + def _handle_publish_event(self, mid: int) -> None: + """Drop the replay copy once the broker has acknowledged the message.""" + with self._retained_lock: + self._inflight_retained.pop(mid, None) + + def _wait_for_publish(self, message_info: mqtt.MQTTMessageInfo) -> None: + """Pump the loop until a shutdown-critical publish is acknowledged.""" + deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT + + while not message_info.is_published() and time.monotonic() < deadline: + if self.client is None: + return + + try: + result = self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL) + except (OSError, mqtt.WebsocketConnectionError) as err: + logger.warning("MQTT publish wait failed: %s", err) + self._schedule_reconnect() + return + + self._drain_callback_queue() + + if result != mqtt.MQTT_ERR_SUCCESS: + logger.error( + "MQTT loop returned error code while waiting for publish: %s", + result, + ) + self._schedule_reconnect() + return + + def _queue_retained( + self, + topic: str, + payload: Any, + retain: bool, + overwrite: bool = True, + ) -> None: + """Store the last retained value per topic for replay after reconnect.""" + with self._retained_lock: + if overwrite or topic not in self._pending_retained: + self._pending_retained[topic] = (payload, retain) + + def _handle_connect_event(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined] + """Begin a new session by subscribing before any replay is published.""" + if self.client is None: + return + + self.connected = True + self._subscription_ready = False + self._subscription_mid = None + logger.debug("MQTT connected") + + try: + result, mid = self.client.subscribe( + f"{self.mqtt_config.topic_prefix}/#", + qos=self.config.mqtt.qos, + ) + except (OSError, mqtt.WebsocketConnectionError) as err: + logger.warning("MQTT subscribe failed: %s", err) + self._schedule_reconnect() + return + + if result != mqtt.MQTT_ERR_SUCCESS: + logger.error( + "Unable to subscribe to MQTT command tree: mqtt error %s", result + ) + self._schedule_reconnect() + return + + self._subscription_mid = mid + + def _handle_connect_failure(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined] + """Record a failed connect attempt and transition into reconnect state.""" + self.connected = False + logger.error( + "Unable to connect to MQTT server: %s", self._reason_code_name(reason_code) + ) + self._schedule_reconnect() + + def _handle_disconnect_event(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined] + """Handle broker disconnects idempotently from the worker thread.""" + if not self.connected: + return + + self.connected = False + self._subscription_ready = False + self._subscription_mid = None + + if self._stop_event.is_set(): + logger.debug("MQTT disconnected") + self._cleanup_client() + return + + logger.error("MQTT disconnected: %s", self._reason_code_name(reason_code)) + self._schedule_reconnect() + + def _handle_subscribe_event( + self, + mid: int, + reason_codes: list[mqtt.ReasonCode], # type: ignore[name-defined] + ) -> None: + """Mark the session ready after SUBACK, then replay retained/runtime state.""" + if mid != self._subscription_mid: + return + + if any( + getattr(reason_code, "is_failure", False) for reason_code in reason_codes + ): + logger.error("MQTT subscription was rejected by the broker") + self._schedule_reconnect() + return + + self._subscription_ready = True + self._subscription_mid = None + + # a bug in replay should cost a snapshot, not the MQTT session + try: + self._publish_retained_state() + + if self._command_router is not None: + self._command_router.publish_runtime_snapshot(self.publish) + except Exception: + logger.exception("Error replaying MQTT state after subscribe") + + def _handle_inbound_message(self, topic: str, payload: str) -> None: + """Forward supported command topics into Dispatcher semantics.""" + if self._dispatcher is None: + return + + if not self._is_supported_command_topic(topic): + return + + if topic == "onConnect": + now = time.monotonic() + if now - self._last_on_connect_dispatch < MQTT_ON_CONNECT_RATE_LIMIT: + logger.debug("Skipping MQTT onConnect replay request due to rate limit") + return + self._last_on_connect_dispatch = now + + # a raise here used to end the network thread and take MQTT down + try: + self._dispatcher(topic, payload) + except Exception: + logger.exception("Error handling MQTT command topic %s", topic) + + def _is_supported_command_topic(self, topic: str) -> bool: + """Filter the wildcard subscription down to Dispatcher's command surface. + + Load-bearing rather than an optimization: the broker echoes Frigate's own + publishes back through frigate/#, and Dispatcher republishes topics it + does not recognize, so forwarding unfiltered would loop. + """ + if self._command_router is None: + return False + + # mirrors the gate on the state topic in _publish_retained_state() + if topic == "notifications/set" and not self._notifications_enabled_in_config(): + return False + + return self._command_router.is_command_topic(topic) + + def _strip_topic_prefix(self, topic: str) -> str: + return topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1) + + def _is_success_reason_code(self, reason_code: mqtt.ReasonCode) -> bool: # type: ignore[name-defined] + if hasattr(reason_code, "is_failure"): + return not bool(reason_code.is_failure) + + return bool(reason_code == 0) + + def _reason_code_name(self, reason_code: mqtt.ReasonCode) -> str: # type: ignore[name-defined] + if hasattr(reason_code, "getName"): + return str(reason_code.getName()) + + return str(reason_code) def _on_connect( self, @@ -205,29 +787,11 @@ class MqttClient(Communicator): reason_code: mqtt.ReasonCode, # type: ignore[name-defined] properties: Any, ) -> None: - """Mqtt connection callback.""" - threading.current_thread().name = "mqtt" - if reason_code != 0: - if reason_code == "Server unavailable": - logger.error( - "Unable to connect to MQTT server: MQTT Server unavailable" - ) - elif reason_code == "Bad user name or password": - logger.error( - "Unable to connect to MQTT server: MQTT Bad username or password" - ) - elif reason_code == "Not authorized": - logger.error("Unable to connect to MQTT server: MQTT Not authorized") - else: - logger.error( - "Unable to connect to MQTT server: Connection refused. Error code: %s", - reason_code.getName(), - ) - - self.connected = True - logger.debug("MQTT connected") - client.subscribe(f"{self.mqtt_config.topic_prefix}/#", qos=self.config.mqtt.qos) - self._set_initial_topics() + """Handle broker connect notifications from paho.""" + if self._is_success_reason_code(reason_code): + self._callback_queue.put(("connect", reason_code)) + else: + self._callback_queue.put(("connect_failure", reason_code)) def _on_disconnect( self, @@ -237,126 +801,63 @@ class MqttClient(Communicator): reason_code: mqtt.ReasonCode, # type: ignore[name-defined] properties: Any, ) -> None: - """Mqtt disconnection callback.""" - self.connected = False - logger.error("MQTT disconnected") + """Handle broker disconnect notifications from paho.""" + self._callback_queue.put(("disconnect", reason_code)) - def _start(self) -> None: - """Start mqtt client.""" - self.client = mqtt.Client( - callback_api_version=CallbackAPIVersion.VERSION2, - client_id=self.mqtt_config.client_id, - ) - self.client.on_connect = self._on_connect - self.client.on_disconnect = self._on_disconnect - self.client.will_set( - self.mqtt_config.topic_prefix + "/available", - payload="offline", - qos=1, - retain=True, - ) + def _on_subscribe( + self, + client: mqtt.Client, + userdata: Any, + mid: int, + reason_codes: list[mqtt.ReasonCode], # type: ignore[name-defined] + properties: Any, + ) -> None: + """Handle subscribe acknowledgements from paho.""" + self._callback_queue.put(("subscribed", mid, reason_codes)) - # register callbacks - callback_types = [ - "enabled", - "recordings", - "snapshots", - "detect", - "audio", - "audio_transcription", - "motion", - "improve_contrast", - "ptz_autotracker", - "motion_threshold", - "motion_contour_area", - "birdseye", - "birdseye_modes", - "review_alerts", - "review_detections", - "object_descriptions", - "review_descriptions", - "notifications", - ] + def _on_publish( + self, + client: mqtt.Client, + userdata: Any, + mid: int, + reason_code: mqtt.ReasonCode, # type: ignore[name-defined] + properties: Any, + ) -> None: + """Handle publish acknowledgements from paho. - for name in self.config.cameras.keys(): - for callback in callback_types: - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/{callback}/set", - self.on_mqtt_command, - ) + Only tracked retained messages need an event. At the default qos 0 + nothing is tracked, so this stays off the hot publish path. + """ + with self._retained_lock: + if mid not in self._inflight_retained: + return - # notifications suspend doesn't follow the /set topic pattern - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/notifications/suspend", - self.on_mqtt_command, - ) + self._callback_queue.put(("published", mid)) - if self.config.cameras[name].onvif.host: - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/ptz", - self.on_mqtt_command, - ) + def _on_message( + self, + client: mqtt.Client, + userdata: Any, + message: mqtt.MQTTMessage, + ) -> None: + """Queue inbound MQTT messages for processing in the worker loop.""" + topic = self._strip_topic_prefix(message.topic) - for mask_name in self.config.cameras[name].motion.mask.keys(): - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/motion_mask/{mask_name}/set", - self.on_mqtt_command, - ) - - for mask_name in self.config.cameras[name].objects.mask.keys(): - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/object_mask/{mask_name}/set", - self.on_mqtt_command, - ) - - for zone_name in self.config.cameras[name].zones.keys(): - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/{name}/zone/{zone_name}/set", - self.on_mqtt_command, - ) - - if self._notifications_enabled_in_config(): - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/notifications/set", - self.on_mqtt_command, - ) - - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/profile/set", - self.on_mqtt_command, - ) - - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/onConnect", self.on_mqtt_command - ) - - self.client.message_callback_add( - f"{self.mqtt_config.topic_prefix}/restart", self.on_mqtt_command - ) - - if self.mqtt_config.tls_ca_certs is not None: - if ( - self.mqtt_config.tls_client_cert is not None - and self.mqtt_config.tls_client_key is not None - ): - self.client.tls_set( - self.mqtt_config.tls_ca_certs, - self.mqtt_config.tls_client_cert, - self.mqtt_config.tls_client_key, - ) - else: - self.client.tls_set(self.mqtt_config.tls_ca_certs) - if self.mqtt_config.tls_insecure is not None: - self.client.tls_insecure_set(self.mqtt_config.tls_insecure) - if self.mqtt_config.user is not None: - self.client.username_pw_set( - self.mqtt_config.user, password=self.mqtt_config.password - ) - try: - # https://stackoverflow.com/a/55390477 - # with connect_async, retries are handled automatically - self.client.connect_async(self.mqtt_config.host, self.mqtt_config.port, 60) - self.client.loop_start() - except Exception as e: - logger.error(f"Unable to connect to MQTT server: {e}") + # Ignore everything outside Frigate's command surface before decoding or + # dispatching into the rest of the app. + if not self._is_supported_command_topic(topic): return + + try: + payload = message.payload.decode() + except UnicodeDecodeError: + logger.debug("Ignoring non-UTF-8 MQTT payload for topic %s", topic) + return + + self._callback_queue.put( + ( + "message", + topic, + payload, + ) + ) diff --git a/frigate/comms/webpush.py b/frigate/comms/webpush.py index a4aab58df8..bb41642fe4 100644 --- a/frigate/comms/webpush.py +++ b/frigate/comms/webpush.py @@ -63,14 +63,8 @@ class WebPushClient(Communicator): self.last_notification_time: float = 0 self.user_cameras: dict[str, set[str]] = {} self.notification_queue: queue.Queue[PushNotification] = queue.Queue() - self.notification_thread = threading.Thread( - target=self._process_notifications, daemon=True - ) - self.notification_thread.start() - self.suspension_thread = threading.Thread( - target=self._process_suspensions, daemon=True - ) - self.suspension_thread.start() + self.notification_thread: threading.Thread | None = None + self.suspension_thread: threading.Thread | None = None if not self.config.notifications.email: logger.warning("Email must be provided for push notifications to be sent.") @@ -99,6 +93,16 @@ class WebPushClient(Communicator): """Wrapper for allowing dispatcher to subscribe.""" pass + def start(self) -> None: + self.notification_thread = threading.Thread( + target=self._process_notifications, daemon=True + ) + self.notification_thread.start() + self.suspension_thread = threading.Thread( + target=self._process_suspensions, daemon=True + ) + self.suspension_thread.start() + def check_registrations(self) -> None: # check for valid claim or create new one now = datetime.datetime.now().timestamp() @@ -608,4 +612,5 @@ class WebPushClient(Communicator): def stop(self) -> None: logger.info("Closing notification queue") - self.notification_thread.join() + if self.notification_thread is not None: + self.notification_thread.join() diff --git a/frigate/comms/ws.py b/frigate/comms/ws.py index ccb5d42890..6e4eb212a5 100644 --- a/frigate/comms/ws.py +++ b/frigate/comms/ws.py @@ -466,7 +466,6 @@ class WebSocketClient(Communicator): def subscribe(self, receiver: Callable) -> None: self._dispatcher = receiver - self.start() def start(self) -> None: """Start the websocket client.""" diff --git a/frigate/test/test_mqtt_lifecycle.py b/frigate/test/test_mqtt_lifecycle.py new file mode 100644 index 0000000000..7352cd1578 --- /dev/null +++ b/frigate/test/test_mqtt_lifecycle.py @@ -0,0 +1,907 @@ +import os +import threading +import unittest +from typing import Any +from unittest.mock import MagicMock, patch + +import paho.mqtt.client as mqtt + +from frigate.comms.dispatcher import Dispatcher +from frigate.comms.mqtt import MqttClient, QueuedPublish +from frigate.config import FrigateConfig +from frigate.const import MODEL_CACHE_DIR + + +class RuntimeSnapshotReceiver: + def __init__(self) -> None: + self.messages: list[tuple[str, str]] = [] + + def _receive(self, topic: str, payload: str) -> None: + self.messages.append((topic, payload)) + + +class FakeMessage: + def __init__(self, topic: str, payload: bytes) -> None: + self.topic = topic + self.payload = payload + + +class FakeCommunicator: + def __init__(self) -> None: + self.receiver = None + self.dispatcher = None + self.started = False + + def subscribe(self, receiver) -> None: + self.receiver = receiver + + def attach_dispatcher(self, dispatcher) -> None: + self.dispatcher = dispatcher + + def start(self) -> None: + self.started = True + + def publish(self, topic: str, payload: Any, retain: bool = False) -> None: + return None + + def stop(self) -> None: + return None + + +class FakeSnapshotCommunicator(FakeCommunicator): + def start(self) -> None: + self.started = True + assert self.dispatcher is not None + self.dispatcher.publish_runtime_snapshot(lambda *_args, **_kwargs: None) + + +def build_config() -> FrigateConfig: + config = { + "mqtt": { + "host": "mqtt", + "client_id": "frigate-test", + "topic_prefix": "frigate", + }, + "notifications": {"enabled": True}, + "cameras": { + "front": { + "ffmpeg": { + "inputs": [ + { + "path": "rtsp://10.0.0.1:554/video", + "roles": ["detect", "audio"], + } + ] + }, + "detect": { + "height": 1080, + "width": 1920, + "fps": 5, + }, + "audio": {"enabled": True}, + "notifications": {"enabled": False}, + "onvif": {"host": "10.0.0.5"}, + "motion": { + "mask": { + "motion_mask_1": { + "coordinates": "0,0,1,0,1,1,0,1", + } + } + }, + "objects": { + "track": ["person"], + "mask": { + "object_mask_1": { + "coordinates": "0,0,1,0,1,1,0,1", + } + }, + }, + "zones": { + "driveway": { + "coordinates": "0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9", + "objects": ["person"], + } + }, + } + }, + } + return FrigateConfig(**config) + + +def build_dispatcher(config: FrigateConfig, communicators: list[Any]) -> Dispatcher: + """Build a real Dispatcher with only the activity managers stubbed out.""" + with ( + patch("frigate.comms.dispatcher.CameraActivityManager") as mock_camera_activity, + patch("frigate.comms.dispatcher.AudioActivityManager") as mock_audio_activity, + ): + mock_camera_activity.return_value.last_camera_activity = {} + mock_audio_activity.return_value.current_audio_detections = {} + + return Dispatcher(config, MagicMock(), MagicMock(), {}, communicators) + + +class TestMqttClientLifecycle(unittest.TestCase): + def setUp(self) -> None: + if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR): + os.makedirs(MODEL_CACHE_DIR) + + self.config = build_config() + self.client = MqttClient(self.config) + self.receiver = RuntimeSnapshotReceiver() + self.client.attach_dispatcher(build_dispatcher(self.config, [])) + + def test_subscribe_stores_receiver_without_starting_worker(self) -> None: + client = MqttClient(self.config) + + with patch.object(client, "_start_worker") as mock_start_worker: + client.subscribe(self.receiver._receive) + + self.assertIsNotNone(client._dispatcher) + self.assertIs(client._dispatcher.__self__, self.receiver) + self.assertIs(client._dispatcher.__func__, RuntimeSnapshotReceiver._receive) + mock_start_worker.assert_not_called() + + def test_attach_dispatcher_supplies_command_surface(self) -> None: + client = MqttClient(self.config) + + self.assertFalse(client._is_supported_command_topic("front/detect/set")) + + dispatcher = build_dispatcher(self.config, []) + client.attach_dispatcher(dispatcher) + + self.assertIs(client._command_router, dispatcher) + self.assertTrue(client._is_supported_command_topic("front/detect/set")) + + def test_start_starts_worker_after_receiver_registration(self) -> None: + self.client.subscribe(self.receiver._receive) + + with patch.object(self.client, "_start_worker") as mock_start_worker: + self.client.start() + + mock_start_worker.assert_called_once() + + def test_dispatcher_initializes_state_before_starting_communicators(self) -> None: + fake_comm = FakeSnapshotCommunicator() + dispatcher = build_dispatcher(self.config, [fake_comm]) + + self.assertIsNone(dispatcher.web_push_client) + self.assertIs(fake_comm.receiver.__self__, dispatcher) + self.assertIs(fake_comm.dispatcher, dispatcher) + self.assertFalse(fake_comm.started) + + dispatcher.start_communicators() + + self.assertTrue(fake_comm.started) + + def test_publish_drops_ephemeral_and_coalesces_retained_while_disconnected( + self, + ) -> None: + self.client.publish("front/events", "payload") + self.assertTrue(self.client._publish_queue.empty()) + + self.client.publish("profile/state", "armed", retain=True) + self.client.publish("profile/state", "disarmed", retain=True) + + self.assertEqual( + self.client._pending_retained["frigate/profile/state"], + ("disarmed", True), + ) + + def test_outage_buffer_replays_before_newer_queued_value(self) -> None: + """A topic that changed since the reconnect must not be reverted by the + replay of the value buffered during the outage.""" + mock_client = MagicMock() + mock_client.publish.return_value = MagicMock( + rc=mqtt.MQTT_ERR_SUCCESS, mid=1, **{"is_published.return_value": True} + ) + self.client._pending_retained = {"frigate/front/detect/state": ("OFF", True)} + self.client.client = mock_client + self.client.connected = True + self.client._subscription_ready = True + self.client._publish_queue.put( + QueuedPublish("frigate/front/detect/state", "ON", True) + ) + + self.client._drain_publish_queue() + + published = [ + (call.args[0], call.args[1]) for call in mock_client.publish.call_args_list + ] + self.assertEqual( + published, + [ + ("frigate/front/detect/state", "OFF"), + ("frigate/front/detect/state", "ON"), + ], + ) + + def test_flush_failure_does_not_clobber_newer_queued_value(self) -> None: + """If the session dies mid-replay, the newer queued value that gets + rebuffered has to win over the older entries still being replayed.""" + self.client._pending_retained = { + "frigate/a/detect/state": ("OLD", True), + "frigate/front/detect/state": ("OFF", True), + } + self.client.connected = True + self.client._subscription_ready = True + self.client._publish_queue.put( + QueuedPublish("frigate/front/detect/state", "ON", True) + ) + + mock_client = MagicMock() + mock_client.publish.side_effect = BrokenPipeError("broken pipe") + self.client.client = mock_client + + self.client._drain_publish_queue() + + self.assertEqual( + self.client._pending_retained["frigate/front/detect/state"], ("ON", True) + ) + + def test_publish_buffers_messages_until_subscription_ready(self) -> None: + self.client.connected = True + + self.client.publish("front/events", "payload") + self.client.publish("profile/state", "armed", retain=True) + + self.assertEqual(self.client._publish_queue.qsize(), 2) + self.assertEqual(self.client._pending_retained, {}) + + @patch("frigate.comms.mqtt.mqtt.Client") + def test_connect_client_initializes_manual_loop_client( + self, mock_client_cls + ) -> None: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + + connected = self.client._connect_client() + + self.assertTrue(connected) + mock_client_cls.assert_called_once_with( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id="frigate-test", + reconnect_on_failure=False, + ) + self.assertIs(mock_client.on_connect.__self__, self.client) + self.assertIs(mock_client.on_connect.__func__, MqttClient._on_connect) + self.assertIs(mock_client.on_disconnect.__self__, self.client) + self.assertIs(mock_client.on_disconnect.__func__, MqttClient._on_disconnect) + self.assertIs(mock_client.on_message.__self__, self.client) + self.assertIs(mock_client.on_message.__func__, MqttClient._on_message) + self.assertIs(mock_client.on_subscribe.__self__, self.client) + self.assertIs(mock_client.on_subscribe.__func__, MqttClient._on_subscribe) + mock_client.connect.assert_called_once_with("mqtt", 1883, 60) + + def test_handle_connect_failure_leaves_client_disconnected(self) -> None: + self.client.client = MagicMock() + reason_code = MagicMock() + reason_code.getName.return_value = "Not authorized" + + with patch.object( + self.client, "_schedule_reconnect" + ) as mock_schedule_reconnect: + self.client._handle_connect_failure(reason_code) + + self.assertFalse(self.client.connected) + mock_schedule_reconnect.assert_called_once() + + def test_handle_connect_event_subscribes_wildcard_topic(self) -> None: + self.client.client = MagicMock() + self.client.client.subscribe.return_value = (mqtt.MQTT_ERR_SUCCESS, 42) + + self.client._handle_connect_event(MagicMock()) + + self.assertTrue(self.client.connected) + self.assertEqual(self.client._subscription_mid, 42) + self.client.client.subscribe.assert_called_once_with("frigate/#", qos=0) + + def test_handle_connect_event_reconnects_on_recoverable_subscribe_error( + self, + ) -> None: + self.client.client = MagicMock() + self.client.client.subscribe.side_effect = BrokenPipeError("broken pipe") + + with patch.object( + self.client, "_schedule_reconnect" + ) as mock_schedule_reconnect: + self.client._handle_connect_event(MagicMock()) + + mock_schedule_reconnect.assert_called_once() + + def test_handle_subscribe_event_publishes_snapshots_after_matching_suback( + self, + ) -> None: + self.client.connected = True + self.client._subscription_mid = 7 + + with ( + patch.object(self.client, "_publish_retained_state") as mock_retained, + patch.object( + self.client._command_router, "publish_runtime_snapshot" + ) as mock_snapshot, + ): + self.client._handle_subscribe_event(7, [MagicMock(is_failure=False)]) + + self.assertTrue(self.client._subscription_ready) + mock_retained.assert_called_once() + mock_snapshot.assert_called_once_with(self.client.publish) + + def test_handle_subscribe_event_ignores_other_suback_mid(self) -> None: + self.client.connected = True + self.client._subscription_mid = 8 + + with ( + patch.object(self.client, "_publish_retained_state") as mock_retained, + patch.object( + self.client._command_router, "publish_runtime_snapshot" + ) as mock_snapshot, + ): + self.client._handle_subscribe_event(9, [MagicMock(is_failure=False)]) + + self.assertFalse(self.client._subscription_ready) + mock_retained.assert_not_called() + mock_snapshot.assert_not_called() + + def test_on_message_strips_prefix_and_dispatches_supported_topic(self) -> None: + dispatcher = MagicMock() + self.client._dispatcher = dispatcher + + self.client._on_message( + MagicMock(), + None, + FakeMessage("frigate/front/detect/set", b"ON"), + ) + self.client._drain_callback_queue() + + dispatcher.assert_called_once_with("front/detect/set", "ON") + + def test_on_message_ignores_unsupported_state_topic(self) -> None: + dispatcher = MagicMock() + self.client._dispatcher = dispatcher + + self.client._on_message( + MagicMock(), + None, + FakeMessage("frigate/front/detect/state", b"ON"), + ) + self.client._drain_callback_queue() + + dispatcher.assert_not_called() + + def test_on_message_ignores_non_utf8_payloads(self) -> None: + self.client._on_message( + MagicMock(), + None, + FakeMessage("frigate/onConnect", b"\xff"), + ) + + self.assertTrue(self.client._callback_queue.empty()) + + def test_on_connect_requests_are_rate_limited(self) -> None: + dispatcher = MagicMock() + self.client._dispatcher = dispatcher + + with patch("frigate.comms.mqtt.time.monotonic", side_effect=[100.0, 100.1]): + self.client._handle_inbound_message("onConnect", "") + self.client._handle_inbound_message("onConnect", "") + + dispatcher.assert_called_once_with("onConnect", "") + + def test_supported_command_topics_preserve_command_surface(self) -> None: + for topic in ( + "front/detect/set", + "front/audio_transcription/set", + "front/notifications/set", + "front/notifications/suspend", + "front/zone/driveway/set", + "front/motion_mask/motion_mask_1/set", + "front/ptz", + "notifications/set", + "profile/set", + "onConnect", + "restart", + ): + with self.subTest(topic=topic): + self.assertTrue(self.client._is_supported_command_topic(topic)) + + for topic in ( + # Frigate's own publishes echoing back through the wildcard + "front/detect/state", + "available", + "front/notifications/suspended", + "front/zone/set", + "front/nonsense/set", + "nonsense/set", + ): + with self.subTest(topic=topic): + self.assertFalse(self.client._is_supported_command_topic(topic)) + + def test_command_surface_tracks_dispatcher_handlers(self) -> None: + """The allowlist is derived, so a new handler is routable for free.""" + router = self.client._command_router + router._camera_settings_handlers["brand_new_toggle"] = MagicMock() + + self.assertTrue( + self.client._is_supported_command_topic("front/brand_new_toggle/set") + ) + + def test_global_notifications_set_follows_config_gate(self) -> None: + """The command gate must match the publish gate in + _publish_retained_state: with notifications unconfigured there is no + state topic, so the command must not flip runtime state either.""" + self.assertTrue(self.client._notifications_enabled_in_config()) + self.assertTrue(self.client._is_supported_command_topic("notifications/set")) + + with patch.object( + self.client, "_notifications_enabled_in_config", return_value=False + ): + self.assertFalse( + self.client._is_supported_command_topic("notifications/set") + ) + # per-camera topics stay routable regardless of the global gate + self.assertTrue( + self.client._is_supported_command_topic("front/notifications/set") + ) + self.assertTrue( + self.client._is_supported_command_topic("front/notifications/suspend") + ) + + def test_publish_direct_waits_for_flush_barrier(self) -> None: + mock_client = MagicMock() + mock_client.loop.return_value = mqtt.MQTT_ERR_SUCCESS + self.client.client = mock_client + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=1) + # inflight tracking checks once, then _wait_for_publish polls + message_info.is_published.side_effect = [False, False, True] + mock_client.publish.return_value = message_info + barrier = MagicMock() + + self.client._publish_direct( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + mock_client.loop.assert_called_once() + barrier.set.assert_called_once() + + def test_shutdown_barrier_releases_when_publish_raises(self) -> None: + """stop() waits on this barrier, so a broker error must not stall + shutdown for the full flush timeout.""" + self.client.client = MagicMock() + self.client.client.publish.side_effect = BrokenPipeError("broken pipe") + barrier = threading.Event() + + with patch.object(self.client, "_schedule_reconnect"): + self.client._publish_direct( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + self.assertTrue(barrier.is_set()) + self.assertEqual( + self.client._pending_retained["frigate/available"], ("stopped", True) + ) + + def test_shutdown_barrier_releases_on_publish_error_code(self) -> None: + self.client.client = MagicMock() + self.client.client.publish.return_value = MagicMock(rc=mqtt.MQTT_ERR_NO_CONN) + barrier = threading.Event() + + with patch.object(self.client, "_schedule_reconnect"): + self.client._publish_direct( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + self.assertTrue(barrier.is_set()) + + def test_shutdown_barrier_releases_when_requeued_while_disconnected(self) -> None: + barrier = threading.Event() + self.client._publish_queue.put( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + self.client._requeue_disconnected_publishes() + + self.assertTrue(barrier.is_set()) + self.assertEqual( + self.client._pending_retained["frigate/available"], ("stopped", True) + ) + + def test_shutdown_barrier_releases_when_worker_crashes(self) -> None: + """stop() can queue the final publish just as the worker dies, and + nothing drains the queue after that.""" + barrier = threading.Event() + self.client._publish_queue.put( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + with patch.object( + self.client, + "_mqtt_loop_worker", + side_effect=RuntimeError("unexpected bug"), + ): + self.client._worker_main() + + self.assertTrue(barrier.is_set()) + + def test_shutdown_barrier_releases_when_publish_raises_unexpectedly(self) -> None: + """The message is off the queue by the time this runs, so crash cleanup + cannot recover it and only _publish_direct can release the waiter.""" + self.client.client = MagicMock() + self.client.client.publish.side_effect = RuntimeError("unexpected bug") + barrier = threading.Event() + + with self.assertRaises(RuntimeError): + self.client._publish_direct( + QueuedPublish("frigate/available", "stopped", True, barrier) + ) + + self.assertTrue(barrier.is_set()) + + def test_newest_inflight_retained_value_wins(self) -> None: + """Several updates to one topic can be unacked at once above qos 0, and + the newest is the one subscribers should end up with.""" + mock_client = MagicMock() + self.client.client = mock_client + self.client.connected = True + + for mid, payload in ((1, "ON"), (2, "OFF")): + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid) + message_info.is_published.return_value = False + mock_client.publish.return_value = message_info + self.client._publish_direct( + QueuedPublish("frigate/front/detect/state", payload, True) + ) + + self.client._requeue_inflight_retained() + + self.assertEqual( + self.client._pending_retained["frigate/front/detect/state"], ("OFF", True) + ) + + def test_inflight_retained_does_not_clobber_queued_value(self) -> None: + """Anything still queued was written later than anything in flight.""" + self.client._pending_retained = {"frigate/front/detect/state": ("OFF", True)} + self.client._inflight_retained = {1: ("frigate/front/detect/state", "ON")} + + self.client._requeue_inflight_retained() + + self.assertEqual( + self.client._pending_retained["frigate/front/detect/state"], ("OFF", True) + ) + + def test_unacked_retained_publish_survives_reconnect(self) -> None: + """Above qos 0 a successful rc only means paho queued the message, and + dropping the client drops its outbound queue with it.""" + mock_client = MagicMock() + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12) + message_info.is_published.return_value = False + mock_client.publish.return_value = message_info + self.client.client = mock_client + self.client.connected = True + + self.client._publish_direct( + QueuedPublish("frigate/profile/state", "armed", True) + ) + self.assertEqual( + self.client._inflight_retained[12], ("frigate/profile/state", "armed") + ) + + self.client._cleanup_client() + + self.assertEqual(self.client._inflight_retained, {}) + self.assertEqual( + self.client._pending_retained["frigate/profile/state"], ("armed", True) + ) + + def test_acked_retained_publish_is_not_replayed(self) -> None: + mock_client = MagicMock() + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12) + message_info.is_published.return_value = False + mock_client.publish.return_value = message_info + self.client.client = mock_client + self.client.connected = True + + self.client._publish_direct( + QueuedPublish("frigate/profile/state", "armed", True) + ) + self.client._on_publish(mock_client, None, 12, MagicMock(), None) + self.client._drain_callback_queue() + + self.assertEqual(self.client._inflight_retained, {}) + + self.client._cleanup_client() + + self.assertEqual(self.client._pending_retained, {}) + + def test_already_published_retained_is_not_tracked(self) -> None: + """At the default qos 0 paho reports the message as published inline, + so there is nothing to wait on.""" + mock_client = MagicMock() + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12) + message_info.is_published.return_value = True + mock_client.publish.return_value = message_info + self.client.client = mock_client + self.client.connected = True + + self.client._publish_direct( + QueuedPublish("frigate/profile/state", "armed", True) + ) + + self.assertEqual(self.client._inflight_retained, {}) + + def test_wait_for_publish_survives_disconnect_during_wait(self) -> None: + mock_client = MagicMock() + self.client.client = mock_client + self.client.connected = True + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS) + message_info.is_published.side_effect = [False, False] + + loop_calls = [0] + + def loop_side_effect(timeout: float) -> int: + if loop_calls[0] == 0: + self.client._callback_queue.put(("disconnect", 1)) + loop_calls[0] += 1 + return mqtt.MQTT_ERR_SUCCESS + + mock_client.loop.side_effect = loop_side_effect + + self.client._wait_for_publish(message_info) + + self.assertEqual(loop_calls[0], 1) + self.assertIsNone(self.client.client) + + def test_publish_direct_reconnects_on_recoverable_publish_error(self) -> None: + self.client.client = MagicMock() + self.client.client.publish.side_effect = BrokenPipeError("broken pipe") + + with patch.object( + self.client, "_schedule_reconnect" + ) as mock_schedule_reconnect: + self.client._publish_direct( + QueuedPublish("frigate/profile/state", "armed", True) + ) + + self.assertEqual( + self.client._pending_retained["frigate/profile/state"], + ("armed", True), + ) + mock_schedule_reconnect.assert_called_once() + + def test_mqtt_loop_worker_reconnects_on_recoverable_loop_error(self) -> None: + self.client.client = MagicMock() + self.client.client.loop.side_effect = OSError("socket closed") + + def stop_after_reconnect() -> None: + self.client._stop_event.set() + + with patch.object( + self.client, + "_schedule_reconnect", + side_effect=stop_after_reconnect, + ) as mock_schedule_reconnect: + self.client._mqtt_loop_worker() + + mock_schedule_reconnect.assert_called_once() + + def test_worker_main_goes_dormant_on_unexpected_exception(self) -> None: + with patch.object( + self.client, + "_mqtt_loop_worker", + side_effect=RuntimeError("unexpected bug"), + ): + self.client._worker_main() + + self.assertTrue(self.client._stop_event.is_set()) + self.assertFalse(self.client.connected) + self.assertFalse(self.client._subscription_ready) + self.assertIsNone(self.client.client) + + def test_worker_crash_announces_offline_before_disconnecting(self) -> None: + """The clean disconnect in cleanup suppresses the will, so a dormant + MQTT session must say so itself or consumers keep the stale values.""" + mock_client = MagicMock() + mock_client.publish.return_value = MagicMock( + rc=mqtt.MQTT_ERR_SUCCESS, **{"is_published.return_value": True} + ) + self.client.client = mock_client + self.client.connected = True + + with patch.object( + self.client, + "_mqtt_loop_worker", + side_effect=RuntimeError("unexpected bug"), + ): + self.client._worker_main() + + mock_client.publish.assert_called_once_with( + "frigate/available", + "offline", + qos=self.config.mqtt.qos, + retain=True, + ) + # the announcement has to land before the socket is torn down + self.assertLess( + mock_client.method_calls.index( + next(c for c in mock_client.method_calls if c[0] == "publish") + ), + mock_client.method_calls.index( + next(c for c in mock_client.method_calls if c[0] == "disconnect") + ), + ) + + def test_worker_crash_stays_dormant_when_offline_publish_fails(self) -> None: + self.client.client = MagicMock() + self.client.client.publish.side_effect = BrokenPipeError("broken pipe") + + with patch.object( + self.client, + "_mqtt_loop_worker", + side_effect=RuntimeError("unexpected bug"), + ): + self.client._worker_main() + + self.assertTrue(self.client._stop_event.is_set()) + self.assertIsNone(self.client.client) + + def test_command_handler_exception_does_not_kill_worker(self) -> None: + """A raise in a dispatcher handler used to end the network thread and + take MQTT down until the next Frigate restart.""" + self.client._dispatcher = MagicMock(side_effect=RuntimeError("handler bug")) + self.client._callback_queue.put(("message", "front/detect/set", "ON")) + self.client._callback_queue.put(("message", "front/motion/set", "ON")) + + with self.assertLogs("frigate.comms.mqtt", level="ERROR"): + self.client._drain_callback_queue() + + self.assertEqual(self.client._dispatcher.call_count, 2) + + def test_snapshot_replay_exception_does_not_kill_worker(self) -> None: + self.client.connected = True + self.client._subscription_mid = 3 + + with ( + patch.object( + self.client, + "_publish_retained_state", + side_effect=RuntimeError("replay bug"), + ), + self.assertLogs("frigate.comms.mqtt", level="ERROR"), + ): + self.client._handle_subscribe_event(3, [MagicMock(is_failure=False)]) + + self.assertTrue(self.client._subscription_ready) + + def test_schedule_reconnect_drops_stale_ephemeral_and_preserves_retained( + self, + ) -> None: + mock_client = MagicMock() + message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS) + mock_client.publish.return_value = message_info + self.client.client = mock_client + self.client.connected = True + self.client._subscription_ready = True + + self.client.publish("front/events", "ephemeral") + self.client.publish("profile/state", "armed", retain=True) + + self.client._schedule_reconnect() + + self.assertTrue(self.client._publish_queue.empty()) + self.assertEqual( + self.client._pending_retained["frigate/profile/state"], + ("armed", True), + ) + + self.client.client = mock_client + self.client.connected = True + self.client._subscription_ready = True + self.client._drain_publish_queue() + + mock_client.publish.assert_called_once_with( + "frigate/profile/state", + "armed", + qos=self.config.mqtt.qos, + retain=True, + ) + + def test_stop_disconnects_client_and_joins_worker(self) -> None: + worker = MagicMock() + worker.is_alive.return_value = True + self.client._worker = worker + mock_client = MagicMock() + self.client.client = mock_client + + self.client.stop() + + self.assertTrue(self.client._stop_event.is_set()) + self.assertGreaterEqual(mock_client.disconnect.call_count, 1) + worker.join.assert_called_once() + + def test_stop_skips_stopped_publish_until_subscription_ready(self) -> None: + worker = MagicMock() + worker.is_alive.return_value = True + self.client._worker = worker + self.client.client = MagicMock() + self.client.connected = True + self.client._subscription_ready = False + + with patch.object(self.client._publish_queue, "put") as mock_queue_put: + self.client.stop() + + mock_queue_put.assert_not_called() + + def test_duplicate_disconnect_callback_is_safe(self) -> None: + self.client.connected = True + + with patch.object( + self.client, "_schedule_reconnect" + ) as mock_schedule_reconnect: + self.client._handle_disconnect_event(MagicMock()) + self.client._handle_disconnect_event(MagicMock()) + + mock_schedule_reconnect.assert_called_once() + + def test_publish_retained_state_emits_expected_topic_families(self) -> None: + published_topics: list[tuple[str, Any, bool]] = [] + + with patch.object( + self.client, + "publish", + side_effect=lambda topic, payload, retain=False: published_topics.append( + (topic, payload, retain) + ), + ): + self.client._publish_retained_state() + + topics = {topic for topic, _, _ in published_topics} + self.assertIn("front/enabled/state", topics) + self.assertIn("front/motion_mask/motion_mask_1/state", topics) + self.assertIn("front/object_mask/object_mask_1/state", topics) + self.assertIn("front/zone/driveway/state", topics) + self.assertIn("notifications/state", topics) + self.assertIn("profile/state", topics) + self.assertIn("available", topics) + + def test_publish_retained_state_uses_runtime_values(self) -> None: + camera = self.config.cameras["front"] + camera.enabled_in_config = True + camera.enabled = False + camera.record.enabled_in_config = True + camera.record.enabled = False + camera.audio.enabled_in_config = True + camera.audio.enabled = False + camera.motion.enabled = False + camera.onvif.autotracking.enabled_in_config = True + camera.onvif.autotracking.enabled = False + camera.review.alerts.enabled_in_config = True + camera.review.alerts.enabled = False + camera.review.detections.enabled_in_config = True + camera.review.detections.enabled = False + camera.objects.genai.enabled_in_config = True + camera.objects.genai.enabled = False + camera.review.genai.enabled_in_config = True + camera.review.genai.enabled = False + self.config.notifications.enabled_in_config = True + self.config.notifications.enabled = False + + published_states: dict[str, Any] = {} + + with patch.object( + self.client, + "publish", + side_effect=lambda topic, payload, retain=False: ( + published_states.__setitem__(topic, payload) + ), + ): + self.client._publish_retained_state() + + self.assertEqual(published_states["front/enabled/state"], "OFF") + self.assertEqual(published_states["front/recordings/state"], "OFF") + self.assertEqual(published_states["front/audio/state"], "OFF") + self.assertEqual(published_states["front/motion/state"], "OFF") + self.assertEqual(published_states["front/ptz_autotracker/state"], "OFF") + self.assertEqual(published_states["front/review_alerts/state"], "OFF") + self.assertEqual(published_states["front/review_detections/state"], "OFF") + self.assertEqual(published_states["front/object_descriptions/state"], "OFF") + self.assertEqual(published_states["front/review_descriptions/state"], "OFF") + self.assertEqual(published_states["notifications/state"], "OFF") diff --git a/frigate/test/test_mqtt_topic_registration.py b/frigate/test/test_mqtt_topic_registration.py deleted file mode 100644 index 309fafb983..0000000000 --- a/frigate/test/test_mqtt_topic_registration.py +++ /dev/null @@ -1,100 +0,0 @@ -"""Tests for MQTT command topic callback registration.""" - -import unittest -from unittest.mock import MagicMock, patch - -from frigate.comms.mqtt import MqttClient - - -def _make_camera_mock( - *, - enabled: bool = True, - notifications_enabled_in_config: bool = False, -) -> MagicMock: - """Build a camera config mock with the fields _start() reads.""" - camera = MagicMock() - camera.enabled = enabled - camera.notifications.enabled_in_config = notifications_enabled_in_config - camera.onvif.host = None - camera.motion.mask = {} - camera.objects.mask = {} - camera.zones = {} - return camera - - -def _registered_topics( - cameras: dict[str, MagicMock], - *, - global_notifications_enabled_in_config: bool = False, -) -> set[str]: - """Start an MqttClient against a mocked paho client and collect the - topics registered via message_callback_add.""" - config = MagicMock() - config.cameras = cameras - config.notifications.enabled_in_config = global_notifications_enabled_in_config - config.mqtt.topic_prefix = "frigate" - config.mqtt.client_id = "frigate" - config.mqtt.user = None - config.mqtt.tls_ca_certs = None - config.mqtt.tls_insecure = None - - with patch("frigate.comms.mqtt.mqtt.Client") as client_cls: - mqtt_client = MqttClient(config) - mqtt_client.subscribe(MagicMock()) - - paho_client = client_cls.return_value - return {call.args[0] for call in paho_client.message_callback_add.call_args_list} - - -class TestMqttTopicRegistration(unittest.TestCase): - def test_camera_notification_topics_registered(self): - """Per-camera notification set/suspend must be registered so paho - routes them to the dispatcher (unregistered topics drop silently).""" - topics = _registered_topics( - {"front_door": _make_camera_mock(notifications_enabled_in_config=True)} - ) - - self.assertIn("frigate/front_door/notifications/set", topics) - self.assertIn("frigate/front_door/notifications/suspend", topics) - - def test_global_set_registered_with_camera_only_notifications(self): - """The global topic must work when notifications are enabled only at - the camera level, matching the WebPushClient gating in app.py.""" - topics = _registered_topics( - {"front_door": _make_camera_mock(notifications_enabled_in_config=True)}, - global_notifications_enabled_in_config=False, - ) - - self.assertIn("frigate/notifications/set", topics) - - def test_global_set_registered_with_global_notifications(self): - topics = _registered_topics( - {"front_door": _make_camera_mock()}, - global_notifications_enabled_in_config=True, - ) - - self.assertIn("frigate/notifications/set", topics) - - def test_global_set_not_registered_when_notifications_unconfigured(self): - topics = _registered_topics( - {"front_door": _make_camera_mock()}, - global_notifications_enabled_in_config=False, - ) - - self.assertNotIn("frigate/notifications/set", topics) - - def test_disabled_camera_does_not_enable_global_set(self): - topics = _registered_topics( - { - "front_door": _make_camera_mock( - enabled=False, notifications_enabled_in_config=True - ) - }, - global_notifications_enabled_in_config=False, - ) - - self.assertNotIn("frigate/notifications/set", topics) - - -if __name__ == "__main__": - unittest.main() diff --git a/frigate/test/test_profiles.py b/frigate/test/test_profiles.py index fd949f76b3..09229f5642 100644 --- a/frigate/test/test_profiles.py +++ b/frigate/test/test_profiles.py @@ -738,6 +738,53 @@ class TestProfileManager(unittest.TestCase): assert "objects" in api_base assert api_base["objects"]["track"] == ["person"] + @patch.object(ProfileManager, "_persist_active_profile") + def test_update_config_reapplies_active_profile(self, mock_persist): + """Replacing the config preserves the active profile overrides.""" + self.manager.activate_profile("armed") + + new_config = FrigateConfig(**self.config_data) + self.manager.update_config(new_config) + + assert self.manager.config.active_profile == "armed" + assert self.manager.config.cameras["front"].notifications.enabled is True + assert self.manager.config.cameras["front"].objects.track == [ + "person", + "car", + "package", + ] + + @patch.object(ProfileManager, "_persist_active_profile") + def test_update_config_deactivates_removed_profile(self, mock_persist): + """Replacing the config clears the active profile if it no longer exists.""" + self.manager.activate_profile("armed") + + new_config_data = json.loads(json.dumps(self.config_data)) + del new_config_data["profiles"]["armed"] + del new_config_data["cameras"]["front"]["profiles"]["armed"] + del new_config_data["cameras"]["back"]["profiles"]["armed"] + + new_config = FrigateConfig(**new_config_data) + self.manager.update_config(new_config) + + assert self.manager.config.active_profile is None + + @patch.object(ProfileManager, "_persist_active_profile") + def test_enabled_state_is_published_via_dispatcher_when_profile_changes( + self, mock_persist + ): + """Enabled state updates are published through the dispatcher when present.""" + self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") + self.config.cameras["front"].profiles["away"] = CameraProfileConfig( + enabled=False + ) + dispatcher = MagicMock() + self.manager = ProfileManager(self.config, self.mock_updater, dispatcher) + + self.manager.activate_profile("away") + + dispatcher.publish.assert_any_call("front/enabled/state", "OFF", retain=True) + def test_base_configs_for_api_are_json_serializable(self): """API base configs are JSON-serializable (mode='json').""" import json