diff --git a/esphome/__main__.py b/esphome/__main__.py index bba254436e..0147a82530 100644 --- a/esphome/__main__.py +++ b/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, - show_mqtt: bool, - show_api: bool, - purpose: str | None = None, + purpose: Purpose, ) -> 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 ( - (show_ota and "ota" in CORE.config) - or (show_api and "api" in CORE.config) + # ensure IP adresses are used first + if is_ip_address(CORE.address) and ( + (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 show_mqtt and has_mqtt_logging(): - mqtt_config = CORE.config[CONF_MQTT] - options.append((f"MQTT ({mqtt_config[CONF_BROKER]})", "MQTT")) + + if purpose == Purpose.LOGGING: + if has_mqtt_logging(): + mqtt_config = CORE.config[CONF_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: - """Check if MQTT logging is available.""" - return (mqtt_config := CORE.config.get(CONF_MQTT)) and mqtt_logging_enabled( - mqtt_config - ) +def has_mqtt() -> bool: + """Check if MQTT is available.""" + return CONF_MQTT in CORE.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 "MQTT" - return "NETWORK" + return _PORT_TO_PORT_TYPE.get(port, "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 - # This happens when no device was specified, or the current host is "MQTT"/"OTA" - if ( - 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 - ) - ] + # MQTT address resolution + if get_port_type(host) in ("MQTT", "MQTTIP"): + devices = mqtt_get_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) - if get_port_type(port) in ("NETWORK", "MQTT") and "mqtt" in config: + return run_logs(config, addresses_to_use) + + 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, - show_mqtt=False, - show_api=False, - purpose="uploading", + purpose=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, - show_mqtt=True, - show_api=True, - purpose="logging", + purpose=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, - show_mqtt=False, - show_api=True, - purpose="uploading", + purpose=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, - show_mqtt=True, - show_api=True, - purpose="logging", + purpose=Purpose.LOGGING, ) return show_logs(config, args, devices) diff --git a/esphome/const.py b/esphome/const.py index 308abe7706..677b9173ec 100644 --- a/esphome/const.py +++ b/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" diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 96ee43a55f..bfebb44545 100644 --- a/tests/unit_tests/test_main.py +++ b/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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.LOGGING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=True, - show_api=False, + purpose=Purpose.UPLOADING, ) - 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, - purpose="testing", + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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.""" - setup_core(config={"api": {}}, address="192.168.1.100") + """Test OTA device when API is configured (no upload without OTA in config).""" + setup_core(config={CONF_API: {}}, address="192.168.1.100") result = choose_upload_log_host( default="OTA", check_default=None, - show_ota=False, - show_mqtt=False, - show_api=True, + purpose=Purpose.UPLOADING, + ) + 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, - show_mqtt=True, - show_api=False, + purpose=Purpose.LOGGING, ) 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, - show_mqtt=True, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, - purpose="uploading", + purpose=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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=True, + purpose=Purpose.LOGGING, ) 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, - show_mqtt=True, - show_api=False, + purpose=Purpose.LOGGING, ) 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, - show_mqtt=True, - show_api=True, - purpose="testing", + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=True, + purpose=Purpose.UPLOADING, ) 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, - show_mqtt=False, - show_api=False, + purpose=Purpose.UPLOADING, ) 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( - [["192.168.1.100"]], 3232, "", expected_firmware - ) + mock_run_ota.assert_called_once_with(["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( - mock_get_port_type: Mock, -) -> None: +def test_show_logs_no_method_configured() -> 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"