[ota] Address remaining Copilot comments on partition-table OTA

- upload_program: allow MQTT/MQTTIP devices for --partition-table.
  MQTTIP gets resolved to a real IP by _resolve_network_devices(), so
  rejecting any non-NETWORK port_type was incorrect; only SERIAL and
  BOOTSEL are non-OTA upload paths.
- update_partition_table: re-initialize NVS on every failure path past
  nvs_flash_deinit() so components that survive a failed OTA aren't
  left with broken NVS handles. Success path stays as-is because the
  device reboots immediately afterwards.

Adds an MQTTIP upload test and refreshes the gate's comment.
This commit is contained in:
J. Nick Koston
2026-05-03 09:29:05 -05:00
parent 36c120fb07
commit b75f5034e5
3 changed files with 54 additions and 1 deletions
+5 -1
View File
@@ -1091,7 +1091,11 @@ def upload_program(
port_type = get_port_type(host)
if port_type != PortType.NETWORK and getattr(args, "partition_table", False):
# MQTT and MQTTIP are also OTA paths; MQTTIP gets resolved to a real IP later by
# _resolve_network_devices(). Only SERIAL and BOOTSEL are non-OTA upload paths.
if port_type in (PortType.SERIAL, PortType.BOOTSEL) and getattr(
args, "partition_table", False
):
raise EsphomeError(
"The option --partition-table can only be used for Over The Air updates."
)
@@ -375,6 +375,9 @@ OTAResponseTypes IDFOTABackend::update_partition_table() {
// Deinitialize NVS just before the first destructive write to the partition-table region. Doing
// this here (instead of earlier) means that any failure path in the verify or copy phases above
// returns with NVS still functional, so other components on the device aren't broken until reboot.
// Each failure path past this point calls nvs_flash_init() before returning so that, if the
// device keeps running, components that depend on NVS aren't permanently broken. The success
// path skips reinit because the device reboots immediately afterwards.
nvs_flash_deinit();
// Update the partition table
@@ -383,6 +386,7 @@ OTAResponseTypes IDFOTABackend::update_partition_table() {
esp_ota_abort(this->update_handle_);
this->update_handle_ = 0;
ESP_LOGE(TAG, "esp_ota_begin failed (err=0x%X)", err);
nvs_flash_init();
return OTA_RESPONSE_ERROR_PARTITION_TABLE_UPDATE;
}
err = esp_ota_write(this->update_handle_, this->buf_, ESP_PARTITION_TABLE_MAX_LEN);
@@ -392,12 +396,14 @@ OTAResponseTypes IDFOTABackend::update_partition_table() {
esp_ota_abort(this->update_handle_);
this->update_handle_ = 0;
ESP_LOGE(TAG, "esp_ota_write failed (err=0x%X)", err);
nvs_flash_init();
return OTA_RESPONSE_ERROR_PARTITION_TABLE_UPDATE;
}
err = esp_ota_end(this->update_handle_);
this->update_handle_ = 0; // esp_ota_end releases the handle internally regardless of result
if (err != ESP_OK) {
ESP_LOGE(TAG, "esp_ota_end failed (err=0x%X)", err);
nvs_flash_init();
return OTA_RESPONSE_ERROR_PARTITION_TABLE_UPDATE;
}
// esp_partition_unload_all() invalidates every cached partition entry, including the externally
@@ -412,12 +418,14 @@ OTAResponseTypes IDFOTABackend::update_partition_table() {
const esp_partition_t *new_boot_partition = find_app_partition_at(new_part->pos.offset, 0);
if (new_boot_partition == nullptr) {
ESP_LOGE(TAG, "Selected app partition not found after partition table update");
nvs_flash_init();
return OTA_RESPONSE_ERROR_PARTITION_TABLE_UPDATE;
}
ESP_LOGD(TAG, "Setting next boot partition to 0x%X", new_boot_partition->address);
err = esp_ota_set_boot_partition(new_boot_partition);
if (err != ESP_OK) {
ESP_LOGE(TAG, "esp_ota_set_boot_partition failed (err=0x%X)", err);
nvs_flash_init();
return OTA_RESPONSE_ERROR_PARTITION_TABLE_UPDATE;
}
return OTA_RESPONSE_OK;
+41
View File
@@ -1705,6 +1705,47 @@ def test_upload_program_serial_partition_table(
upload_program(config, args, devices)
def test_upload_program_ota_partition_table_mqttip(
mock_run_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""--partition-table is allowed for MQTTIP devices; they resolve to a real IP at OTA time."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "MQTTIP"
mock_run_ota.return_value = (0, "192.168.1.100")
partition_file = tmp_path / "partitions.bin"
partition_file.write_bytes(_make_partition_table_bytes())
config = {
CONF_OTA: [
{
CONF_PLATFORM: CONF_ESPHOME,
CONF_PORT: 3232,
"allow_partition_access": True,
}
]
}
args = MockArgs(file=str(partition_file), partition_table=True)
with patch(
"esphome.__main__._resolve_network_devices", return_value=["192.168.1.100"]
):
exit_code, host = upload_program(config, args, ["MQTTIP"])
assert exit_code == 0
assert host == "192.168.1.100"
mock_run_ota.assert_called_once_with(
["192.168.1.100"],
3232,
None,
partition_file,
OTA_TYPE_UPDATE_PARTITION_TABLE,
)
def test_validate_partition_table_binary_accepts_valid(tmp_path: Path) -> None:
f = tmp_path / "partitions.bin"
f.write_bytes(_make_partition_table_bytes())