sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-09-17 10:42:21 +01:00
Code Issues Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Markus
2025-09-12 23:31:53 +02:00
committed by Jesse Hills
parent 5b702a1efa
commit 46235684b1
3 changed files with 547 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

523
tests/unit_tests/test_main.py
View File

@@ -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"