diff --git a/app/api/routes/api/config.py b/app/api/routes/api/config.py index ee91333..0cc239b 100644 --- a/app/api/routes/api/config.py +++ b/app/api/routes/api/config.py @@ -9,12 +9,14 @@ from sqlalchemy.orm import Session from app.api.routes.api.deps import require_csrf, require_session from app.config import Settings, get_settings from app.dependencies import get_app_settings, get_db +from app.integrations.mqtt import MQTT_SETTINGS_KEYS, mqtt_manager from app.schemas.config import ( ConfigField, ConfigResponse, ConfigSection, ConfigUpdateRequest, ConfigUpdateResponse, + MqttTestResponse, SmtpTestResponse, ) from app.services.auth import AuthenticatedSession @@ -58,7 +60,12 @@ def put_config( - Blank secret value keeps the existing stored value (no change). - Invalid values return 422 and nothing is written to the database. + - If MQTT-related settings changed, the MQTT client reconnects automatically. """ + # Detect whether any MQTT-related key is being submitted (non-secret change + # or non-blank secret change) so we know to reconnect after saving. + mqtt_keys_submitted = any(k.lower() in MQTT_SETTINGS_KEYS for k in body.updates) + try: save_config_updates(db, body.updates, settings) except ConfigSaveError as exc: @@ -70,6 +77,12 @@ def put_config( # Re-read settings after save (save_config_updates clears the settings cache) refreshed_settings = get_settings() + + # Reconnect MQTT client if any MQTT setting was updated. + if mqtt_keys_submitted: + logger.info("MQTT settings changed — triggering reconnect.") + mqtt_manager.reconnect(refreshed_settings) + sections_raw = build_config_sections(db, refreshed_settings) return ConfigUpdateResponse(sections=_sections_from_raw(sections_raw)) @@ -117,3 +130,174 @@ def post_smtp_test( status_code=status.HTTP_200_OK, content={"result": "success", "message": "Test email sent successfully."}, ) + + +# --------------------------------------------------------------------------- +# POST /api/config/mqtt/test — M5-T10 +# --------------------------------------------------------------------------- + +class _MqttConfigurationError(ValueError): + """Raised when MQTT settings are incomplete or disabled.""" + + +class _MqttConnectionError(RuntimeError): + """Raised when MQTT broker connection or publish fails.""" + + +@router.post( + "/config/mqtt/test", + responses={ + 200: {"model": MqttTestResponse}, + 400: {"model": MqttTestResponse}, + 502: {"model": MqttTestResponse}, + }, +) +def post_mqtt_test( + settings: Settings = Depends(get_app_settings), + _auth: AuthenticatedSession = Depends(require_session), + _csrf: None = Depends(require_csrf), +) -> JSONResponse: + """ + Test MQTT broker connectivity by attempting to connect and publishing a + test message to ``/home-automation/test``. + + The message is visible in MQTT Explorer (or any subscriber) so users can + confirm the full broker publish path is working. + + Three possible outcomes: + - 200 { "result": "success", "message": ... } + - 400 { "result": "config-error", "message": ... } (not configured) + - 502 { "result": "failed", "message": ... } (connection/publish error) + + MQTT credentials are never echoed in the response. + """ + try: + _run_mqtt_test(settings) + except _MqttConfigurationError as exc: + logger.warning("MQTT test rejected due to configuration: %s", exc) + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content={"result": "config-error", "message": str(exc)}, + ) + except _MqttConnectionError as exc: + logger.warning("MQTT test connection/publish failed: %s", exc) + return JSONResponse( + status_code=status.HTTP_502_BAD_GATEWAY, + content={"result": "failed", "message": str(exc)}, + ) + + return JSONResponse( + status_code=status.HTTP_200_OK, + content={ + "result": "success", + "message": ( + f"Test message published to " + f"{settings.ha_discovery_prefix}/home-automation/test." + ), + }, + ) + + +def _run_mqtt_test(settings: Settings) -> None: + """Attempt a transient MQTT connection and publish a test message. + + Raises + ------ + _MqttConfigurationError + When MQTT is not enabled or broker host is not configured. + _MqttConnectionError + When the broker is unreachable or the publish fails. + """ + import socket + + import paho.mqtt.client as mqtt + + if not settings.mqtt_broker_host: + raise _MqttConfigurationError("MQTT broker host is not configured.") + + test_topic = f"{settings.ha_discovery_prefix}/home-automation/test" + test_payload = '{"source": "home-automation", "event": "mqtt_test"}' + + connected_event = __import__("threading").Event() + published_event = __import__("threading").Event() + connect_error: list[str] = [] + + client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id="home-automation-test", + ) + + def _on_connect( + _client: mqtt.Client, + _userdata: object, + _flags: mqtt.ConnectFlags, + reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + if reason_code.is_failure: + connect_error.append(f"Broker refused connection: {reason_code}") + connected_event.set() + + def _on_publish( + _client: mqtt.Client, + _userdata: object, + _mid: int, + _reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + published_event.set() + + client.on_connect = _on_connect + client.on_publish = _on_publish + + if settings.mqtt_tls_enabled: + try: + client.tls_set() + except Exception as exc: + raise _MqttConnectionError(f"TLS setup failed: {exc}") from exc + + if settings.mqtt_username: + client.username_pw_set( + username=settings.mqtt_username, + password=settings.mqtt_password or None, + ) + + client.loop_start() + try: + try: + client.connect( + host=settings.mqtt_broker_host, + port=settings.mqtt_broker_port, + keepalive=10, + ) + except (OSError, socket.error) as exc: + # Sanitise: ensure password never leaks into the error message. + msg = _sanitize_mqtt_error(str(exc), settings.mqtt_password) + raise _MqttConnectionError(f"Cannot reach broker: {msg}") from exc + + # Wait up to 5 s for connection acknowledgement. + if not connected_event.wait(timeout=5): + raise _MqttConnectionError("Broker connection timed out (5 s).") + + if connect_error: + raise _MqttConnectionError(connect_error[0]) + + # Publish test message. + client.publish(test_topic, payload=test_payload, qos=1, retain=False) + # Wait up to 5 s for the publish ACK (QoS 1). + if not published_event.wait(timeout=5): + raise _MqttConnectionError("Publish timed out (5 s) — broker reachable but no ACK.") + + finally: + try: + client.disconnect() + except Exception: + pass + client.loop_stop() + + +def _sanitize_mqtt_error(message: str, password: str | None) -> str: + """Replace *password* in *message* with ``[redacted]``.""" + if password: + return message.replace(password, "[redacted]") + return message diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py new file mode 100644 index 0000000..e58d3ab --- /dev/null +++ b/app/integrations/mqtt.py @@ -0,0 +1,257 @@ +"""MQTT client integration (paho-mqtt 2.x). + +Provides :class:`MqttManager`: a long-lived MQTT client that runs paho's +loop in a background thread. Designed to be started in FastAPI's lifespan +and shared across the application as a module-level singleton (``mqtt_manager``). + +Key design decisions +-------------------- +- **paho 2.x API**: ``Client`` requires ``CallbackAPIVersion.VERSION2`` as the + first positional argument. VERSION2 on_connect callback receives + ``(client, userdata, connect_flags, reason_code, properties)`` — not the + older RC int. +- **Background thread**: ``loop_start()`` spawns a daemon thread; ``loop_stop()`` + joins it cleanly on shutdown. +- **Never crash the process**: all paho operations are wrapped in try/except; + connection failure is logged but does not propagate to the caller. +- **Password safety**: the MQTT password is never passed to ``logger`` calls. +- **Reconnect**: ``reconnect(settings)`` tears down the old client and + establishes a fresh connection with the new settings. Callers (e.g. PUT + /api/config) call this after saving MQTT-related config values. +""" + +from __future__ import annotations + +import logging +import threading +from typing import TYPE_CHECKING + +import paho.mqtt.client as mqtt + +if TYPE_CHECKING: + from app.config import Settings + +logger = logging.getLogger(__name__) + +# MQTT settings keys that, when changed, should trigger a reconnect. +MQTT_SETTINGS_KEYS = { + "mqtt_enabled", + "mqtt_broker_host", + "mqtt_broker_port", + "mqtt_username", + "mqtt_password", + "mqtt_tls_enabled", +} + + +def _is_configured(settings: Settings) -> bool: + """Return True if MQTT is enabled *and* the broker host is set.""" + return bool(settings.mqtt_enabled and settings.mqtt_broker_host) + + +class MqttManager: + """Long-lived MQTT client wrapper. + + Lifecycle + --------- + 1. ``connect(settings)`` — start background loop + connect to broker. + 2. ``publish(...)`` — publish messages while connected. + 3. ``disconnect()`` — graceful teardown (called on app shutdown). + 4. ``reconnect(settings)`` — disconnect then re-connect with new settings + (called after saving updated MQTT configuration). + + When MQTT is not enabled or not configured, all methods are no-ops. + Connection failures are caught and logged; they do not raise. + """ + + def __init__(self) -> None: + self._client: mqtt.Client | None = None + self._lock = threading.Lock() + self._connected = False + + # ------------------------------------------------------------------ + # Public properties + # ------------------------------------------------------------------ + + @property + def is_connected(self) -> bool: + """True if the underlying paho client is currently connected.""" + return self._connected + + # ------------------------------------------------------------------ + # Public interface + # ------------------------------------------------------------------ + + def is_configured(self, settings: Settings) -> bool: + """Return True if MQTT is enabled and broker host is configured.""" + return _is_configured(settings) + + def connect(self, settings: Settings) -> None: + """Connect to the MQTT broker and start the background loop thread. + + No-op if MQTT is not enabled or broker host is not set. + Connection errors are logged but do not raise. + """ + if not _is_configured(settings): + logger.debug("MQTT not configured or not enabled — skipping connect.") + return + + with self._lock: + self._start_client(settings) + + def disconnect(self) -> None: + """Disconnect from the broker and stop the background loop thread. + + No-op if no client is active. + """ + with self._lock: + self._stop_client() + + def reconnect(self, settings: Settings) -> None: + """Disconnect the current client (if any) and reconnect with *settings*. + + Call this after saving updated MQTT configuration values. + No-op if MQTT is not enabled or broker host is not set. + """ + with self._lock: + self._stop_client() + self.connect(settings) + + def publish( + self, + topic: str, + payload: str | bytes | None, + *, + retain: bool = False, + qos: int = 0, + ) -> None: + """Publish a message to *topic*. + + If the client is not connected the call is silently skipped. + Errors are logged but do not raise. + """ + with self._lock: + client = self._client + + if client is None or not self._connected: + logger.debug("MQTT publish skipped — not connected (topic=%s).", topic) + return + + try: + client.publish(topic, payload=payload, qos=qos, retain=retain) + except Exception: + logger.exception("MQTT publish error (topic=%s).", topic) + + # ------------------------------------------------------------------ + # Internal helpers — must be called with self._lock held + # ------------------------------------------------------------------ + + def _start_client(self, settings: Settings) -> None: + """Build a fresh paho Client, configure it, and call loop_start + connect.""" + client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id="home-automation", + ) + + # Callbacks — VERSION2 on_connect signature: + # (client, userdata, connect_flags, reason_code, properties) + def _on_connect( + _client: mqtt.Client, + _userdata: object, + _flags: mqtt.ConnectFlags, + reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + if reason_code.is_failure: + logger.warning("MQTT connection refused: %s", reason_code) + self._connected = False + else: + logger.info("MQTT connected (reason_code=%s).", reason_code) + self._connected = True + + # VERSION2 on_disconnect signature: + # (client, userdata, disconnect_flags, reason_code, properties) + def _on_disconnect( + _client: mqtt.Client, + _userdata: object, + _disconnect_flags: mqtt.DisconnectFlags, + reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + self._connected = False + if reason_code.is_failure: + logger.warning("MQTT disconnected unexpectedly (reason_code=%s).", reason_code) + else: + logger.info("MQTT disconnected cleanly.") + + client.on_connect = _on_connect + client.on_disconnect = _on_disconnect + + # TLS + if settings.mqtt_tls_enabled: + try: + client.tls_set() + except Exception: + logger.exception("MQTT TLS setup failed.") + return + + # Credentials — never log the password + if settings.mqtt_username: + client.username_pw_set( + username=settings.mqtt_username, + password=settings.mqtt_password or None, + ) + + # Start background loop thread *before* connect so paho can handle + # the connection handshake asynchronously. + client.loop_start() + + try: + client.connect( + host=settings.mqtt_broker_host, + port=settings.mqtt_broker_port, + keepalive=60, + ) + except Exception: + # Log without leaking the password + logger.exception( + "MQTT connect failed (host=%s, port=%s). Stopping loop.", + settings.mqtt_broker_host, + settings.mqtt_broker_port, + ) + try: + client.loop_stop() + except Exception: + pass + return + + self._client = client + logger.info( + "MQTT client started (host=%s, port=%s).", + settings.mqtt_broker_host, + settings.mqtt_broker_port, + ) + + def _stop_client(self) -> None: + """Disconnect and stop the background loop. Idempotent.""" + client = self._client + if client is None: + return + self._client = None + self._connected = False + try: + client.disconnect() + except Exception: + logger.debug("MQTT disconnect raised (ignoring).", exc_info=True) + try: + client.loop_stop() + except Exception: + logger.debug("MQTT loop_stop raised (ignoring).", exc_info=True) + logger.info("MQTT client stopped.") + + +# --------------------------------------------------------------------------- +# Module-level singleton — shared across lifespan and route handlers +# --------------------------------------------------------------------------- + +mqtt_manager = MqttManager() diff --git a/app/main.py b/app/main.py index f6688a7..1606e7a 100644 --- a/app/main.py +++ b/app/main.py @@ -23,6 +23,7 @@ from app.api.routes.poo import router as poo_router from app.api.routes.public_ip import router as public_ip_router from app.api.routes.ticktick import router as ticktick_router from app.config import get_settings +from app.integrations.mqtt import mqtt_manager from app.services.auth import AuthBootstrapError, initialize_auth_schema from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap from app.services.public_ip import check_public_ipv4_and_notify @@ -103,7 +104,15 @@ async def lifespan(_: FastAPI): coalesce=True, ) scheduler.start() + + # MQTT: connect if configured. + mqtt_manager.connect(get_settings()) + yield + + # MQTT: clean shutdown before the process exits. + mqtt_manager.disconnect() + scheduler.shutdown(wait=False) diff --git a/app/schemas/config.py b/app/schemas/config.py index eab94b1..9548f05 100644 --- a/app/schemas/config.py +++ b/app/schemas/config.py @@ -38,3 +38,10 @@ class SmtpTestResponse(BaseModel): result: Literal["success", "config-error", "failed"] message: str + + +class MqttTestResponse(BaseModel): + """Response from POST /api/config/mqtt/test.""" + + result: Literal["success", "config-error", "failed"] + message: str diff --git a/docs/design/m5-iot-energy.md b/docs/design/m5-iot-energy.md index 90f1443..a301889 100644 --- a/docs/design/m5-iot-energy.md +++ b/docs/design/m5-iot-energy.md @@ -416,7 +416,7 @@ Phase C(MQTT / Discovery,依赖 B 的设备数据与 provider 接口) - **Reviewer checklist**: `key` 稳定(用设备 `uuid` + 量 key,不用自增 id,避免重建漂移);component 支持多类型;目录由 provider 计算、元数据源自 profile 而非写死。 ### M5-T10 — MQTT 客户端(paho,lifespan 连接 + 重连) -- **Status**: `todo` · **Depends**: M5-T08 +- **Status**: `done` · **Depends**: M5-T08 - **Context**: 长连接 MQTT 客户端,配置变更可重连;含连接测试端点。 - **Files**: `create app/integrations/mqtt.py`;`modify app/main.py`(lifespan 起/停)、`app/api/routes/api/config.py`(`POST /api/config/mqtt/test`);`modify requirements.in`/`requirements.txt`(加 `paho-mqtt` 重新锁定);`create tests/test_mqtt_client.py` - **Steps**: diff --git a/openapi/openapi.json b/openapi/openapi.json index 6820a2f..dfb3c55 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -53,7 +53,7 @@ "api-config" ], "summary": "Put Config", - "description": "Save configuration updates.\n\n- Blank secret value keeps the existing stored value (no change).\n- Invalid values return 422 and nothing is written to the database.", + "description": "Save configuration updates.\n\n- Blank secret value keeps the existing stored value (no change).\n- Invalid values return 422 and nothing is written to the database.\n- If MQTT-related settings changed, the MQTT client reconnects automatically.", "operationId": "put_config_api_config_put", "parameters": [ { @@ -177,6 +177,76 @@ } } }, + "/api/config/mqtt/test": { + "post": { + "tags": [ + "api-config" + ], + "summary": "Post Mqtt Test", + "description": "Test MQTT broker connectivity by attempting to connect and publishing a\ntest message to ``/home-automation/test``.\n\nThe message is visible in MQTT Explorer (or any subscriber) so users can\nconfirm the full broker publish path is working.\n\nThree possible outcomes:\n- 200 { \"result\": \"success\", \"message\": ... }\n- 400 { \"result\": \"config-error\", \"message\": ... } (not configured)\n- 502 { \"result\": \"failed\", \"message\": ... } (connection/publish error)\n\nMQTT credentials are never echoed in the response.", + "operationId": "post_mqtt_test_api_config_mqtt_test_post", + "parameters": [ + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MqttTestResponse" + } + } + } + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MqttTestResponse" + } + } + }, + "description": "Bad Request" + }, + "502": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MqttTestResponse" + } + } + }, + "description": "Bad Gateway" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/api/locations": { "get": { "tags": [ @@ -2329,6 +2399,30 @@ "title": "ModbusTestReadResponse", "description": "Response for POST /api/modbus/devices/{uuid}/test.\n\nOn success ``ok=True`` and ``payload`` contains the decoded values.\nOn failure ``ok=False`` and ``error`` describes the problem." }, + "MqttTestResponse": { + "properties": { + "result": { + "type": "string", + "enum": [ + "success", + "config-error", + "failed" + ], + "title": "Result" + }, + "message": { + "type": "string", + "title": "Message" + } + }, + "type": "object", + "required": [ + "result", + "message" + ], + "title": "MqttTestResponse", + "description": "Response from POST /api/config/mqtt/test." + }, "PasswordChangeRequest": { "properties": { "current_password": { diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 083b1db..a74e67f 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -42,7 +42,9 @@ paths: - Blank secret value keeps the existing stored value (no change). - - Invalid values return 422 and nothing is written to the database.' + - Invalid values return 422 and nothing is written to the database. + + - If MQTT-related settings changed, the MQTT client reconnects automatically.' operationId: put_config_api_config_put parameters: - name: X-CSRF-Token @@ -127,6 +129,68 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /api/config/mqtt/test: + post: + tags: + - api-config + summary: Post Mqtt Test + description: 'Test MQTT broker connectivity by attempting to connect and publishing + a + + test message to ``/home-automation/test``. + + + The message is visible in MQTT Explorer (or any subscriber) so users can + + confirm the full broker publish path is working. + + + Three possible outcomes: + + - 200 { "result": "success", "message": ... } + + - 400 { "result": "config-error", "message": ... } (not configured) + + - 502 { "result": "failed", "message": ... } (connection/publish + error) + + + MQTT credentials are never echoed in the response.' + operationId: post_mqtt_test_api_config_mqtt_test_post + parameters: + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/MqttTestResponse' + '400': + content: + application/json: + schema: + $ref: '#/components/schemas/MqttTestResponse' + description: Bad Request + '502': + content: + application/json: + schema: + $ref: '#/components/schemas/MqttTestResponse' + description: Bad Gateway + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /api/locations: get: tags: @@ -1699,6 +1763,24 @@ components: On success ``ok=True`` and ``payload`` contains the decoded values. On failure ``ok=False`` and ``error`` describes the problem.' + MqttTestResponse: + properties: + result: + type: string + enum: + - success + - config-error + - failed + title: Result + message: + type: string + title: Message + type: object + required: + - result + - message + title: MqttTestResponse + description: Response from POST /api/config/mqtt/test. PasswordChangeRequest: properties: current_password: diff --git a/requirements.in b/requirements.in index 03db00e..e3c72fd 100644 --- a/requirements.in +++ b/requirements.in @@ -3,6 +3,7 @@ apscheduler>=3.10,<4.0 argon2-cffi>=25.1,<26.0 fastapi>=0.115,<0.116 httpx>=0.28,<1.0 +paho-mqtt>=2.0,<3.0 pymodbus>=3.6,<4.0 pydantic-settings>=2.6,<3.0 pyotp>=2.9,<3.0 diff --git a/requirements.txt b/requirements.txt index 1a8acb5..749bcd8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -49,6 +49,8 @@ mako==1.3.11 # via alembic markupsafe==3.0.3 # via mako +paho-mqtt==2.1.0 + # via -r requirements.in pycparser==2.23 # via cffi pydantic==2.13.2 diff --git a/tests/test_mqtt_client.py b/tests/test_mqtt_client.py new file mode 100644 index 0000000..88b4a9d --- /dev/null +++ b/tests/test_mqtt_client.py @@ -0,0 +1,433 @@ +"""Tests for M5-T10: MqttManager + POST /api/config/mqtt/test. + +All paho interaction is mocked — no real broker is required. +""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from app.integrations.mqtt import MqttManager + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + +def _make_settings( + *, + mqtt_enabled: bool = True, + mqtt_broker_host: str = "broker.test", + mqtt_broker_port: int = 1883, + mqtt_username: str = "", + mqtt_password: str = "", + mqtt_tls_enabled: bool = False, + ha_discovery_prefix: str = "homeassistant", +): + """Return a simple namespace that acts like a Settings object for MqttManager tests.""" + s = MagicMock() + s.mqtt_enabled = mqtt_enabled + s.mqtt_broker_host = mqtt_broker_host + s.mqtt_broker_port = mqtt_broker_port + s.mqtt_username = mqtt_username + s.mqtt_password = mqtt_password + s.mqtt_tls_enabled = mqtt_tls_enabled + s.ha_discovery_prefix = ha_discovery_prefix + return s + + +def _login(client: TestClient) -> None: + resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 200, f"Login failed: {resp.status_code}" + + +# --------------------------------------------------------------------------- +# MqttManager.is_configured +# --------------------------------------------------------------------------- + +def test_is_configured_returns_false_when_disabled() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_enabled=False, mqtt_broker_host="broker.test") + assert manager.is_configured(settings) is False + + +def test_is_configured_returns_false_when_host_empty() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_enabled=True, mqtt_broker_host="") + assert manager.is_configured(settings) is False + + +def test_is_configured_returns_true_when_enabled_and_host_set() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_enabled=True, mqtt_broker_host="broker.test") + assert manager.is_configured(settings) is True + + +# --------------------------------------------------------------------------- +# MqttManager.connect — no-op when not configured +# --------------------------------------------------------------------------- + +def test_connect_is_noop_when_disabled() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_enabled=False) + # Should not raise and should not create a client + with patch("app.integrations.mqtt.mqtt.Client") as mock_client_cls: + manager.connect(settings) + mock_client_cls.assert_not_called() + assert manager._client is None + + +def test_connect_is_noop_when_host_empty() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_broker_host="") + with patch("app.integrations.mqtt.mqtt.Client") as mock_client_cls: + manager.connect(settings) + mock_client_cls.assert_not_called() + assert manager._client is None + + +# --------------------------------------------------------------------------- +# MqttManager.connect — calls paho correctly when configured +# --------------------------------------------------------------------------- + +def test_connect_creates_paho_client_with_version2() -> None: + """connect() must pass CallbackAPIVersion.VERSION2 to paho.Client.""" + manager = MqttManager() + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client) as mock_cls: + with patch("app.integrations.mqtt.mqtt.CallbackAPIVersion"): + manager.connect(settings) + # Verify Client was instantiated with VERSION2 + mock_cls.assert_called_once() + call_kwargs = mock_cls.call_args + # First positional arg or 'callback_api_version' kwarg + assert call_kwargs is not None + + # loop_start must have been called + mock_client.loop_start.assert_called_once() + # connect must have been called with the right host/port + mock_client.connect.assert_called_once_with( + host="broker.test", port=1883, keepalive=60 + ) + + +def test_connect_sets_credentials_when_username_provided() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_username="user", mqtt_password="s3cr3t") + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + mock_client.username_pw_set.assert_called_once_with( + username="user", password="s3cr3t" + ) + + +def test_connect_skips_credentials_when_username_empty() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_username="", mqtt_password="") + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + mock_client.username_pw_set.assert_not_called() + + +def test_connect_calls_tls_set_when_tls_enabled() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_tls_enabled=True) + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + mock_client.tls_set.assert_called_once() + + +# --------------------------------------------------------------------------- +# MqttManager.connect — connection failure does not raise +# --------------------------------------------------------------------------- + +def test_connect_does_not_raise_on_connection_failure() -> None: + manager = MqttManager() + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.side_effect = OSError("Connection refused") + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + # Must not raise — failures are caught and logged + manager.connect(settings) + + # loop_stop should have been called to clean up the thread + mock_client.loop_stop.assert_called() + # Client reference should not be stored + assert manager._client is None + + +# --------------------------------------------------------------------------- +# MqttManager.disconnect +# --------------------------------------------------------------------------- + +def test_disconnect_stops_loop_and_clears_client() -> None: + manager = MqttManager() + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + manager._client = mock_client # ensure it's set + + manager.disconnect() + + mock_client.disconnect.assert_called() + mock_client.loop_stop.assert_called() + assert manager._client is None + + +def test_disconnect_is_noop_when_not_connected() -> None: + manager = MqttManager() + # Should not raise when called before any connect + manager.disconnect() + + +# --------------------------------------------------------------------------- +# MqttManager.publish +# --------------------------------------------------------------------------- + +def test_publish_passes_topic_payload_retain_to_paho() -> None: + manager = MqttManager() + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + # Simulate connected state + manager._connected = True + manager._client = mock_client + + manager.publish("test/topic", '{"key": "value"}', retain=True, qos=1) + + mock_client.publish.assert_called_once_with( + "test/topic", payload='{"key": "value"}', qos=1, retain=True + ) + + +def test_publish_is_noop_when_not_connected() -> None: + manager = MqttManager() + # No connect — should silently skip + manager.publish("topic", "payload", retain=False) + # No exception raised + + +def test_publish_does_not_raise_on_paho_error() -> None: + manager = MqttManager() + mock_client = MagicMock() + mock_client.publish.side_effect = RuntimeError("network error") + manager._client = mock_client + manager._connected = True + + # Must not raise + manager.publish("topic", "payload") + + +# --------------------------------------------------------------------------- +# MqttManager.reconnect +# --------------------------------------------------------------------------- + +def test_reconnect_stops_old_client_and_starts_new_one() -> None: + """reconnect() must tear down the existing client and establish a fresh one.""" + manager = MqttManager() + settings = _make_settings() + + # Simulate an already-connected client + old_client = MagicMock() + manager._client = old_client + manager._connected = True + + new_client = MagicMock() + new_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=new_client): + manager.reconnect(settings) + + # Old client must have been stopped + old_client.disconnect.assert_called() + old_client.loop_stop.assert_called() + + # New client must have been connected + new_client.loop_start.assert_called_once() + new_client.connect.assert_called_once_with( + host="broker.test", port=1883, keepalive=60 + ) + + +# --------------------------------------------------------------------------- +# POST /api/config/mqtt/test — session + CSRF guards +# --------------------------------------------------------------------------- + +_MQTT_TEST_URL = "/api/config/mqtt/test" + + +def test_mqtt_test_unauthenticated_returns_401(client: TestClient) -> None: + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) + assert response.status_code == 401 + + +def test_mqtt_test_authenticated_missing_csrf_returns_403(client: TestClient) -> None: + _login(client) + response = client.post(_MQTT_TEST_URL) + assert response.status_code == 403 + + +def test_mqtt_test_authenticated_empty_csrf_returns_403(client: TestClient) -> None: + _login(client) + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": ""}) + assert response.status_code == 403 + + +# --------------------------------------------------------------------------- +# POST /api/config/mqtt/test — three-state responses +# --------------------------------------------------------------------------- + +_RUN_MQTT_TEST_PATH = "app.api.routes.api.config._run_mqtt_test" + + +def test_mqtt_test_success_returns_200(client: TestClient) -> None: + _login(client) + with patch(_RUN_MQTT_TEST_PATH, return_value=None) as mock_run: + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) + + mock_run.assert_called_once() + assert response.status_code == 200 + body = response.json() + assert body["result"] == "success" + assert "message" in body + + +def test_mqtt_test_config_error_returns_400(client: TestClient) -> None: + _login(client) + from app.api.routes.api.config import _MqttConfigurationError + + with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConfigurationError("Host not configured")): + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) + + assert response.status_code == 400 + body = response.json() + assert body["result"] == "config-error" + assert "Host not configured" in body["message"] + + +def test_mqtt_test_connection_error_returns_502(client: TestClient) -> None: + _login(client) + from app.api.routes.api.config import _MqttConnectionError + + with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConnectionError("connection refused")): + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) + + assert response.status_code == 502 + body = response.json() + assert body["result"] == "failed" + assert "connection refused" in body["message"] + + +def test_mqtt_test_response_does_not_echo_password(client: TestClient) -> None: + """MQTT password must not appear in any test response body.""" + _login(client) + from app.api.routes.api.config import _MqttConnectionError + + with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConnectionError("error: [redacted]")): + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "token"}) + + assert response.status_code == 502 + # Actual password "s3cr3t" must not appear + assert "s3cr3t" not in response.text + + +def test_mqtt_test_empty_host_config_error(client: TestClient) -> None: + """When broker host is empty, endpoint should return 400 config-error.""" + _login(client) + + # The real _run_mqtt_test raises _MqttConfigurationError for empty host. + # We don't mock it here — we let the real implementation run. + # The default settings have mqtt_broker_host="" so this should 400. + response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) + assert response.status_code == 400 + body = response.json() + assert body["result"] == "config-error" + + +# --------------------------------------------------------------------------- +# _run_mqtt_test unit tests (lower-level, mock paho entirely) +# --------------------------------------------------------------------------- + +def test_run_mqtt_test_raises_config_error_when_host_empty() -> None: + from app.api.routes.api.config import _run_mqtt_test, _MqttConfigurationError + + settings = _make_settings(mqtt_broker_host="") + with pytest.raises(_MqttConfigurationError, match="host"): + _run_mqtt_test(settings) + + +def test_run_mqtt_test_raises_connection_error_on_os_error() -> None: + from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError + + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.side_effect = OSError("Connection refused") + + # _run_mqtt_test imports paho.mqtt.client locally inside the function body, + # so we patch via the real module path (paho.mqtt.client.Client). + with patch("paho.mqtt.client.Client", return_value=mock_client): + with pytest.raises(_MqttConnectionError, match="Cannot reach broker"): + _run_mqtt_test(settings) + + # loop_stop must be called in the finally block + mock_client.loop_stop.assert_called() + + +def test_run_mqtt_test_raises_connection_error_on_timeout() -> None: + """If connected_event never fires, should raise _MqttConnectionError.""" + from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError + + settings = _make_settings() + + mock_client = MagicMock() + mock_client.connect.return_value = None + + class _NeverSetEvent: + def wait(self, timeout=None): + return False # simulate timeout + + def set(self): + pass + + def _make_event(): + return _NeverSetEvent() + + with patch("paho.mqtt.client.Client", return_value=mock_client): + with patch("threading.Event", side_effect=_make_event): + with pytest.raises(_MqttConnectionError, match="timed out"): + _run_mqtt_test(settings)