Coverage for src/updates2mqtt/mqtt.py: 76%
335 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-09 06:17 +0000
1import asyncio
2import json
3import re
4import time
5from collections.abc import Callable
6from dataclasses import dataclass, field
7from threading import Event
8from typing import Any, Literal, cast
10import paho.mqtt.client as mqtt
11import paho.mqtt.subscribeoptions
12import structlog
13from paho.mqtt.client import MQTT_CLEAN_START_FIRST_ONLY, MQTTMessage, MQTTMessageInfo
14from paho.mqtt.enums import CallbackAPIVersion, MQTTErrorCode, MQTTProtocolVersion
15from paho.mqtt.properties import Properties
16from paho.mqtt.reasoncodes import ReasonCode
18from updates2mqtt.model import Discovery, ReleaseProvider
20from .config import HomeAssistantConfig, MqttConfig, NodeConfig, PublishPolicy, TlsMode
21from .hass_formatter import hass_format_config, hass_format_state
23log = structlog.get_logger()
25MQTT_NAME = r"[A-Za-z0-9_\-\.]+"
28@dataclass
29class LocalMessage:
30 topic: str | None = field(default=None)
31 payload: str | None = field(default=None)
34class MqttPublisher:
35 def __init__(self, cfg: MqttConfig, node_cfg: NodeConfig, hass_cfg: HomeAssistantConfig) -> None:
36 self.cfg: MqttConfig = cfg
37 self.node_cfg: NodeConfig = node_cfg
38 self.hass_cfg: HomeAssistantConfig = hass_cfg
39 self.providers_by_topic: dict[str, ReleaseProvider] = {}
40 self.providers_by_type: dict[str, ReleaseProvider] = {}
41 self.event_loop: asyncio.AbstractEventLoop | None = None
42 self.client: mqtt.Client | None = None
43 self.fatal_failure = Event()
44 self.connected = Event()
45 self.commands_in_progress: set[tuple[str, str]] = set()
46 self.log = structlog.get_logger().bind(host=cfg.host, integration="mqtt")
48 def start(self, event_loop: asyncio.AbstractEventLoop | None = None) -> None:
49 logger = self.log.bind(action="start")
50 try:
51 protocol: MQTTProtocolVersion
52 if self.cfg.protocol in ("3", "3.11"):
53 protocol = MQTTProtocolVersion.MQTTv311
54 elif self.cfg.protocol == "3.1":
55 protocol = MQTTProtocolVersion.MQTTv31
56 elif self.cfg.protocol in ("5", "5.0"):
57 protocol = MQTTProtocolVersion.MQTTv5
58 else:
59 logger.info("No valid MQTT protocol version found (%s), setting to default v3.11", self.cfg.protocol)
60 protocol = MQTTProtocolVersion.MQTTv311
61 logger.debug("MQTT protocol set to %r", protocol)
63 self.event_loop = event_loop or asyncio.get_event_loop()
64 self.client = mqtt.Client(
65 callback_api_version=CallbackAPIVersion.VERSION2,
66 client_id=f"updates2mqtt_{self.node_cfg.name}",
67 clean_session=True if protocol != MQTTProtocolVersion.MQTTv5 else None,
68 protocol=protocol,
69 transport=cast(Literal["tcp", "websockets", "unix"], self.cfg.transport),
70 )
71 self.client.username_pw_set(self.cfg.user, password=self.cfg.password)
73 if self.cfg.tls_mode != TlsMode.OFF:
74 logger.debug("Configuring TLS, ca_certs=%s,cert_reqs=%s", self.cfg.ca_certs, self.cfg.cert_reqs)
75 # https://eclipse.dev/paho/files/paho.mqtt.python/html/client.html#paho.mqtt.client.Client.tls_set
76 self.client.tls_set(
77 ca_certs=self.cfg.ca_certs or None,
78 certfile=self.cfg.client_cert or None,
79 keyfile=self.cfg.client_key or None,
80 keyfile_password=self.cfg.client_key_password or None,
81 cert_reqs=self.cfg.cert_reqs or None,
82 )
83 if self.cfg.tls_mode == TlsMode.INSECURE:
84 self.client.tls_insecure_set(True)
85 logger.warning(
86 "TLS set to insecure mode, broker host name will not be validated, do not use for production."
87 )
88 rc: MQTTErrorCode = self.client.connect(
89 host=self.cfg.host,
90 port=self.cfg.port,
91 keepalive=self.cfg.keepalive,
92 clean_start=MQTT_CLEAN_START_FIRST_ONLY,
93 )
94 logger.info("Client connection requested", result_code=rc)
96 self.client.on_connect = self.on_connect
97 self.client.on_disconnect = self.on_disconnect
98 self.client.on_message = self.on_message
99 self.client.on_subscribe = self.on_subscribe
100 self.client.on_unsubscribe = self.on_unsubscribe
102 self.client.loop_start()
104 if not self.connected.wait(timeout=self.cfg.connect_timeout) and not self.fatal_failure.is_set(): 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true
105 logger.warning("Timed out waiting for broker connection, continuing anyway", timeout=self.cfg.connect_timeout)
107 logger.debug("MQTT Publisher loop started", host=self.cfg.host, port=self.cfg.port)
108 except Exception as e:
109 logger.error("Failed to connect to broker", host=self.cfg.host, port=self.cfg.port, error=str(e))
110 raise OSError(f"Connection Failure to {self.cfg.host}:{self.cfg.port} as {self.cfg.user} -- {e}") from e
112 def stop(self) -> None:
113 if self.client:
114 self.client.loop_stop()
115 self.client.disconnect()
116 self.client = None
117 self.connected.clear()
119 def is_available(self) -> bool:
120 return self.client is not None and not self.fatal_failure.is_set()
122 def on_connect(
123 self, _client: mqtt.Client, _userdata: Any, _flags: mqtt.ConnectFlags, rc: ReasonCode, _props: Properties | None
124 ) -> None:
125 if not self.client or self.fatal_failure.is_set():
126 self.log.warning("No client, check if started and authorized")
127 return
128 if rc.getName() == "Not authorized":
129 self.fatal_failure.set()
130 log.error("Invalid MQTT credentials", result_code=rc)
131 return
132 if rc != 0:
133 self.log.warning("Connection failed to broker", result_code=rc)
134 else:
135 self.log.debug("Connected successfully to MQTT broker")
136 self.connected.set()
137 for topic, provider in self.providers_by_topic.items():
138 self.log.debug("(Re)subscribing", topic=topic, provider=provider.source_type)
139 self.client.subscribe(topic)
141 def on_disconnect(
142 self,
143 _client: mqtt.Client,
144 _userdata: Any,
145 _disconnect_flags: mqtt.DisconnectFlags,
146 rc: ReasonCode,
147 _props: Properties | None,
148 ) -> None:
149 self.connected.clear()
150 if rc == 0:
151 self.log.debug("Disconnected from broker", result_code=rc)
152 else:
153 self.log.warning("Disconnect failure from broker", result_code=rc)
155 async def clean_topics(
156 self, provider: ReleaseProvider, wait_time: int = 5, max_time: int = 120, initial: bool = False
157 ) -> None:
158 logger = self.log.bind(action="clean")
160 if self.fatal_failure.is_set():
161 return
162 try:
163 logger.info("Starting clean cycle, wait time: %s, max time: %s, initial: %s", wait_time, max_time, initial)
164 cutoff_time: float = time.time() + max_time
165 cleaner = mqtt.Client(
166 callback_api_version=CallbackAPIVersion.VERSION1,
167 client_id=f"updates2mqtt_clean_{self.node_cfg.name}",
168 clean_session=True,
169 )
170 results = {"cleaned": 0, "matched": 0, "discovered": 0, "last_timestamp": time.time()}
171 cleaner.username_pw_set(self.cfg.user, password=self.cfg.password)
172 cleaner.connect(host=self.cfg.host, port=self.cfg.port, keepalive=self.cfg.keepalive)
174 def cleanup(_client: mqtt.Client, _userdata: Any, msg: mqtt.MQTTMessage) -> None:
175 discovery: Discovery | None = None
176 if msg.topic.startswith(
177 f"{self.hass_cfg.discovery.prefix}/update/{self.node_cfg.name}_{provider.source_type}_"
178 ):
179 discovery = self.reverse_config_topic(msg.topic, provider.source_type)
180 elif msg.topic.startswith( 180 ↛ 183line 180 didn't jump to line 183 because the condition on line 180 was never true
181 f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/"
182 ) and msg.topic.endswith("/state"):
183 discovery = self.reverse_state_topic(msg.topic, provider.source_type)
184 elif msg.topic.startswith(f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/"): 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 discovery = self.reverse_general_topic(msg.topic, provider.source_type)
186 else:
187 logger.debug("Ignoring other topic ", topic=msg.topic)
188 return
189 results["discovered"] += 1
190 if not initial and discovery is None:
191 logger.debug("Removing unknown discovery", topic=msg.topic)
192 cleaner.publish(msg.topic, "", retain=True)
193 results["cleaned"] += 1
194 elif discovery is not None: 194 ↛ 197line 194 didn't jump to line 197 because the condition on line 194 was always true
195 results["matched"] += 1
197 try:
198 if msg.payload: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 payload = json.loads(msg.payload)
200 update_section = payload.get("update") if isinstance(payload.get("update"), dict) else None
201 lingering_in_progress = payload.get("in_progress") or (
202 update_section and update_section.get("in_progress")
203 )
204 if lingering_in_progress and initial:
205 logger.info("Clearing lingering in-progress state at %s", msg.topic)
206 payload["in_progress"] = False
207 if update_section is not None:
208 update_section["in_progress"] = False
209 cleaner.publish(msg.topic, json.dumps(payload), retain=True)
210 results["cleaned"] += 1
211 elif (
212 initial
213 and msg.topic.endswith("/state")
214 and (payload.get("installed_version") is None or payload.get("latest_version") is None)
215 ):
216 # Stale/incomplete state message (e.g. from an older schema) leaves HA showing
217 # "unknown" forever, since it has nothing to compare against. Clear it so the
218 # upcoming scan can publish a complete replacement.
219 logger.info("Clearing stale incomplete state at %s", msg.topic)
220 cleaner.publish(msg.topic, "", retain=True)
221 results["cleaned"] += 1
222 except Exception as e:
223 logger.warning("Invalid payload at %s: %s", msg.topic, e)
224 cleaner.publish(msg.topic, "", retain=True)
225 results["cleaned"] += 1
227 results["last_timestamp"] = time.time()
229 cleaner.on_message = cleanup
230 options = paho.mqtt.subscribeoptions.SubscribeOptions(noLocal=True)
231 cleaner.subscribe(f"{self.hass_cfg.discovery.prefix}/update/#", options=options)
232 cleaner.subscribe(f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/#", options=options)
234 while time.time() - results["last_timestamp"] <= wait_time and time.time() <= cutoff_time:
235 cleaner.loop(0.5)
237 logger.info(
238 f"Cleaned - discovered:{results['discovered']}, matched:{results['matched']}, cleaned:{results['cleaned']}"
239 )
240 except Exception:
241 logger.exception("Cleaning topics of stale entries failed")
243 def safe_json_decode(self, jsonish: str | bytes | None) -> dict:
244 if jsonish is None:
245 return {}
246 try:
247 return json.loads(jsonish)
248 except Exception:
249 log.exception("JSON decode fail (%s)", jsonish)
250 try:
251 return json.loads(jsonish[1:-1])
252 except Exception:
253 log.exception("JSON decode fail (%s)", jsonish[1:-1])
254 return {}
256 def validate_command(self, msg: MQTTMessage | LocalMessage) -> tuple[ReleaseProvider, str, str] | None:
258 logger = self.log.bind(topic=msg.topic, payload=msg.payload)
259 comp_name: str | None = None
260 command: str | None = None
261 try:
262 logger.info("Command received for %s", msg.topic)
263 source_type: str | None = None
265 payload: str | None = None
266 if isinstance(msg.payload, bytes):
267 payload = msg.payload.decode("utf-8")
268 elif isinstance(msg.payload, str): 268 ↛ 270line 268 didn't jump to line 270 because the condition on line 268 was always true
269 payload = msg.payload
270 if payload and "|" in payload:
271 source_type, comp_name, command = payload.split("|")
272 else:
273 logger.warning("Invalid command format, expecting `source_type|comp_name|command`")
274 return None
275 logger.debug("Validating %s:%s:%s", source_type, comp_name, command)
277 provider: ReleaseProvider | None = self.providers_by_topic.get(msg.topic) if msg.topic else None
279 if not provider: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true
280 logger.warning("Unexpected provider type %s", msg.topic)
281 return None
282 if source_type is None or provider.source_type != source_type:
283 logger.warning("Unexpected source type %s", source_type)
284 return None
285 if command != "install": 285 ↛ 286line 285 didn't jump to line 286 because the condition on line 285 was never true
286 logger.warning("Unknown command: %s", command)
287 return None
288 if not comp_name: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true
289 logger.warning("Missing comp_name in command message: %s", msg.payload)
290 return None
292 in_progress_key: tuple[str, str] = (source_type, comp_name)
293 if in_progress_key in self.commands_in_progress:
294 logger.warning("Ignoring duplicate %s command for %s, already in progress", command, comp_name)
295 else:
296 self.commands_in_progress.add(in_progress_key)
297 return (provider, comp_name, command)
298 except Exception:
299 logger.error("Unexpected error validating command")
300 return None
302 async def execute_command(
303 self, provider: ReleaseProvider, comp_name: str, command: str, on_update_start: Callable, on_update_end: Callable
304 ) -> None:
305 # TODO: defer handling of commands where repository is throttled
306 logger = self.log.bind(source_type=provider.source_type, comp_name=comp_name, command=command)
307 try:
308 logger.info("Execution starting for %s %s", command, comp_name)
310 in_progress_key: tuple[str, str] = (provider.source_type, comp_name)
311 logger.info(
312 "Passing %s command to %s scanner for %s",
313 command,
314 provider.source_type,
315 comp_name,
316 )
317 try:
318 updated: bool = provider.command(comp_name, command, on_update_start, on_update_end)
319 discovery = provider.resolve(comp_name)
320 if updated and discovery: 320 ↛ 328line 320 didn't jump to line 328 because the condition on line 320 was always true
321 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT and self.hass_cfg.discovery.enabled:
322 self.publish_hass_config(discovery)
323 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):
324 self.publish_discovery(discovery)
325 if discovery and discovery.publish_policy == PublishPolicy.HOMEASSISTANT:
326 self.publish_hass_state(discovery)
327 else:
328 logger.debug("No change to republish after execution")
329 finally:
330 if in_progress_key in self.commands_in_progress: 330 ↛ 332line 330 didn't jump to line 332 because the condition on line 330 was always true
331 self.commands_in_progress.discard(in_progress_key)
332 logger.info("Execution ended")
333 except Exception:
334 logger.exception("Execution failed")
336 def local_message(self, discovery: Discovery, command: str) -> None:
337 """Simulate an incoming MQTT message for local commands"""
338 msg = LocalMessage(
339 topic=self.command_topic(discovery.provider), payload=f"{discovery.source_type}|{discovery.name}|{command}"
340 )
341 self.handle_message(msg)
343 def on_subscribe(
344 self,
345 _client: mqtt.Client,
346 userdata: Any,
347 mid: int,
348 reason_code_list: list[ReasonCode],
349 properties: Properties | None = None,
350 ) -> None:
351 self.log.debug(
352 "on_subscribe, userdata=%s, mid=%s, reasons=%s, properties=%s", userdata, mid, reason_code_list, properties
353 )
355 def on_unsubscribe(
356 self,
357 _client: mqtt.Client,
358 userdata: Any,
359 mid: int,
360 reason_code_list: list[ReasonCode],
361 properties: Properties | None = None,
362 ) -> None:
363 self.log.debug(
364 "on_unsubscribe, userdata=%s, mid=%s, reasons=%s, properties=%s", userdata, mid, reason_code_list, properties
365 )
367 def on_message(self, _client: mqtt.Client, _userdata: Any, msg: mqtt.MQTTMessage) -> None:
368 """Callback for incoming MQTT messages"""
369 if msg.topic in self.providers_by_topic:
370 self.handle_message(msg)
371 else:
372 # apparently the root non-wildcard sub sometimes brings in child topics
373 self.log.debug("Unhandled message #%s on %s:%s", msg.mid, msg.topic, msg.payload)
375 def handle_message(self, msg: mqtt.MQTTMessage | LocalMessage) -> None:
376 def update_start(discovery: Discovery) -> None:
377 self.log.debug("on_update_start: %s", topic=msg.topic)
378 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT:
379 self.publish_hass_state(discovery, in_progress=True)
380 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):
381 self.publish_discovery(discovery, in_progress=True)
383 def update_end(discovery: Discovery) -> None:
384 self.log.debug("on_update_end: %s", topic=msg.topic)
385 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT:
386 self.publish_hass_state(discovery, in_progress=False)
387 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):
388 self.publish_discovery(discovery, in_progress=False)
390 # TODO: fix double publish on callback and in command exec
391 if self.event_loop is not None: 391 ↛ 407line 391 didn't jump to line 407 because the condition on line 391 was always true
392 self.log.debug("Executing command topic", topic=msg.topic)
393 parsed: tuple[ReleaseProvider, str, str] | None = self.validate_command(msg=msg)
394 if parsed is not None: 394 ↛ exitline 394 didn't return from function 'handle_message' because the condition on line 394 was always true
395 provider, comp_name, command = parsed
396 asyncio.run_coroutine_threadsafe(
397 self.execute_command(
398 provider=provider,
399 comp_name=comp_name,
400 command=command,
401 on_update_start=update_start,
402 on_update_end=update_end,
403 ),
404 loop=self.event_loop,
405 )
406 else:
407 self.log.error("No event loop to handle message", topic=msg.topic)
409 def config_topic(self, discovery: Discovery) -> str:
410 prefix = self.hass_cfg.discovery.prefix
411 return f"{prefix}/update/{self.node_cfg.name}_{discovery.source_type}_{discovery.name}/update/config"
413 def reverse_config_topic(self, topic: str, source_type: str) -> Discovery | None:
414 match = re.fullmatch(
415 f"{self.hass_cfg.discovery.prefix}/update/{self.node_cfg.name}_{source_type}_({MQTT_NAME})/update/config",
416 topic,
417 )
418 if match and len(match.groups()) == 1: 418 ↛ 423line 418 didn't jump to line 423 because the condition on line 418 was always true
419 discovery_name: str = match.group(1)
420 if source_type in self.providers_by_type and discovery_name in self.providers_by_type[source_type].discoveries:
421 return self.providers_by_type[source_type].discoveries[discovery_name]
423 self.log.debug("MQTT CONFIG no match for %s", topic)
424 return None
426 def state_topic(self, discovery: Discovery) -> str:
427 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{discovery.source_type}/{discovery.name}/state"
429 def reverse_state_topic(self, topic: str, source_type: str) -> Discovery | None:
430 match = re.fullmatch(
431 f"{self.cfg.topic_root}/{self.node_cfg.name}/{source_type}/({MQTT_NAME})/state",
432 topic,
433 )
434 if match and len(match.groups()) == 1:
435 discovery_name: str = match.group(1)
436 if discovery_name in self.providers_by_type[source_type].discoveries:
437 return self.providers_by_type[source_type].discoveries[discovery_name]
439 self.log.debug("MQTT STATE no match for %s", topic)
440 return None
442 def general_topic(self, discovery: Discovery) -> str:
443 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{discovery.source_type}/{discovery.name}"
445 def reverse_general_topic(self, topic: str, source_type: str) -> Discovery | None:
446 match = re.fullmatch(f"{self.cfg.topic_root}/{self.node_cfg.name}/{source_type}/({MQTT_NAME})", topic)
447 if match and len(match.groups()) == 1:
448 discovery_name: str = match.group(1)
449 if discovery_name in self.providers_by_type[source_type].discoveries:
450 return self.providers_by_type[source_type].discoveries[discovery_name]
452 self.log.debug("MQTT ATTR no match for %s", topic)
453 return None
455 def command_topic(self, provider: ReleaseProvider) -> str:
456 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}"
458 def publish_discovery(self, discovery: Discovery, in_progress: bool = False) -> None:
459 """Comprehensive, non Home Assistant specific, base publication"""
460 if discovery.publish_policy not in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT): 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 return
462 self.log.debug("Discovery publish: %s", discovery)
463 payload: dict[str, Any] = discovery.as_dict()
464 payload["update"]["in_progress"] = in_progress # ty:ignore[invalid-assignment]
465 if payload.get("release", {}).get("summary") and self.hass_cfg.release_summary_max_size: 465 ↛ 466line 465 didn't jump to line 466 because the condition on line 465 was never true
466 payload["release"]["summary"] = payload["release"]["summary"][: self.hass_cfg.release_summary_max_size]
467 self.publish(self.general_topic(discovery), payload)
469 def publish_hass_state(self, discovery: Discovery, in_progress: bool = False) -> None:
470 if discovery.publish_policy != PublishPolicy.HOMEASSISTANT: 470 ↛ 471line 470 didn't jump to line 471 because the condition on line 470 was never true
471 return
472 self.log.debug("HASS State update, in progress: %s, discovery: %s", in_progress, discovery)
473 self.publish(
474 self.state_topic(discovery),
475 hass_format_state(
476 discovery, in_progress=in_progress, release_summary_max_size=self.hass_cfg.release_summary_max_size
477 ),
478 )
480 def publish_hass_config(self, discovery: Discovery) -> None:
481 if discovery.publish_policy != PublishPolicy.HOMEASSISTANT: 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 return
483 object_id = f"{discovery.source_type}_{self.node_cfg.name}_{discovery.name}"
484 self.log.debug("HASS Config: %s", object_id)
486 self.publish(
487 self.config_topic(discovery),
488 hass_format_config(
489 discovery=discovery,
490 object_id=object_id,
491 area=self.hass_cfg.area,
492 state_topic=self.state_topic(discovery),
493 attrs_topic=self.general_topic(discovery) if self.hass_cfg.extra_attributes else None,
494 command_topic=self.command_topic(discovery.provider),
495 force_command_topic=self.hass_cfg.force_command_topic,
496 device_creation=self.hass_cfg.device_creation,
497 ),
498 )
500 def subscribe_hass_command(self, provider: ReleaseProvider):
501 topic = self.command_topic(provider)
502 if topic in self.providers_by_topic or self.client is None:
503 self.log.debug("Skipping subscription", topic=topic)
504 else:
505 self.log.info("Handler subscribing", topic=topic)
506 self.providers_by_topic[topic] = provider
507 self.providers_by_type[provider.source_type] = provider
508 self.client.subscribe(topic)
509 return topic
511 def loop_once(self) -> None:
512 if self.client:
513 self.client.loop()
515 def publish(self, topic: str, payload: dict, qos: int = 1, retain: bool = True) -> None:
516 if self.client:
517 info: MQTTMessageInfo = self.client.publish(topic, payload=json.dumps(payload), qos=qos, retain=retain)
518 if info.rc == MQTTErrorCode.MQTT_ERR_SUCCESS:
519 self.log.debug(
520 "Publish to %s, mid: %s, published: %s, qos: %s, rc: %s", topic, info.mid, info.is_published(), qos, info.rc
521 )
522 elif info.rc == MQTTErrorCode.MQTT_ERR_NO_CONN and qos > 0: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 self.log.debug(
524 "Not currently connected, queued for delivery on reconnect: %s, mid: %s, qos: %s",
525 topic,
526 info.mid,
527 qos,
528 )
529 else:
530 self.log.warning("Problem publishing to %s, mid: %s, qos: %s, rc: %s", topic, info.mid, qos, info.rc)
531 else:
532 self.log.debug("No client to publish at %s", topic)