[mqtt] Fix crash with empty broker during upload/logs (#11866)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

esphome/mqtt.py

@@ -30,6 +30,7 @@ from esphome.const import (
 from esphome.core import CORE, EsphomeError
 from esphome.helpers import get_int_env, get_str_env
 from esphome.log import AnsiFore, color
+from esphome.types import ConfigType
 from esphome.util import safe_print
 
 _LOGGER = logging.getLogger(__name__)
@@ -154,8 +155,12 @@ def show_discover(config, username=None, password=None, client_id=None):
 
 
 def get_esphome_device_ip(
-    config, username=None, password=None, client_id=None, timeout=25
+    config: ConfigType,
-):
+    username: str | None = None,
+    password: str | None = None,
+    client_id: str | None = None,
+    timeout: int | float = 25,
+) -> list[str]:
     if CONF_MQTT not in config:
         raise EsphomeError(
             "Cannot discover IP via MQTT as the config does not include the mqtt: "
@@ -166,6 +171,10 @@ def get_esphome_device_ip(
             "Cannot discover IP via MQTT as the config does not include the device name: "
             "component"
         )
+    if not config[CONF_MQTT].get(CONF_BROKER):
+        raise EsphomeError(
+            "Cannot discover IP via MQTT as the broker is not configured"
+        )
 
     dev_name = config[CONF_ESPHOME][CONF_NAME]
     dev_ip = None

tests/unit_tests/test_main.py

@@ -1166,6 +1166,56 @@ def test_upload_program_ota_with_mqtt_resolution(
     )
 
 
+def test_upload_program_ota_with_mqtt_empty_broker(
+    mock_mqtt_get_ip: Mock,
+    mock_is_ip_address: Mock,
+    mock_run_ota: Mock,
+    tmp_path: Path,
+    caplog: CaptureFixture,
+) -> None:
+    """Test upload_program with OTA when MQTT broker is empty (issue #11653)."""
+    setup_core(address="192.168.1.50", platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_is_ip_address.return_value = True
+    mock_mqtt_get_ip.side_effect = EsphomeError(
+        "Cannot discover IP via MQTT as the broker is not configured"
+    )
+    mock_run_ota.return_value = (0, "192.168.1.50")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ],
+        CONF_MQTT: {
+            CONF_BROKER: "",
+        },
+        CONF_MDNS: {
+            CONF_DISABLED: True,
+        },
+    }
+    args = MockArgs(username="user", password="pass", client_id="client")
+    devices = ["MQTTIP", "192.168.1.50"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.50"
+    # Verify MQTT was attempted but failed gracefully
+    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    # Verify we fell back to the IP address
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.50"], 3232, None, expected_firmware
+    )
+    # Verify warning was logged
+    assert "MQTT IP discovery failed" in caplog.text
+
+
 @patch("esphome.__main__.importlib.import_module")
 def test_upload_program_platform_specific_handler(
     mock_import: Mock,

tests/unit_tests/test_mqtt.py

@@ -0,0 +1,91 @@
+"""Unit tests for esphome.mqtt module."""
+
+from __future__ import annotations
+
+import pytest
+
+from esphome.const import CONF_BROKER, CONF_ESPHOME, CONF_MQTT, CONF_NAME
+from esphome.core import EsphomeError
+from esphome.mqtt import get_esphome_device_ip
+
+
+def test_get_esphome_device_ip_empty_broker() -> None:
+    """Test that get_esphome_device_ip raises EsphomeError when broker is empty."""
+    config = {
+        CONF_MQTT: {
+            CONF_BROKER: "",
+        },
+        CONF_ESPHOME: {
+            CONF_NAME: "test-device",
+        },
+    }
+
+    with pytest.raises(
+        EsphomeError,
+        match="Cannot discover IP via MQTT as the broker is not configured",
+    ):
+        get_esphome_device_ip(config)
+
+
+def test_get_esphome_device_ip_none_broker() -> None:
+    """Test that get_esphome_device_ip raises EsphomeError when broker is None."""
+    config = {
+        CONF_MQTT: {
+            CONF_BROKER: None,
+        },
+        CONF_ESPHOME: {
+            CONF_NAME: "test-device",
+        },
+    }
+
+    with pytest.raises(
+        EsphomeError,
+        match="Cannot discover IP via MQTT as the broker is not configured",
+    ):
+        get_esphome_device_ip(config)
+
+
+def test_get_esphome_device_ip_missing_mqtt() -> None:
+    """Test that get_esphome_device_ip raises EsphomeError when mqtt config is missing."""
+    config = {
+        CONF_ESPHOME: {
+            CONF_NAME: "test-device",
+        },
+    }
+
+    with pytest.raises(
+        EsphomeError,
+        match="Cannot discover IP via MQTT as the config does not include the mqtt:",
+    ):
+        get_esphome_device_ip(config)
+
+
+def test_get_esphome_device_ip_missing_esphome() -> None:
+    """Test that get_esphome_device_ip raises EsphomeError when esphome config is missing."""
+    config = {
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+    }
+
+    with pytest.raises(
+        EsphomeError,
+        match="Cannot discover IP via MQTT as the config does not include the device name:",
+    ):
+        get_esphome_device_ip(config)
+
+
+def test_get_esphome_device_ip_missing_name() -> None:
+    """Test that get_esphome_device_ip raises EsphomeError when device name is missing."""
+    config = {
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+        CONF_ESPHOME: {},
+    }
+
+    with pytest.raises(
+        EsphomeError,
+        match="Cannot discover IP via MQTT as the config does not include the device name:",
+    ):
+        get_esphome_device_ip(config)