github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-07-10 17:05:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add extended OTA protocol (split from PR #15780)

Browse Source
This commit is contained in:
Mat931
2026-04-30 10:51:12 +02:00
parent 1398dcebb4
commit c99e60be7d
9 changed files with 242 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/__main__.py
+6 -5
View File
@@ -1125,15 +1125,16 @@ def upload_program(
remote_port = int(ota_conf[CONF_PORT])
password = ota_conf.get(CONF_PASSWORD)
if getattr(args, "file", None) is not None:
binary = Path(args.file)
else:
binary = CORE.firmware_bin
# Resolve MQTT magic strings to actual IP addresses
network_devices = _resolve_network_devices(devices, config, args)
return espota2.run_ota(network_devices, remote_port, password, binary)
binary = CORE.firmware_bin
ota_type = espota2.OTA_TYPE_UPDATE_APP
if getattr(args, "file", None) is not None:
binary = Path(args.file)
return espota2.run_ota(network_devices, remote_port, password, binary, ota_type)
def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int | None:
esphome/components/esphome/ota/__init__.py
+12 -1
View File
@@ -16,11 +16,13 @@ from esphome.const import (
CONF_SAFE_MODE,
CONF_VERSION,
)
from esphome.core import coroutine_with_priority
from esphome.core import CORE, coroutine_with_priority
from esphome.coroutine import CoroPriority
import esphome.final_validate as fv
from esphome.types import ConfigType
CONF_ALLOW_PARTITION_ACCESS = "allow_partition_access"
_LOGGER = logging.getLogger(__name__)
@@ -75,6 +77,10 @@ def ota_esphome_final_validate(config):
merged_ota_esphome_configs_by_port[conf_port] = merge_config(
merged_ota_esphome_configs_by_port[conf_port], ota_conf
)
if ota_conf.get(CONF_ALLOW_PARTITION_ACCESS) and not CORE.is_esp32:
raise cv.Invalid(
f"{CONF_ALLOW_PARTITION_ACCESS} is only supported on the esp32"
)
else:
new_ota_conf.append(ota_conf)
@@ -125,6 +131,7 @@ CONFIG_SCHEMA = cv.All(
ln882x=8820,
rtl87xx=8892,
): cv.port,
cv.Optional(CONF_ALLOW_PARTITION_ACCESS, default=False): cv.boolean,
cv.Optional(CONF_PASSWORD): cv.string,
cv.Optional(CONF_NUM_ATTEMPTS): cv.invalid(
f"'{CONF_SAFE_MODE}' (and its related configuration variables) has moved from 'ota' to its own component. See https://esphome.io/components/safe_mode"
@@ -159,6 +166,10 @@ async def to_code(config: ConfigType) -> None:
if config[CONF_PASSWORD]:
cg.add(var.set_auth_password(config[CONF_PASSWORD]))
cg.add_define("USE_OTA_VERSION", config[CONF_VERSION])
if config.get(CONF_ALLOW_PARTITION_ACCESS):
cg.add_define("USE_OTA_PARTITIONS")
# Build flag so lwip_fast_select.c (a .c file that can't include defines.h) sees it.
cg.add_build_flag("-DUSE_OTA_PLATFORM_ESPHOME")
esphome/components/esphome/ota/ota_esphome.cpp
+56 -9
View File
@@ -100,6 +100,9 @@ void ESPHomeOTAComponent::dump_config() {
ESP_LOGCONFIG(TAG, " Password configured");
}
#endif
#ifdef USE_OTA_PARTITIONS
ESP_LOGCONFIG(TAG, " Partition access allowed");
#endif
}
void ESPHomeOTAComponent::loop() {
@@ -114,8 +117,11 @@ void ESPHomeOTAComponent::loop() {
this->handle_handshake_();
}
static const uint8_t FEATURE_SUPPORTS_COMPRESSION = 0x01;
static const uint8_t FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01;
static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04;
static constexpr uint8_t SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01;
static constexpr uint8_t SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02;
void ESPHomeOTAComponent::handle_handshake_() {
/// Handle the OTA handshake and authentication.
@@ -201,16 +207,36 @@ void ESPHomeOTAComponent::handle_handshake_() {
this->ota_features_ = this->handshake_buf_[0];
ESP_LOGV(TAG, "Features: 0x%02X", this->ota_features_);
this->transition_ota_state_(OTAState::FEATURE_ACK);
this->handshake_buf_[0] =
((this->ota_features_ & FEATURE_SUPPORTS_COMPRESSION) != 0 && this->backend_->supports_compression())
? ota::OTA_RESPONSE_SUPPORTS_COMPRESSION
: ota::OTA_RESPONSE_HEADER_OK;
const bool supports_compression =
(this->ota_features_ & CLIENT_FEATURE_SUPPORTS_COMPRESSION) != 0 && this->backend_->supports_compression();
// Compose the feature-ack response. When USE_OTA_PARTITIONS is enabled and the client
// negotiates the extended protocol we emit a 2-byte response (marker + server feature flags);
// otherwise we emit the single-byte legacy response. The #ifdef wraps only the extended-proto
// branch so the legacy branch reads as unconditional code in either build configuration.
#ifdef USE_OTA_PARTITIONS
this->extended_proto_ = (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL) != 0;
if (this->extended_proto_) {
this->handshake_buf_[0] = ota::OTA_RESPONSE_FEATURE_FLAGS;
this->handshake_buf_[1] =
SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS | (supports_compression ? SERVER_FEATURE_SUPPORTS_COMPRESSION : 0);
} else
#endif
{
this->handshake_buf_[0] =
supports_compression ? ota::OTA_RESPONSE_SUPPORTS_COMPRESSION : ota::OTA_RESPONSE_HEADER_OK;
}
[[fallthrough]];
}
case OTAState::FEATURE_ACK: {
// Acknowledge header - 1 byte
if (!this->try_write_(1, LOG_STR("ack feature"))) {
#ifdef USE_OTA_PARTITIONS
const size_t ack_size = this->extended_proto_ ? 2 : 1;
#else
const size_t ack_size = 1;
#endif
if (!this->try_write_(ack_size, LOG_STR("ack feature"))) {
return;
}
#ifdef USE_OTA_PASSWORD
@@ -296,6 +322,9 @@ void ESPHomeOTAComponent::handle_data_() {
uint8_t buf[OTA_BUFFER_SIZE];
char *sbuf = reinterpret_cast<char *>(buf);
size_t ota_size;
#ifdef USE_OTA_PARTITIONS
ota::OTAType ota_type = ota::OTA_TYPE_UPDATE_APP;
#endif
#if USE_OTA_VERSION == 2
size_t size_acknowledged = 0;
#endif
@@ -311,6 +340,18 @@ void ESPHomeOTAComponent::handle_data_() {
// Acknowledge auth OK - 1 byte
this->write_byte_(ota::OTA_RESPONSE_AUTH_OK);
#ifdef USE_OTA_PARTITIONS
if (this->extended_proto_) {
// Read ota type, 1 byte
if (!this->readall_(buf, 1)) {
this->log_read_error_(LOG_STR("OTA type"));
goto error; // NOLINT(cppcoreguidelines-avoid-goto)
}
ota_type = static_cast<ota::OTAType>(buf[0]);
}
ESP_LOGV(TAG, "OTA type is 0x%02x", ota_type);
#endif
// Read size, 4 bytes MSB first
if (!this->readall_(buf, 4)) {
this->log_read_error_(LOG_STR("size"));
@@ -331,6 +372,12 @@ void ESPHomeOTAComponent::handle_data_() {
#endif
// This will block for a few seconds as it locks flash
#ifdef USE_OTA_PARTITIONS
if (ota_type != ota::OTA_TYPE_UPDATE_APP) {
error_code = ota::OTA_RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE;
goto error; // NOLINT(cppcoreguidelines-avoid-goto)
}
#endif
error_code = this->backend_->begin(ota_size);
if (error_code != ota::OTA_RESPONSE_OK)
goto error; // NOLINT(cppcoreguidelines-avoid-goto)
@@ -616,7 +663,7 @@ void ESPHomeOTAComponent::yield_and_feed_watchdog_() {
void ESPHomeOTAComponent::log_auth_warning_(const LogString *msg) { ESP_LOGW(TAG, "Auth: %s", LOG_STR_ARG(msg)); }
bool ESPHomeOTAComponent::select_auth_type_() {
bool client_supports_sha256 = (this->ota_features_ & FEATURE_SUPPORTS_SHA256_AUTH) != 0;
bool client_supports_sha256 = (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_SHA256_AUTH) != 0;
// Require SHA256
if (!client_supports_sha256) {
esphome/components/esphome/ota/ota_esphome.h
+3
View File
@@ -91,6 +91,9 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
std::string password_;
std::unique_ptr<uint8_t[]> auth_buf_;
#endif // USE_OTA_PASSWORD
#ifdef USE_OTA_PARTITIONS
bool extended_proto_{false};
#endif
socket::ListenSocket *server_{nullptr};
std::unique_ptr<socket::Socket> client_;
esphome/components/ota/ota_backend.h
+8
View File
@@ -4,6 +4,8 @@
#include "esphome/core/defines.h"
#include "esphome/core/helpers.h"
#include <cstdint>
#ifdef USE_OTA_STATE_LISTENER
#include <vector>
#endif
@@ -23,6 +25,7 @@ enum OTAResponseTypes {
OTA_RESPONSE_UPDATE_END_OK = 0x45,
OTA_RESPONSE_SUPPORTS_COMPRESSION = 0x46,
OTA_RESPONSE_CHUNK_OK = 0x47,
OTA_RESPONSE_FEATURE_FLAGS = 0x48,
OTA_RESPONSE_ERROR_MAGIC = 0x80,
OTA_RESPONSE_ERROR_UPDATE_PREPARE = 0x81,
@@ -38,6 +41,7 @@ enum OTAResponseTypes {
OTA_RESPONSE_ERROR_MD5_MISMATCH = 0x8B,
OTA_RESPONSE_ERROR_RP2040_NOT_ENOUGH_SPACE = 0x8C,
OTA_RESPONSE_ERROR_SIGNATURE_INVALID = 0x8D,
OTA_RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE = 0x8E,
OTA_RESPONSE_ERROR_UNKNOWN = 0xFF,
};
@@ -49,6 +53,10 @@ enum OTAState {
OTA_ERROR,
};
enum OTAType : uint8_t {
OTA_TYPE_UPDATE_APP = 0x00,
};
/** Listener interface for OTA state changes.
*
* Components can implement this interface to receive OTA state updates
esphome/espota2.py
+117 -73
View File
@@ -15,6 +15,8 @@ from typing import Any
from esphome.core import EsphomeError
from esphome.helpers import ProgressBar, resolve_ip_address
OTA_TYPE_UPDATE_APP = 0x00
RESPONSE_OK = 0x00
RESPONSE_REQUEST_AUTH = 0x01
RESPONSE_REQUEST_SHA256_AUTH = 0x02
@@ -27,6 +29,7 @@ RESPONSE_RECEIVE_OK = 0x44
RESPONSE_UPDATE_END_OK = 0x45
RESPONSE_SUPPORTS_COMPRESSION = 0x46
RESPONSE_CHUNK_OK = 0x47
RESPONSE_FEATURE_FLAGS = 0x48
RESPONSE_ERROR_MAGIC = 0x80
RESPONSE_ERROR_UPDATE_PREPARE = 0x81
@@ -42,6 +45,7 @@ RESPONSE_ERROR_NO_UPDATE_PARTITION = 0x8A
RESPONSE_ERROR_MD5_MISMATCH = 0x8B
RESPONSE_ERROR_RP2040_NOT_ENOUGH_SPACE = 0x8C
RESPONSE_ERROR_SIGNATURE_INVALID = 0x8D
RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE = 0x8E
RESPONSE_ERROR_UNKNOWN = 0xFF
OTA_VERSION_1_0 = 1
@@ -49,9 +53,11 @@ OTA_VERSION_2_0 = 2
MAGIC_BYTES = [0x6C, 0x26, 0xF7, 0x5C, 0x45]
FEATURE_SUPPORTS_COMPRESSION = 0x01
FEATURE_SUPPORTS_SHA256_AUTH = 0x02
CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01
CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02
CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04
SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01
SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02
UPLOAD_BLOCK_SIZE = 8192
UPLOAD_BUFFER_SIZE = UPLOAD_BLOCK_SIZE * 8
@@ -64,6 +70,62 @@ _AUTH_METHODS: dict[int, tuple[Callable[..., Any], int, str]] = {
RESPONSE_REQUEST_AUTH: (hashlib.md5, 32, "MD5"),
}
# Error response code -> human-readable message (without the "Error: " prefix; check_error()
# prepends it uniformly). Looked up by check_error() to translate a single byte from the device
# into an OTAError. Add new error codes here rather than extending the if-chain in check_error().
_ERROR_MESSAGES: dict[int, str] = {
RESPONSE_ERROR_MAGIC: "Invalid magic byte",
RESPONSE_ERROR_UPDATE_PREPARE: (
"Couldn't prepare flash memory for update. Is the binary too big? "
"Please try restarting the ESP."
),
RESPONSE_ERROR_AUTH_INVALID: "Authentication invalid. Is the password correct?",
RESPONSE_ERROR_WRITING_FLASH: (
"Writing OTA data to flash memory failed. See USB logs for more information."
),
RESPONSE_ERROR_UPDATE_END: (
"Finishing update failed. See the MQTT/USB logs for more information."
),
RESPONSE_ERROR_INVALID_BOOTSTRAPPING: (
"Please press the reset button on the ESP. A manual reset is "
"required on the first OTA-Update after flashing via USB."
),
RESPONSE_ERROR_WRONG_CURRENT_FLASH_CONFIG: (
"ESP has been flashed with wrong flash size. Please choose the "
"correct 'board' option (esp01_1m always works) and then flash over USB."
),
RESPONSE_ERROR_WRONG_NEW_FLASH_CONFIG: (
"ESP does not have the requested flash size (wrong board). Please "
"choose the correct 'board' option (esp01_1m always works) and try "
"uploading again."
),
RESPONSE_ERROR_ESP8266_NOT_ENOUGH_SPACE: (
"ESP does not have enough space to store OTA file. Please try "
"flashing a minimal firmware (remove everything except ota)"
),
RESPONSE_ERROR_ESP32_NOT_ENOUGH_SPACE: (
"The OTA partition on the ESP is too small. ESPHome needs to resize "
"this partition, please flash over USB."
),
RESPONSE_ERROR_NO_UPDATE_PARTITION: (
"The OTA partition on the ESP couldn't be found. ESPHome needs to "
"create this partition, please flash over USB."
),
RESPONSE_ERROR_MD5_MISMATCH: (
"Application MD5 code mismatch. Please try again "
"or flash over USB with a good quality cable."
),
RESPONSE_ERROR_SIGNATURE_INVALID: (
"Firmware signature verification failed. The firmware was not signed "
"with the correct key. Ensure the signing key matches the one used to build "
"the firmware currently running on the device."
),
RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE: (
"The requested OTA type is not supported by the device."
),
RESPONSE_ERROR_UNKNOWN: "Unknown error from ESP",
}
class OTAError(EsphomeError):
pass
@@ -139,69 +201,9 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None
"a network issue, or the connection was interrupted."
)
dat = data[0]
if dat == RESPONSE_ERROR_MAGIC:
raise OTAError("Error: Invalid magic byte")
if dat == RESPONSE_ERROR_UPDATE_PREPARE:
raise OTAError(
"Error: Couldn't prepare flash memory for update. Is the binary too big? "
"Please try restarting the ESP."
)
if dat == RESPONSE_ERROR_AUTH_INVALID:
raise OTAError("Error: Authentication invalid. Is the password correct?")
if dat == RESPONSE_ERROR_WRITING_FLASH:
raise OTAError(
"Error: Writing OTA data to flash memory failed. See USB logs for more "
"information."
)
if dat == RESPONSE_ERROR_UPDATE_END:
raise OTAError(
"Error: Finishing update failed. See the MQTT/USB logs for more "
"information."
)
if dat == RESPONSE_ERROR_INVALID_BOOTSTRAPPING:
raise OTAError(
"Error: Please press the reset button on the ESP. A manual reset is "
"required on the first OTA-Update after flashing via USB."
)
if dat == RESPONSE_ERROR_WRONG_CURRENT_FLASH_CONFIG:
raise OTAError(
"Error: ESP has been flashed with wrong flash size. Please choose the "
"correct 'board' option (esp01_1m always works) and then flash over USB."
)
if dat == RESPONSE_ERROR_WRONG_NEW_FLASH_CONFIG:
raise OTAError(
"Error: ESP does not have the requested flash size (wrong board). Please "
"choose the correct 'board' option (esp01_1m always works) and try "
"uploading again."
)
if dat == RESPONSE_ERROR_ESP8266_NOT_ENOUGH_SPACE:
raise OTAError(
"Error: ESP does not have enough space to store OTA file. Please try "
"flashing a minimal firmware (remove everything except ota)"
)
if dat == RESPONSE_ERROR_ESP32_NOT_ENOUGH_SPACE:
raise OTAError(
"Error: The OTA partition on the ESP is too small. ESPHome needs to resize "
"this partition, please flash over USB."
)
if dat == RESPONSE_ERROR_NO_UPDATE_PARTITION:
raise OTAError(
"Error: The OTA partition on the ESP couldn't be found. ESPHome needs to create "
"this partition, please flash over USB."
)
if dat == RESPONSE_ERROR_MD5_MISMATCH:
raise OTAError(
"Error: Application MD5 code mismatch. Please try again "
"or flash over USB with a good quality cable."
)
if dat == RESPONSE_ERROR_SIGNATURE_INVALID:
raise OTAError(
"Error: Firmware signature verification failed. The firmware was not signed "
"with the correct key. Ensure the signing key matches the one used to build "
"the firmware currently running on the device."
)
if dat == RESPONSE_ERROR_UNKNOWN:
raise OTAError("Unknown error from ESP")
error_msg = _ERROR_MESSAGES.get(dat)
if error_msg is not None:
raise OTAError(f"Error: {error_msg}")
if not isinstance(expect, (list, tuple)):
expect = [expect]
if dat not in expect:
@@ -232,7 +234,11 @@ def send_check(
def perform_ota(
sock: socket.socket, password: str | None, file_handle: io.IOBase, filename: Path
sock: socket.socket,
password: str | None,
file_handle: io.IOBase,
filename: Path,
ota_type: int = OTA_TYPE_UPDATE_APP,
) -> None:
file_contents = file_handle.read()
file_size = len(file_contents)
@@ -251,7 +257,11 @@ def perform_ota(
)
# Features - send both compression and SHA256 auth support
features_to_send = FEATURE_SUPPORTS_COMPRESSION | FEATURE_SUPPORTS_SHA256_AUTH
features_to_send = (
CLIENT_FEATURE_SUPPORTS_COMPRESSION
| CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
| CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
)
send_check(sock, features_to_send, "features")
features = receive_exactly(
sock,
@@ -260,7 +270,30 @@ def perform_ota(
None, # Accept any response
)[0]
if features == RESPONSE_SUPPORTS_COMPRESSION:
extended_proto = False
if features == RESPONSE_FEATURE_FLAGS:
extended_proto = True
features = receive_exactly(
sock,
1,
"feature flags",
None, # Accept any response
)[0]
elif features == RESPONSE_SUPPORTS_COMPRESSION:
features = SERVER_FEATURE_SUPPORTS_COMPRESSION
else:
features = 0
if ota_type not in (OTA_TYPE_UPDATE_APP):
raise OTAError(f"Unsupported OTA type: 0x{ota_type:02X}")
if (
ota_type != OTA_TYPE_UPDATE_APP
and not features & SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS
):
raise OTAError("Device only supports app updates")
if features & SERVER_FEATURE_SUPPORTS_COMPRESSION:
upload_contents = gzip.compress(file_contents, compresslevel=9)
_LOGGER.info("Compressed to %s bytes", len(upload_contents))
else:
@@ -315,6 +348,9 @@ def perform_ota(
# Timeout must match device-side OTA_SOCKET_TIMEOUT_DATA to prevent premature failures
sock.settimeout(90.0)
if extended_proto:
send_check(sock, ota_type, "ota type")
upload_size = len(upload_contents)
upload_size_encoded = [
(upload_size >> 24) & 0xFF,
@@ -375,7 +411,11 @@ def perform_ota(
def run_ota_impl_(
remote_host: str | list[str], remote_port: int, password: str | None, filename: Path
remote_host: str | list[str],
remote_port: int,
password: str | None,
filename: Path,
ota_type: int = OTA_TYPE_UPDATE_APP,
) -> tuple[int, str | None]:
from esphome.core import CORE
@@ -413,7 +453,7 @@ def run_ota_impl_(
_LOGGER.info("Connected to %s", sa[0])
with open(filename, "rb") as file_handle:
try:
perform_ota(sock, password, file_handle, filename)
perform_ota(sock, password, file_handle, filename, ota_type)
except OTAError as err:
_LOGGER.error(str(err))
return 1, None
@@ -428,10 +468,14 @@ def run_ota_impl_(
def run_ota(
remote_host: str | list[str], remote_port: int, password: str | None, filename: Path
remote_host: str | list[str],
remote_port: int,
password: str | None,
filename: Path,
ota_type: int = OTA_TYPE_UPDATE_APP,
) -> tuple[int, str | None]:
try:
return run_ota_impl_(remote_host, remote_port, password, filename)
return run_ota_impl_(remote_host, remote_port, password, filename, ota_type)
except OTAError as err:
_LOGGER.error(err)
return 1, None
tests/components/ota/test-partition_access.esp32-idf.yaml
+5
View File
@@ -0,0 +1,5 @@
ota:
- platform: esphome
allow_partition_access: true
<<: !include common.yaml
tests/unit_tests/test_espota2.py
+19 -8
View File
@@ -185,6 +185,14 @@ def test_receive_exactly_socket_error(mock_socket: Mock) -> None:
"Error: The OTA partition on the ESP couldn't be found",
),
(espota2.RESPONSE_ERROR_MD5_MISMATCH, "Error: Application MD5 code mismatch"),
(
espota2.RESPONSE_ERROR_SIGNATURE_INVALID,
"Error: Firmware signature verification failed",
),
(
espota2.RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE,
"Error: The requested OTA type is not supported by the device",
),
(espota2.RESPONSE_ERROR_UNKNOWN, "Unknown error from ESP"),
],
)
@@ -270,12 +278,13 @@ def test_perform_ota_successful_md5_auth(
# Verify magic bytes were sent
assert mock_socket.sendall.call_args_list[0] == call(bytes(espota2.MAGIC_BYTES))
# Verify features were sent (compression + SHA256 support)
# Verify features were sent (compression + SHA256 support + extended protocol)
assert mock_socket.sendall.call_args_list[1] == call(
bytes(
[
espota2.FEATURE_SUPPORTS_COMPRESSION
| espota2.FEATURE_SUPPORTS_SHA256_AUTH
espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
| espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
| espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
]
)
)
@@ -640,12 +649,13 @@ def test_perform_ota_successful_sha256_auth(
# Verify magic bytes were sent
assert mock_socket.sendall.call_args_list[0] == call(bytes(espota2.MAGIC_BYTES))
# Verify features were sent (compression + SHA256 support)
# Verify features were sent (compression + SHA256 support + extended protocol)
assert mock_socket.sendall.call_args_list[1] == call(
bytes(
[
espota2.FEATURE_SUPPORTS_COMPRESSION
| espota2.FEATURE_SUPPORTS_SHA256_AUTH
espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
| espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
| espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
]
)
)
@@ -699,8 +709,9 @@ def test_perform_ota_sha256_fallback_to_md5(
assert mock_socket.sendall.call_args_list[1] == call(
bytes(
[
espota2.FEATURE_SUPPORTS_COMPRESSION
| espota2.FEATURE_SUPPORTS_SHA256_AUTH
espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
| espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
| espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
]
)
)
tests/unit_tests/test_main.py
+16 -7
View File
@@ -83,6 +83,7 @@ from esphome.const import (
PLATFORM_RP2040,
)
from esphome.core import CORE, EsphomeError
from esphome.espota2 import OTA_TYPE_UPDATE_APP
from esphome.util import BootselResult
from esphome.zeroconf import _await_discovery, discover_mdns_devices
@@ -1593,7 +1594,7 @@ def test_upload_program_ota_success(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, "secret", expected_firmware
["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP
)
@@ -1624,7 +1625,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, None, Path("custom.bin")
["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP
)
@@ -1682,7 +1683,7 @@ def test_upload_program_ota_with_mqtt_resolution(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, expected_firmware
["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
)
@@ -1730,7 +1731,7 @@ def test_upload_program_ota_with_mqtt_empty_broker(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.50"], 3232, None, expected_firmware
["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
)
# Verify warning was logged
assert "MQTT IP discovery failed" in caplog.text
@@ -3207,7 +3208,11 @@ def test_upload_program_ota_static_ip_with_mqttip(
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
["192.168.1.100", "192.168.2.50"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
)
@@ -3250,7 +3255,11 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
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
["192.168.2.50", "192.168.2.51", "192.168.1.100"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
)
@@ -3415,7 +3424,7 @@ def test_upload_program_ota_mqtt_timeout_fallback(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, expected_firmware
["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
)