[core] fix upload to device via MQTT IP lookup (e.g. when mDNS is disable) (#10632)

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
Co-authored-by: J. Nick Koston <nick+github@koston.org>

esphome/__main__.py

@@ -15,9 +15,11 @@ import argcomplete
 
 from esphome import const, writer, yaml_util
 import esphome.codegen as cg
+from esphome.components.mqtt import CONF_DISCOVER_IP
 from esphome.config import iter_component_configs, read_config, strip_default_ids
 from esphome.const import (
     ALLOWED_NAME_CHARS,
+    CONF_API,
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DEASSERT_RTS_DTR,
@@ -43,6 +45,7 @@ from esphome.const import (
     SECRETS_FILES,
 )
 from esphome.core import CORE, EsphomeError, coroutine
+from esphome.enum import StrEnum
 from esphome.helpers import get_bool_env, indent, is_ip_address
 from esphome.log import AnsiFore, color, setup_log
 from esphome.types import ConfigType
@@ -106,13 +109,15 @@ def choose_prompt(options, purpose: str = None):
     return options[opt - 1][1]
 
 
+class Purpose(StrEnum):
+    UPLOADING = "uploading"
+    LOGGING = "logging"
+
+
 def choose_upload_log_host(
     default: list[str] | str | None,
     check_default: str | None,
-    show_ota: bool,
+    purpose: Purpose,
-    show_mqtt: bool,
-    show_api: bool,
-    purpose: str | None = None,
 ) -> list[str]:
     # Convert to list for uniform handling
     defaults = [default] if isinstance(default, str) else default or []
@@ -132,13 +137,30 @@ def choose_upload_log_host(
                 ]
                 resolved.append(choose_prompt(options, purpose=purpose))
             elif device == "OTA":
-                if CORE.address and (
+                # ensure IP adresses are used first
-                    (show_ota and "ota" in CORE.config)
+                if is_ip_address(CORE.address) and (
-                    or (show_api and "api" in CORE.config)
+                    (purpose == Purpose.LOGGING and has_api())
+                    or (purpose == Purpose.UPLOADING and has_ota())
                 ):
                     resolved.append(CORE.address)
-                elif show_mqtt and has_mqtt_logging():
+
-                    resolved.append("MQTT")
+                if purpose == Purpose.LOGGING:
+                    if has_api() and has_mqtt_ip_lookup():
+                        resolved.append("MQTTIP")
+
+                    if has_mqtt_logging():
+                        resolved.append("MQTT")
+
+                    if has_api() and has_non_ip_address():
+                        resolved.append(CORE.address)
+
+                elif purpose == Purpose.UPLOADING:
+                    if has_ota() and has_mqtt_ip_lookup():
+                        resolved.append("MQTTIP")
+
+                    if has_ota() and has_non_ip_address():
+                        resolved.append(CORE.address)
+
             else:
                 resolved.append(device)
         if not resolved:
@@ -149,39 +171,111 @@ def choose_upload_log_host(
     options = [
         (f"{port.path} ({port.description})", port.path) for port in get_serial_ports()
     ]
-    if (show_ota and "ota" in CORE.config) or (show_api and "api" in CORE.config):
+
-        options.append((f"Over The Air ({CORE.address})", CORE.address))
+    if purpose == Purpose.LOGGING:
-    if show_mqtt and has_mqtt_logging():
+        if has_mqtt_logging():
-        mqtt_config = CORE.config[CONF_MQTT]
+            mqtt_config = CORE.config[CONF_MQTT]
-        options.append((f"MQTT ({mqtt_config[CONF_BROKER]})", "MQTT"))
+            options.append((f"MQTT ({mqtt_config[CONF_BROKER]})", "MQTT"))
+
+        if has_api():
+            if has_resolvable_address():
+                options.append((f"Over The Air ({CORE.address})", CORE.address))
+            if has_mqtt_ip_lookup():
+                options.append(("Over The Air (MQTT IP lookup)", "MQTTIP"))
+
+    elif purpose == Purpose.UPLOADING and has_ota():
+        if has_resolvable_address():
+            options.append((f"Over The Air ({CORE.address})", CORE.address))
+        if has_mqtt_ip_lookup():
+            options.append(("Over The Air (MQTT IP lookup)", "MQTTIP"))
 
     if check_default is not None and check_default in [opt[1] for opt in options]:
         return [check_default]
     return [choose_prompt(options, purpose=purpose)]
 
 
-def mqtt_logging_enabled(mqtt_config):
+def has_mqtt_logging() -> bool:
+    """Check if MQTT logging is available."""
+    if CONF_MQTT not in CORE.config:
+        return False
+
+    mqtt_config = CORE.config[CONF_MQTT]
+
+    # enabled by default
+    if CONF_LOG_TOPIC not in mqtt_config:
+        return True
+
     log_topic = mqtt_config[CONF_LOG_TOPIC]
     if log_topic is None:
         return False
+
     if CONF_TOPIC not in log_topic:
         return False
-    return log_topic.get(CONF_LEVEL, None) != "NONE"
+
+    return log_topic[CONF_LEVEL] != "NONE"
 
 
-def has_mqtt_logging() -> bool:
+def has_mqtt() -> bool:
-    """Check if MQTT logging is available."""
+    """Check if MQTT is available."""
-    return (mqtt_config := CORE.config.get(CONF_MQTT)) and mqtt_logging_enabled(
+    return CONF_MQTT in CORE.config
-        mqtt_config
+
-    )
+
+def has_api() -> bool:
+    """Check if API is available."""
+    return CONF_API in CORE.config
+
+
+def has_ota() -> bool:
+    """Check if OTA is available."""
+    return CONF_OTA in CORE.config
+
+
+def has_mqtt_ip_lookup() -> bool:
+    """Check if MQTT is available and IP lookup is supported."""
+    if CONF_MQTT not in CORE.config:
+        return False
+    # Default Enabled
+    if CONF_DISCOVER_IP not in CORE.config[CONF_MQTT]:
+        return True
+    return CORE.config[CONF_MQTT][CONF_DISCOVER_IP]
+
+
+def has_mdns() -> bool:
+    """Check if MDNS is available."""
+    return CONF_MDNS not in CORE.config or not CORE.config[CONF_MDNS][CONF_DISABLED]
+
+
+def has_non_ip_address() -> bool:
+    """Check if CORE.address is set and is not an IP address."""
+    return CORE.address is not None and not is_ip_address(CORE.address)
+
+
+def has_ip_address() -> bool:
+    """Check if CORE.address is a valid IP address."""
+    return CORE.address is not None and is_ip_address(CORE.address)
+
+
+def has_resolvable_address() -> bool:
+    """Check if CORE.address is resolvable (via mDNS or is an IP address)."""
+    return has_mdns() or has_ip_address()
+
+
+def mqtt_get_ip(config: ConfigType, username: str, password: str, client_id: str):
+    from esphome import mqtt
+
+    return mqtt.get_esphome_device_ip(config, username, password, client_id)
+
+
+_PORT_TO_PORT_TYPE = {
+    "MQTT": "MQTT",
+    "MQTTIP": "MQTTIP",
+}
 
 
 def get_port_type(port: str) -> str:
     if port.startswith("/") or port.startswith("COM"):
         return "SERIAL"
-    if port == "MQTT":
+    return _PORT_TO_PORT_TYPE.get(port, "NETWORK")
-        return "MQTT"
-    return "NETWORK"
 
 
 def run_miniterm(config: ConfigType, port: str, args) -> int:
@@ -439,23 +533,9 @@ def upload_program(
     password = ota_conf.get(CONF_PASSWORD, "")
     binary = args.file if getattr(args, "file", None) is not None else CORE.firmware_bin
 
-    # Check if we should use MQTT for address resolution
+    # MQTT address resolution
-    # This happens when no device was specified, or the current host is "MQTT"/"OTA"
+    if get_port_type(host) in ("MQTT", "MQTTIP"):
-    if (
+        devices = mqtt_get_ip(config, args.username, args.password, args.client_id)
-        CONF_MQTT in config  # pylint: disable=too-many-boolean-expressions
-        and (not devices or host in ("MQTT", "OTA"))
-        and (
-            ((config[CONF_MDNS][CONF_DISABLED]) and not is_ip_address(CORE.address))
-            or get_port_type(host) == "MQTT"
-        )
-    ):
-        from esphome import mqtt
-
-        devices = [
-            mqtt.get_esphome_device_ip(
-                config, args.username, args.password, args.client_id
-            )
-        ]
 
     return espota2.run_ota(devices, remote_port, password, binary)
 
@@ -476,20 +556,28 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
     if get_port_type(port) == "SERIAL":
         check_permissions(port)
         return run_miniterm(config, port, args)
-    if get_port_type(port) == "NETWORK" and "api" in config:
-        addresses_to_use = devices
-        if config[CONF_MDNS][CONF_DISABLED] and CONF_MQTT in config:
-            from esphome import mqtt
 
-            mqtt_address = mqtt.get_esphome_device_ip(
+    port_type = get_port_type(port)
+
+    # Check if we should use API for logging
+    if has_api():
+        addresses_to_use: list[str] | None = None
+
+        if port_type == "NETWORK" and (has_mdns() or is_ip_address(port)):
+            addresses_to_use = devices
+        elif port_type in ("NETWORK", "MQTT", "MQTTIP") and has_mqtt_ip_lookup():
+            # Only use MQTT IP lookup if the first condition didn't match
+            # (for MQTT/MQTTIP types, or for NETWORK when mdns/ip check fails)
+            addresses_to_use = mqtt_get_ip(
                 config, args.username, args.password, args.client_id
-            )[0]
+            )
-            addresses_to_use = [mqtt_address]
 
-        from esphome.components.api.client import run_logs
+        if addresses_to_use is not None:
+            from esphome.components.api.client import run_logs
 
-        return run_logs(config, addresses_to_use)
+            return run_logs(config, addresses_to_use)
-    if get_port_type(port) in ("NETWORK", "MQTT") and "mqtt" in config:
+
+    if port_type in ("NETWORK", "MQTT") and has_mqtt_logging():
         from esphome import mqtt
 
         return mqtt.show_logs(
@@ -555,10 +643,7 @@ def command_upload(args: ArgsProtocol, config: ConfigType) -> int | None:
     devices = choose_upload_log_host(
         default=args.device,
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
-        purpose="uploading",
     )
 
     exit_code, _ = upload_program(config, args, devices)
@@ -583,10 +668,7 @@ def command_logs(args: ArgsProtocol, config: ConfigType) -> int | None:
     devices = choose_upload_log_host(
         default=args.device,
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.LOGGING,
-        show_mqtt=True,
-        show_api=True,
-        purpose="logging",
     )
     return show_logs(config, args, devices)
 
@@ -612,10 +694,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
     devices = choose_upload_log_host(
         default=args.device,
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=True,
-        purpose="uploading",
     )
 
     exit_code, successful_device = upload_program(config, args, devices)
@@ -632,10 +711,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
     devices = choose_upload_log_host(
         default=successful_device,
         check_default=successful_device,
-        show_ota=False,
+        purpose=Purpose.LOGGING,
-        show_mqtt=True,
-        show_api=True,
-        purpose="logging",
     )
     return show_logs(config, args, devices)
 

esphome/const.py

@@ -114,6 +114,7 @@ CONF_AND = "and"
 CONF_ANGLE = "angle"
 CONF_ANY = "any"
 CONF_AP = "ap"
+CONF_API = "api"
 CONF_APPARENT_POWER = "apparent_power"
 CONF_ARDUINO_VERSION = "arduino_version"
 CONF_AREA = "area"

tests/unit_tests/test_main.py

@@ -12,16 +12,28 @@ import pytest
 from pytest import CaptureFixture
 
 from esphome.__main__ import (
+    Purpose,
     choose_upload_log_host,
     command_rename,
     command_wizard,
+    get_port_type,
+    has_ip_address,
+    has_mqtt,
+    has_mqtt_ip_lookup,
+    has_mqtt_logging,
+    has_non_ip_address,
+    has_resolvable_address,
+    mqtt_get_ip,
     show_logs,
     upload_program,
 )
 from esphome.const import (
+    CONF_API,
     CONF_BROKER,
     CONF_DISABLED,
     CONF_ESPHOME,
+    CONF_LEVEL,
+    CONF_LOG_TOPIC,
     CONF_MDNS,
     CONF_MQTT,
     CONF_NAME,
@@ -30,6 +42,7 @@ from esphome.const import (
     CONF_PLATFORM,
     CONF_PORT,
     CONF_SUBSTITUTIONS,
+    CONF_TOPIC,
     CONF_USE_ADDRESS,
     CONF_WIFI,
     KEY_CORE,
@@ -147,6 +160,13 @@ def mock_is_ip_address() -> Generator[Mock]:
         yield mock
 
 
+@pytest.fixture
+def mock_mqtt_get_ip() -> Generator[Mock]:
+    """Mock mqtt_get_ip for testing."""
+    with patch("esphome.__main__.mqtt_get_ip") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_serial_ports() -> Generator[Mock]:
     """Mock get_serial_ports to return test ports."""
@@ -189,62 +209,56 @@ def mock_run_external_process() -> Generator[Mock]:
 
 def test_choose_upload_log_host_with_string_default() -> None:
     """Test with a single string default device."""
+    setup_core()
     result = choose_upload_log_host(
         default="192.168.1.100",
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.100"]
 
 
 def test_choose_upload_log_host_with_list_default() -> None:
     """Test with a list of default devices."""
+    setup_core()
     result = choose_upload_log_host(
         default=["192.168.1.100", "192.168.1.101"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.100", "192.168.1.101"]
 
 
 def test_choose_upload_log_host_with_multiple_ip_addresses() -> None:
     """Test with multiple IP addresses as defaults."""
+    setup_core()
     result = choose_upload_log_host(
         default=["1.2.3.4", "4.5.5.6"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.LOGGING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["1.2.3.4", "4.5.5.6"]
 
 
 def test_choose_upload_log_host_with_mixed_hostnames_and_ips() -> None:
     """Test with a mix of hostnames and IP addresses."""
+    setup_core()
     result = choose_upload_log_host(
         default=["host.one", "host.one.local", "1.2.3.4"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["host.one", "host.one.local", "1.2.3.4"]
 
 
 def test_choose_upload_log_host_with_ota_list() -> None:
     """Test with OTA as the only item in the list."""
-    setup_core(config={"ota": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}}, address="192.168.1.100")
 
     result = choose_upload_log_host(
         default=["OTA"],
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.100"]
 
@@ -252,16 +266,27 @@ def test_choose_upload_log_host_with_ota_list() -> None:
 @pytest.mark.usefixtures("mock_has_mqtt_logging")
 def test_choose_upload_log_host_with_ota_list_mqtt_fallback() -> None:
     """Test with OTA list falling back to MQTT when no address."""
-    setup_core()
+    setup_core(config={CONF_OTA: {}, "mqtt": {}})
 
     result = choose_upload_log_host(
         default=["OTA"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=True,
-        show_api=False,
     )
-    assert result == ["MQTT"]
+    assert result == ["MQTTIP"]
+
+
+@pytest.mark.usefixtures("mock_has_mqtt_logging")
+def test_choose_upload_log_host_with_ota_list_mqtt_fallback_logging() -> None:
+    """Test with OTA list with API and MQTT when no address."""
+    setup_core(config={CONF_API: {}, "mqtt": {}})
+
+    result = choose_upload_log_host(
+        default=["OTA"],
+        check_default=None,
+        purpose=Purpose.LOGGING,
+    )
+    assert result == ["MQTTIP", "MQTT"]
 
 
 @pytest.mark.usefixtures("mock_no_serial_ports")
@@ -269,12 +294,11 @@ def test_choose_upload_log_host_with_serial_device_no_ports(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test SERIAL device when no serial ports are found."""
+    setup_core()
     result = choose_upload_log_host(
         default="SERIAL",
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == []
     assert "No serial ports found, skipping SERIAL device" in caplog.text
@@ -285,13 +309,11 @@ def test_choose_upload_log_host_with_serial_device_with_ports(
     mock_choose_prompt: Mock,
 ) -> None:
     """Test SERIAL device when serial ports are available."""
+    setup_core()
     result = choose_upload_log_host(
         default="SERIAL",
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
-        purpose="testing",
     )
     assert result == ["/dev/ttyUSB0"]
     mock_choose_prompt.assert_called_once_with(
@@ -299,34 +321,42 @@ def test_choose_upload_log_host_with_serial_device_with_ports(
             ("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0"),
             ("/dev/ttyUSB1 (Another USB Serial)", "/dev/ttyUSB1"),
         ],
-        purpose="testing",
+        purpose=Purpose.UPLOADING,
     )
 
 
 def test_choose_upload_log_host_with_ota_device_with_ota_config() -> None:
     """Test OTA device when OTA is configured."""
-    setup_core(config={"ota": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}}, address="192.168.1.100")
 
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.100"]
 
 
 def test_choose_upload_log_host_with_ota_device_with_api_config() -> None:
-    """Test OTA device when API is configured."""
+    """Test OTA device when API is configured (no upload without OTA in config)."""
-    setup_core(config={"api": {}}, address="192.168.1.100")
+    setup_core(config={CONF_API: {}}, address="192.168.1.100")
 
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
+    )
-        show_api=True,
+    assert result == []
+
+
+def test_choose_upload_log_host_with_ota_device_with_api_config_logging() -> None:
+    """Test OTA device when API is configured."""
+    setup_core(config={CONF_API: {}}, address="192.168.1.100")
+
+    result = choose_upload_log_host(
+        default="OTA",
+        check_default=None,
+        purpose=Purpose.LOGGING,
     )
     assert result == ["192.168.1.100"]
 
@@ -334,14 +364,12 @@ def test_choose_upload_log_host_with_ota_device_with_api_config() -> None:
 @pytest.mark.usefixtures("mock_has_mqtt_logging")
 def test_choose_upload_log_host_with_ota_device_fallback_to_mqtt() -> None:
     """Test OTA device fallback to MQTT when no OTA/API config."""
-    setup_core()
+    setup_core(config={"mqtt": {}})
 
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.LOGGING,
-        show_mqtt=True,
-        show_api=False,
     )
     assert result == ["MQTT"]
 
@@ -354,9 +382,7 @@ def test_choose_upload_log_host_with_ota_device_no_fallback() -> None:
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=True,
-        show_api=False,
     )
     assert result == []
 
@@ -364,7 +390,7 @@ def test_choose_upload_log_host_with_ota_device_no_fallback() -> None:
 @pytest.mark.usefixtures("mock_choose_prompt")
 def test_choose_upload_log_host_multiple_devices() -> None:
     """Test with multiple devices including special identifiers."""
-    setup_core(config={"ota": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}}, address="192.168.1.100")
 
     mock_ports = [MockSerialPort("/dev/ttyUSB0", "USB Serial")]
 
@@ -372,9 +398,7 @@ def test_choose_upload_log_host_multiple_devices() -> None:
         result = choose_upload_log_host(
             default=["192.168.1.50", "OTA", "SERIAL"],
             check_default=None,
-            show_ota=True,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=False,
-            show_api=False,
         )
         assert result == ["192.168.1.50", "192.168.1.100", "/dev/ttyUSB0"]
 
@@ -393,22 +417,19 @@ def test_choose_upload_log_host_no_defaults_with_serial_ports(
         result = choose_upload_log_host(
             default=None,
             check_default=None,
-            show_ota=False,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=False,
-            show_api=False,
-            purpose="uploading",
         )
         assert result == ["/dev/ttyUSB0"]
         mock_choose_prompt.assert_called_once_with(
             [("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0")],
-            purpose="uploading",
+            purpose=Purpose.UPLOADING,
         )
 
 
 @pytest.mark.usefixtures("mock_no_serial_ports")
 def test_choose_upload_log_host_no_defaults_with_ota() -> None:
     """Test interactive mode with OTA option."""
-    setup_core(config={"ota": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}}, address="192.168.1.100")
 
     with patch(
         "esphome.__main__.choose_prompt", return_value="192.168.1.100"
@@ -416,21 +437,19 @@ def test_choose_upload_log_host_no_defaults_with_ota() -> None:
         result = choose_upload_log_host(
             default=None,
             check_default=None,
-            show_ota=True,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=False,
-            show_api=False,
         )
         assert result == ["192.168.1.100"]
         mock_prompt.assert_called_once_with(
             [("Over The Air (192.168.1.100)", "192.168.1.100")],
-            purpose=None,
+            purpose=Purpose.UPLOADING,
         )
 
 
 @pytest.mark.usefixtures("mock_no_serial_ports")
 def test_choose_upload_log_host_no_defaults_with_api() -> None:
     """Test interactive mode with API option."""
-    setup_core(config={"api": {}}, address="192.168.1.100")
+    setup_core(config={CONF_API: {}}, address="192.168.1.100")
 
     with patch(
         "esphome.__main__.choose_prompt", return_value="192.168.1.100"
@@ -438,14 +457,12 @@ def test_choose_upload_log_host_no_defaults_with_api() -> None:
         result = choose_upload_log_host(
             default=None,
             check_default=None,
-            show_ota=False,
+            purpose=Purpose.LOGGING,
-            show_mqtt=False,
-            show_api=True,
         )
         assert result == ["192.168.1.100"]
         mock_prompt.assert_called_once_with(
             [("Over The Air (192.168.1.100)", "192.168.1.100")],
-            purpose=None,
+            purpose=Purpose.LOGGING,
         )
 
 
@@ -458,14 +475,12 @@ def test_choose_upload_log_host_no_defaults_with_mqtt() -> None:
         result = choose_upload_log_host(
             default=None,
             check_default=None,
-            show_ota=False,
+            purpose=Purpose.LOGGING,
-            show_mqtt=True,
-            show_api=False,
         )
         assert result == ["MQTT"]
         mock_prompt.assert_called_once_with(
             [("MQTT (mqtt.local)", "MQTT")],
-            purpose=None,
+            purpose=Purpose.LOGGING,
         )
 
 
@@ -475,7 +490,7 @@ def test_choose_upload_log_host_no_defaults_with_all_options(
 ) -> None:
     """Test interactive mode with all options available."""
     setup_core(
-        config={"ota": {}, "api": {}, CONF_MQTT: {CONF_BROKER: "mqtt.local"}},
+        config={CONF_OTA: {}, CONF_API: {}, CONF_MQTT: {CONF_BROKER: "mqtt.local"}},
         address="192.168.1.100",
     )
 
@@ -485,32 +500,59 @@ def test_choose_upload_log_host_no_defaults_with_all_options(
         result = choose_upload_log_host(
             default=None,
             check_default=None,
-            show_ota=True,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=True,
-            show_api=True,
-            purpose="testing",
         )
         assert result == ["/dev/ttyUSB0"]
 
         expected_options = [
             ("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0"),
             ("Over The Air (192.168.1.100)", "192.168.1.100"),
-            ("MQTT (mqtt.local)", "MQTT"),
+            ("Over The Air (MQTT IP lookup)", "MQTTIP"),
         ]
-        mock_choose_prompt.assert_called_once_with(expected_options, purpose="testing")
+        mock_choose_prompt.assert_called_once_with(
+            expected_options, purpose=Purpose.UPLOADING
+        )
+
+
+def test_choose_upload_log_host_no_defaults_with_all_options_logging(
+    mock_choose_prompt: Mock,
+) -> None:
+    """Test interactive mode with all options available."""
+    setup_core(
+        config={CONF_OTA: {}, CONF_API: {}, CONF_MQTT: {CONF_BROKER: "mqtt.local"}},
+        address="192.168.1.100",
+    )
+
+    mock_ports = [MockSerialPort("/dev/ttyUSB0", "USB Serial")]
+
+    with patch("esphome.__main__.get_serial_ports", return_value=mock_ports):
+        result = choose_upload_log_host(
+            default=None,
+            check_default=None,
+            purpose=Purpose.LOGGING,
+        )
+        assert result == ["/dev/ttyUSB0"]
+
+        expected_options = [
+            ("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0"),
+            ("MQTT (mqtt.local)", "MQTT"),
+            ("Over The Air (192.168.1.100)", "192.168.1.100"),
+            ("Over The Air (MQTT IP lookup)", "MQTTIP"),
+        ]
+        mock_choose_prompt.assert_called_once_with(
+            expected_options, purpose=Purpose.LOGGING
+        )
 
 
 @pytest.mark.usefixtures("mock_no_serial_ports")
 def test_choose_upload_log_host_check_default_matches() -> None:
     """Test when check_default matches an available option."""
-    setup_core(config={"ota": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}}, address="192.168.1.100")
 
     result = choose_upload_log_host(
         default=None,
         check_default="192.168.1.100",
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.100"]
 
@@ -526,9 +568,7 @@ def test_choose_upload_log_host_check_default_no_match() -> None:
         result = choose_upload_log_host(
             default=None,
             check_default="192.168.1.100",
-            show_ota=False,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=False,
-            show_api=False,
         )
         assert result == ["fallback"]
         mock_prompt.assert_called_once()
@@ -537,13 +577,12 @@ def test_choose_upload_log_host_check_default_no_match() -> None:
 @pytest.mark.usefixtures("mock_no_serial_ports")
 def test_choose_upload_log_host_empty_defaults_list() -> None:
     """Test with an empty list as default."""
+    setup_core()
     with patch("esphome.__main__.choose_prompt", return_value="chosen") as mock_prompt:
         result = choose_upload_log_host(
             default=[],
             check_default=None,
-            show_ota=False,
+            purpose=Purpose.UPLOADING,
-            show_mqtt=False,
-            show_api=False,
         )
         assert result == ["chosen"]
         mock_prompt.assert_called_once()
@@ -559,9 +598,7 @@ def test_choose_upload_log_host_all_devices_unresolved(
     result = choose_upload_log_host(
         default=["SERIAL", "OTA"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == []
     assert (
@@ -577,38 +614,132 @@ def test_choose_upload_log_host_mixed_resolved_unresolved() -> None:
     result = choose_upload_log_host(
         default=["192.168.1.50", "SERIAL", "OTA"],
         check_default=None,
-        show_ota=False,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == ["192.168.1.50"]
 
 
 def test_choose_upload_log_host_ota_both_conditions() -> None:
     """Test OTA device when both OTA and API are configured and enabled."""
-    setup_core(config={"ota": {}, "api": {}}, address="192.168.1.100")
+    setup_core(config={CONF_OTA: {}, CONF_API: {}}, address="192.168.1.100")
 
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=True,
     )
     assert result == ["192.168.1.100"]
 
 
+@pytest.mark.usefixtures("mock_serial_ports")
+def test_choose_upload_log_host_ota_ip_all_options() -> None:
+    """Test OTA device when both static IP, OTA, API and MQTT are configured and enabled but MDNS not."""
+    setup_core(
+        config={
+            CONF_OTA: {},
+            CONF_API: {},
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+            },
+            CONF_MDNS: {
+                CONF_DISABLED: True,
+            },
+        },
+        address="192.168.1.100",
+    )
+
+    result = choose_upload_log_host(
+        default="OTA",
+        check_default=None,
+        purpose=Purpose.UPLOADING,
+    )
+    assert result == ["192.168.1.100", "MQTTIP"]
+
+
+@pytest.mark.usefixtures("mock_serial_ports")
+def test_choose_upload_log_host_ota_local_all_options() -> None:
+    """Test OTA device when both static IP, OTA, API and MQTT are configured and enabled but MDNS not."""
+    setup_core(
+        config={
+            CONF_OTA: {},
+            CONF_API: {},
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+            },
+            CONF_MDNS: {
+                CONF_DISABLED: True,
+            },
+        },
+        address="test.local",
+    )
+
+    result = choose_upload_log_host(
+        default="OTA",
+        check_default=None,
+        purpose=Purpose.UPLOADING,
+    )
+    assert result == ["MQTTIP", "test.local"]
+
+
+@pytest.mark.usefixtures("mock_serial_ports")
+def test_choose_upload_log_host_ota_ip_all_options_logging() -> None:
+    """Test OTA device when both static IP, OTA, API and MQTT are configured and enabled but MDNS not."""
+    setup_core(
+        config={
+            CONF_OTA: {},
+            CONF_API: {},
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+            },
+            CONF_MDNS: {
+                CONF_DISABLED: True,
+            },
+        },
+        address="192.168.1.100",
+    )
+
+    result = choose_upload_log_host(
+        default="OTA",
+        check_default=None,
+        purpose=Purpose.LOGGING,
+    )
+    assert result == ["192.168.1.100", "MQTTIP", "MQTT"]
+
+
+@pytest.mark.usefixtures("mock_serial_ports")
+def test_choose_upload_log_host_ota_local_all_options_logging() -> None:
+    """Test OTA device when both static IP, OTA, API and MQTT are configured and enabled but MDNS not."""
+    setup_core(
+        config={
+            CONF_OTA: {},
+            CONF_API: {},
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+            },
+            CONF_MDNS: {
+                CONF_DISABLED: True,
+            },
+        },
+        address="test.local",
+    )
+
+    result = choose_upload_log_host(
+        default="OTA",
+        check_default=None,
+        purpose=Purpose.LOGGING,
+    )
+    assert result == ["MQTTIP", "MQTT", "test.local"]
+
+
 @pytest.mark.usefixtures("mock_no_mqtt_logging")
 def test_choose_upload_log_host_no_address_with_ota_config() -> None:
     """Test OTA device when OTA is configured but no address is set."""
-    setup_core(config={"ota": {}})
+    setup_core(config={CONF_OTA: {}})
 
     result = choose_upload_log_host(
         default="OTA",
         check_default=None,
-        show_ota=True,
+        purpose=Purpose.UPLOADING,
-        show_mqtt=False,
-        show_api=False,
     )
     assert result == []
 
@@ -806,18 +937,15 @@ def test_upload_program_ota_no_config(
         upload_program(config, args, devices)
 
 
-@patch("esphome.mqtt.get_esphome_device_ip")
 def test_upload_program_ota_with_mqtt_resolution(
     mock_mqtt_get_ip: Mock,
     mock_is_ip_address: Mock,
     mock_run_ota: Mock,
-    mock_get_port_type: Mock,
     tmp_path: Path,
 ) -> None:
     """Test upload_program with OTA using MQTT for address resolution."""
     setup_core(address="device.local", platform=PLATFORM_ESP32, tmp_path=tmp_path)
 
-    mock_get_port_type.side_effect = ["MQTT", "NETWORK"]
     mock_is_ip_address.return_value = False
     mock_mqtt_get_ip.return_value = ["192.168.1.100"]
     mock_run_ota.return_value = (0, "192.168.1.100")
@@ -847,9 +975,7 @@ def test_upload_program_ota_with_mqtt_resolution(
     expected_firmware = str(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
-    mock_run_ota.assert_called_once_with(
+    mock_run_ota.assert_called_once_with(["192.168.1.100"], 3232, "", expected_firmware)
-        [["192.168.1.100"]], 3232, "", expected_firmware
-    )
 
 
 @patch("esphome.__main__.importlib.import_module")
@@ -910,18 +1036,16 @@ def test_show_logs_no_logger() -> None:
 @patch("esphome.components.api.client.run_logs")
 def test_show_logs_api(
     mock_run_logs: Mock,
-    mock_get_port_type: Mock,
 ) -> None:
     """Test show_logs with API."""
     setup_core(
         config={
             "logger": {},
-            "api": {},
+            CONF_API: {},
             CONF_MDNS: {CONF_DISABLED: False},
         },
         platform=PLATFORM_ESP32,
     )
-    mock_get_port_type.return_value = "NETWORK"
     mock_run_logs.return_value = 0
 
     args = MockArgs()
@@ -935,24 +1059,21 @@ def test_show_logs_api(
     )
 
 
-@patch("esphome.mqtt.get_esphome_device_ip")
 @patch("esphome.components.api.client.run_logs")
 def test_show_logs_api_with_mqtt_fallback(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
-    mock_get_port_type: Mock,
 ) -> None:
     """Test show_logs with API using MQTT for address resolution."""
     setup_core(
         config={
             "logger": {},
-            "api": {},
+            CONF_API: {},
             CONF_MDNS: {CONF_DISABLED: True},
             CONF_MQTT: {CONF_BROKER: "mqtt.local"},
         },
         platform=PLATFORM_ESP32,
     )
-    mock_get_port_type.return_value = "NETWORK"
     mock_run_logs.return_value = 0
     mock_mqtt_get_ip.return_value = ["192.168.1.200"]
 
@@ -969,7 +1090,6 @@ def test_show_logs_api_with_mqtt_fallback(
 @patch("esphome.mqtt.show_logs")
 def test_show_logs_mqtt(
     mock_mqtt_show_logs: Mock,
-    mock_get_port_type: Mock,
 ) -> None:
     """Test show_logs with MQTT."""
     setup_core(
@@ -979,7 +1099,6 @@ def test_show_logs_mqtt(
         },
         platform=PLATFORM_ESP32,
     )
-    mock_get_port_type.return_value = "MQTT"
     mock_mqtt_show_logs.return_value = 0
 
     args = MockArgs(
@@ -1001,7 +1120,6 @@ def test_show_logs_mqtt(
 @patch("esphome.mqtt.show_logs")
 def test_show_logs_network_with_mqtt_only(
     mock_mqtt_show_logs: Mock,
-    mock_get_port_type: Mock,
 ) -> None:
     """Test show_logs with network port but only MQTT configured."""
     setup_core(
@@ -1012,7 +1130,6 @@ def test_show_logs_network_with_mqtt_only(
         },
         platform=PLATFORM_ESP32,
     )
-    mock_get_port_type.return_value = "NETWORK"
     mock_mqtt_show_logs.return_value = 0
 
     args = MockArgs(
@@ -1031,9 +1148,7 @@ def test_show_logs_network_with_mqtt_only(
     )
 
 
-def test_show_logs_no_method_configured(
+def test_show_logs_no_method_configured() -> None:
-    mock_get_port_type: Mock,
-) -> None:
     """Test show_logs when no remote logging method is configured."""
     setup_core(
         config={
@@ -1042,7 +1157,6 @@ def test_show_logs_no_method_configured(
         },
         platform=PLATFORM_ESP32,
     )
-    mock_get_port_type.return_value = "NETWORK"
 
     args = MockArgs()
     devices = ["192.168.1.100"]
@@ -1075,6 +1189,175 @@ def test_show_logs_platform_specific_handler(
     mock_module.show_logs.assert_called_once_with(config, args, devices)
 
 
+def test_has_mqtt_logging_no_log_topic() -> None:
+    """Test has_mqtt_logging returns True when CONF_LOG_TOPIC is not in mqtt_config."""
+
+    # Setup MQTT config without CONF_LOG_TOPIC (defaults to enabled - this is the missing test case)
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local"}})
+    assert has_mqtt_logging() is True
+
+    # Setup MQTT config with CONF_LOG_TOPIC set to None (explicitly disabled)
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local", CONF_LOG_TOPIC: None}})
+    assert has_mqtt_logging() is False
+
+    # Setup MQTT config with CONF_LOG_TOPIC set with topic and level (explicitly enabled)
+    setup_core(
+        config={
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+                CONF_LOG_TOPIC: {CONF_TOPIC: "esphome/logs", CONF_LEVEL: "DEBUG"},
+            }
+        }
+    )
+    assert has_mqtt_logging() is True
+
+    # Setup MQTT config with CONF_LOG_TOPIC set but level is NONE (disabled)
+    setup_core(
+        config={
+            CONF_MQTT: {
+                CONF_BROKER: "mqtt.local",
+                CONF_LOG_TOPIC: {CONF_TOPIC: "esphome/logs", CONF_LEVEL: "NONE"},
+            }
+        }
+    )
+    assert has_mqtt_logging() is False
+
+    # Setup without MQTT config at all
+    setup_core(config={})
+    assert has_mqtt_logging() is False
+
+
+def test_has_mqtt() -> None:
+    """Test has_mqtt function."""
+
+    # Test with MQTT configured
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local"}})
+    assert has_mqtt() is True
+
+    # Test without MQTT configured
+    setup_core(config={})
+    assert has_mqtt() is False
+
+    # Test with other components but no MQTT
+    setup_core(config={CONF_API: {}, CONF_OTA: {}})
+    assert has_mqtt() is False
+
+
+def test_get_port_type() -> None:
+    """Test get_port_type function."""
+
+    assert get_port_type("/dev/ttyUSB0") == "SERIAL"
+    assert get_port_type("/dev/ttyACM0") == "SERIAL"
+    assert get_port_type("COM1") == "SERIAL"
+    assert get_port_type("COM10") == "SERIAL"
+
+    assert get_port_type("MQTT") == "MQTT"
+    assert get_port_type("MQTTIP") == "MQTTIP"
+
+    assert get_port_type("192.168.1.100") == "NETWORK"
+    assert get_port_type("esphome-device.local") == "NETWORK"
+    assert get_port_type("10.0.0.1") == "NETWORK"
+
+
+def test_has_mqtt_ip_lookup() -> None:
+    """Test has_mqtt_ip_lookup function."""
+
+    CONF_DISCOVER_IP = "discover_ip"
+
+    setup_core(config={})
+    assert has_mqtt_ip_lookup() is False
+
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local"}})
+    assert has_mqtt_ip_lookup() is True
+
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local", CONF_DISCOVER_IP: True}})
+    assert has_mqtt_ip_lookup() is True
+
+    setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local", CONF_DISCOVER_IP: False}})
+    assert has_mqtt_ip_lookup() is False
+
+
+def test_has_non_ip_address() -> None:
+    """Test has_non_ip_address function."""
+
+    setup_core(address=None)
+    assert has_non_ip_address() is False
+
+    setup_core(address="192.168.1.100")
+    assert has_non_ip_address() is False
+
+    setup_core(address="10.0.0.1")
+    assert has_non_ip_address() is False
+
+    setup_core(address="esphome-device.local")
+    assert has_non_ip_address() is True
+
+    setup_core(address="my-device")
+    assert has_non_ip_address() is True
+
+
+def test_has_ip_address() -> None:
+    """Test has_ip_address function."""
+
+    setup_core(address=None)
+    assert has_ip_address() is False
+
+    setup_core(address="192.168.1.100")
+    assert has_ip_address() is True
+
+    setup_core(address="10.0.0.1")
+    assert has_ip_address() is True
+
+    setup_core(address="esphome-device.local")
+    assert has_ip_address() is False
+
+    setup_core(address="my-device")
+    assert has_ip_address() is False
+
+
+def test_mqtt_get_ip() -> None:
+    """Test mqtt_get_ip function."""
+    config = {CONF_MQTT: {CONF_BROKER: "mqtt.local"}}
+
+    with patch("esphome.mqtt.get_esphome_device_ip") as mock_get_ip:
+        mock_get_ip.return_value = ["192.168.1.100", "192.168.1.101"]
+
+        result = mqtt_get_ip(config, "user", "pass", "client-id")
+
+        assert result == ["192.168.1.100", "192.168.1.101"]
+        mock_get_ip.assert_called_once_with(config, "user", "pass", "client-id")
+
+
+def test_has_resolvable_address() -> None:
+    """Test has_resolvable_address function."""
+
+    # Test with mDNS enabled and hostname address
+    setup_core(config={}, address="esphome-device.local")
+    assert has_resolvable_address() is True
+
+    # Test with mDNS disabled and hostname address
+    setup_core(
+        config={CONF_MDNS: {CONF_DISABLED: True}}, address="esphome-device.local"
+    )
+    assert has_resolvable_address() is False
+
+    # Test with IP address (mDNS doesn't matter)
+    setup_core(config={}, address="192.168.1.100")
+    assert has_resolvable_address() is True
+
+    # Test with IP address and mDNS disabled
+    setup_core(config={CONF_MDNS: {CONF_DISABLED: True}}, address="192.168.1.100")
+    assert has_resolvable_address() is True
+
+    # Test with no address but mDNS enabled (can still resolve mDNS names)
+    setup_core(config={}, address=None)
+    assert has_resolvable_address() is True
+
+    # Test with no address and mDNS disabled
+    setup_core(config={CONF_MDNS: {CONF_DISABLED: True}}, address=None)
+    assert has_resolvable_address() is False
+
+
 def test_command_wizard(tmp_path: Path) -> None:
     """Test command_wizard function."""
     config_file = tmp_path / "test.yaml"