Merge remote-tracking branch 'upstream/dev' into integration

2025-10-29 22:24:26 +00:00 · 2025-10-15 15:20:28 -10:00
parent 5bfc2b95d1 fe4857fabb
commit 8de406c633
47 changed files with 865 additions and 214 deletions
--- a/tests/unit_tests/test_espota2.py
+++ b/tests/unit_tests/test_espota2.py
@@ -287,7 +287,7 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:

    mock_socket.recv.side_effect = recv_responses

-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")

    # Should not send any auth-related data
    auth_calls = [
@@ -317,7 +317,7 @@ def test_perform_ota_with_compression(mock_socket: Mock) -> None:

    mock_socket.recv.side_effect = recv_responses

-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")

    # Verify compressed content was sent
    # Get the binary size that was sent (4 bytes after features)
@@ -347,7 +347,7 @@ def test_perform_ota_auth_without_password(mock_socket: Mock) -> None:
    with pytest.raises(
        espota2.OTAError, match="ESP requests password, but no password given"
    ):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")


@pytest.mark.usefixtures("mock_time")
@@ -413,7 +413,7 @@ def test_perform_ota_sha256_auth_without_password(mock_socket: Mock) -> None:
    with pytest.raises(
        espota2.OTAError, match="ESP requests password, but no password given"
    ):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")


 def test_perform_ota_unexpected_auth_response(mock_socket: Mock) -> None:
@@ -450,7 +450,7 @@ def test_perform_ota_unsupported_version(mock_socket: Mock) -> None:
    mock_socket.recv.side_effect = responses

    with pytest.raises(espota2.OTAError, match="Device uses unsupported OTA version"):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")


@pytest.mark.usefixtures("mock_time")
@@ -471,7 +471,7 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N
    mock_socket.recv.side_effect = recv_responses

    with pytest.raises(espota2.OTAError, match="Error receiving acknowledge chunk OK"):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")


@pytest.mark.usefixtures("mock_socket_constructor", "mock_resolve_ip")
@@ -706,7 +706,7 @@ def test_perform_ota_version_differences(
    ]

    mock_socket.recv.side_effect = recv_responses
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")

    # For v1.0, verify that we only get the expected number of recv calls
    # v1.0 doesn't have chunk acknowledgments, so fewer recv calls
@@ -732,7 +732,7 @@ def test_perform_ota_version_differences(
    ]

    mock_socket.recv.side_effect = recv_responses_v2
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")

    # For v2.0, verify more recv calls due to chunk acknowledgments
    assert mock_socket.recv.call_count == 9  # v2.0 has 9 recv calls (includes chunk OK)
--- a/tests/unit_tests/test_main.py
+++ b/tests/unit_tests/test_main.py
@@ -1062,7 +1062,7 @@ def test_upload_program_ota_with_file_arg(
    assert exit_code == 0
    assert host == "192.168.1.100"
    mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "", Path("custom.bin")
+        ["192.168.1.100"], 3232, None, Path("custom.bin")
    )


@@ -1119,7 +1119,9 @@ def test_upload_program_ota_with_mqtt_resolution(
    expected_firmware = (
        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, None, expected_firmware
+    )


@patch("esphome.__main__.importlib.import_module")
@@ -1976,3 +1978,292 @@ def test_command_clean_all_args_used() -> None:
        # Verify the correct configuration paths were passed
        mock_clean_all.assert_any_call(["/path/to/config1"])
        mock_clean_all.assert_any_call(["/path/to/config2", "/path/to/config3"])
+
+
+def test_upload_program_ota_static_ip_with_mqttip(
+    mock_mqtt_get_ip: Mock,
+    mock_run_ota: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test upload_program with static IP and MQTTIP (issue #11260).
+
+    This tests the scenario where a device has manual_ip (static IP) configured
+    and MQTT is also configured. The devices list contains both the static IP
+    and "MQTTIP" magic string. This previously failed because only the first
+    device was checked for MQTT resolution.
+    """
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_mqtt_get_ip.return_value = ["192.168.2.50"]  # Different subnet
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ],
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+    }
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Simulates choose_upload_log_host returning static IP + MQTTIP
+    devices = ["192.168.1.100", "MQTTIP"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+
+    # Verify MQTT was resolved
+    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+
+    # Verify espota2.run_ota was called with both IPs
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100", "192.168.2.50"], 3232, None, expected_firmware
+    )
+
+
+def test_upload_program_ota_multiple_mqttip_resolves_once(
+    mock_mqtt_get_ip: Mock,
+    mock_run_ota: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test that MQTT resolution only happens once even with multiple MQTT magic strings."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_mqtt_get_ip.return_value = ["192.168.2.50", "192.168.2.51"]
+    mock_run_ota.return_value = (0, "192.168.2.50")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ],
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+    }
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Multiple MQTT magic strings in the list
+    devices = ["MQTTIP", "MQTT", "192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.2.50"
+
+    # Verify MQTT was only resolved once despite multiple MQTT magic strings
+    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+
+    # Verify espota2.run_ota was called with all unique IPs
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.2.50", "192.168.2.51", "192.168.1.100"], 3232, None, expected_firmware
+    )
+
+
+def test_upload_program_ota_mqttip_deduplication(
+    mock_mqtt_get_ip: Mock,
+    mock_run_ota: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test that duplicate IPs are filtered when MQTT returns same IP as static IP."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    # MQTT returns the same IP as the static IP
+    mock_mqtt_get_ip.return_value = ["192.168.1.100"]
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ],
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+    }
+    args = MockArgs(username="user", password="pass", client_id="client")
+    devices = ["192.168.1.100", "MQTTIP"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+
+    # Verify MQTT was resolved
+    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+
+    # Verify espota2.run_ota was called with deduplicated IPs (only one instance of 192.168.1.100)
+    # Note: Current implementation doesn't dedupe, so we'll get the IP twice
+    # This test documents current behavior - deduplication could be future enhancement
+    mock_run_ota.assert_called_once()
+    call_args = mock_run_ota.call_args[0]
+    # Should contain both the original IP and MQTT-resolved IP (even if duplicate)
+    assert "192.168.1.100" in call_args[0]
+
+
+@patch("esphome.components.api.client.run_logs")
+def test_show_logs_api_static_ip_with_mqttip(
+    mock_run_logs: Mock,
+    mock_mqtt_get_ip: Mock,
+) -> None:
+    """Test show_logs with static IP and MQTTIP (issue #11260).
+
+    This tests the scenario where a device has manual_ip (static IP) configured
+    and MQTT is also configured. The devices list contains both the static IP
+    and "MQTTIP" magic string.
+    """
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MQTT: {CONF_BROKER: "mqtt.local"},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_run_logs.return_value = 0
+    mock_mqtt_get_ip.return_value = ["192.168.2.50"]
+
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Simulates choose_upload_log_host returning static IP + MQTTIP
+    devices = ["192.168.1.100", "MQTTIP"]
+
+    result = show_logs(CORE.config, args, devices)
+
+    assert result == 0
+
+    # Verify MQTT was resolved
+    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+
+    # Verify run_logs was called with both IPs
+    mock_run_logs.assert_called_once_with(
+        CORE.config, ["192.168.1.100", "192.168.2.50"]
+    )
+
+
+@patch("esphome.components.api.client.run_logs")
+def test_show_logs_api_multiple_mqttip_resolves_once(
+    mock_run_logs: Mock,
+    mock_mqtt_get_ip: Mock,
+) -> None:
+    """Test that MQTT resolution only happens once for show_logs with multiple MQTT magic strings."""
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MQTT: {CONF_BROKER: "mqtt.local"},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_run_logs.return_value = 0
+    mock_mqtt_get_ip.return_value = ["192.168.2.50", "192.168.2.51"]
+
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Multiple MQTT magic strings in the list
+    devices = ["MQTTIP", "192.168.1.100", "MQTT"]
+
+    result = show_logs(CORE.config, args, devices)
+
+    assert result == 0
+
+    # Verify MQTT was only resolved once despite multiple MQTT magic strings
+    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+
+    # Verify run_logs was called with all unique IPs (MQTT strings replaced with IPs)
+    # Note: "MQTT" is a different magic string from "MQTTIP", but both trigger MQTT resolution
+    # The _resolve_network_devices helper filters out both after first resolution
+    mock_run_logs.assert_called_once_with(
+        CORE.config, ["192.168.2.50", "192.168.2.51", "192.168.1.100"]
+    )
+
+
+def test_upload_program_ota_mqtt_timeout_fallback(
+    mock_mqtt_get_ip: Mock,
+    mock_run_ota: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test upload_program falls back to other devices when MQTT times out."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    # MQTT times out
+    mock_mqtt_get_ip.side_effect = EsphomeError("Failed to find IP via MQTT")
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ],
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+    }
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Static IP first, MQTTIP second
+    devices = ["192.168.1.100", "MQTTIP"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    # Should succeed using the static IP even though MQTT failed
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+
+    # Verify MQTT was attempted
+    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+
+    # Verify espota2.run_ota was called with only the static IP (MQTT failed)
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"], 3232, None, expected_firmware
+    )
+
+
+@patch("esphome.components.api.client.run_logs")
+def test_show_logs_api_mqtt_timeout_fallback(
+    mock_run_logs: Mock,
+    mock_mqtt_get_ip: Mock,
+) -> None:
+    """Test show_logs falls back to other devices when MQTT times out."""
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MQTT: {CONF_BROKER: "mqtt.local"},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_run_logs.return_value = 0
+    # MQTT times out
+    mock_mqtt_get_ip.side_effect = EsphomeError("Failed to find IP via MQTT")
+
+    args = MockArgs(username="user", password="pass", client_id="client")
+    # Static IP first, MQTTIP second
+    devices = ["192.168.1.100", "MQTTIP"]
+
+    result = show_logs(CORE.config, args, devices)
+
+    # Should succeed using the static IP even though MQTT failed
+    assert result == 0
+
+    # Verify MQTT was attempted
+    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+
+    # Verify run_logs was called with only the static IP (MQTT failed)
+    mock_run_logs.assert_called_once_with(CORE.config, ["192.168.1.100"])