mirror of
https://github.com/esphome/esphome.git
synced 2026-09-30 00:10:22 +00:00
Merge branch 'dev' into partition-table-ota
This commit is contained in:
Mat931
@@ -786,6 +786,59 @@ def test_perform_ota_version_differences(
|
||||
assert mock_socket.recv.call_count == 9 # v2.0 has 9 recv calls (includes chunk OK)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_extended_protocol_app(
|
||||
mock_socket: Mock, mock_file: io.BytesIO
|
||||
) -> None:
|
||||
"""Test OTA extended protocol app update."""
|
||||
recv_responses = [
|
||||
bytes([espota2.RESPONSE_OK]), # First byte of version response
|
||||
bytes([espota2.OTA_VERSION_2_0]), # Version number
|
||||
bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Device supports extended protocol
|
||||
bytes(
|
||||
[
|
||||
espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION
|
||||
| espota2.SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS
|
||||
]
|
||||
), # Device feature flags
|
||||
bytes([espota2.RESPONSE_AUTH_OK]), # No auth required
|
||||
bytes([espota2.RESPONSE_UPDATE_PREPARE_OK]), # Binary size OK
|
||||
bytes([espota2.RESPONSE_BIN_MD5_OK]), # MD5 checksum OK
|
||||
bytes([espota2.RESPONSE_CHUNK_OK]), # Chunk OK
|
||||
bytes([espota2.RESPONSE_RECEIVE_OK]), # Receive OK
|
||||
bytes([espota2.RESPONSE_UPDATE_END_OK]), # Update end OK
|
||||
]
|
||||
|
||||
mock_socket.recv.side_effect = recv_responses
|
||||
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
espota2.OTA_TYPE_UPDATE_APP,
|
||||
)
|
||||
|
||||
# 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 + extended protocol)
|
||||
assert mock_socket.sendall.call_args_list[1] == call(
|
||||
bytes(
|
||||
[
|
||||
espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
|
||||
| espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
|
||||
| espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
# Verify ota type was sent
|
||||
assert mock_socket.sendall.call_args_list[2] == call(
|
||||
bytes([espota2.OTA_TYPE_UPDATE_APP])
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_successful_partition_table(
|
||||
mock_socket: Mock, mock_file: io.BytesIO
|
||||
@@ -861,3 +914,167 @@ def test_perform_ota_extended_protocol_unsupported(
|
||||
"partitions.bin",
|
||||
espota2.OTA_TYPE_UPDATE_PARTITION_TABLE,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_device_rejects_with_unsupported_ota_type(
|
||||
mock_socket: Mock, mock_file: io.BytesIO
|
||||
) -> None:
|
||||
"""End-to-end: device returns 0x8E after the size byte; perform_ota must
|
||||
surface the human-readable 'unsupported OTA type' error from the lookup
|
||||
table in check_error()."""
|
||||
recv_responses = [
|
||||
bytes([espota2.RESPONSE_OK]), # First byte of version response
|
||||
bytes([espota2.OTA_VERSION_2_0]), # Version number
|
||||
bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Extended protocol marker
|
||||
bytes(
|
||||
[
|
||||
espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION
|
||||
| espota2.SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS
|
||||
]
|
||||
), # Feature flags
|
||||
bytes([espota2.RESPONSE_AUTH_OK]), # No auth required
|
||||
bytes([espota2.RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE]), # Reject at size step
|
||||
]
|
||||
|
||||
mock_socket.recv.side_effect = recv_responses
|
||||
|
||||
with pytest.raises(
|
||||
espota2.OTAError,
|
||||
match="The requested OTA type is not supported by the device",
|
||||
):
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
espota2.OTA_TYPE_UPDATE_APP,
|
||||
)
|
||||
|
||||
# Verify the client did send the OTA type byte before the size step
|
||||
assert mock_socket.sendall.call_args_list[2] == call(
|
||||
bytes([espota2.OTA_TYPE_UPDATE_APP])
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_unsupported_type_rejected_early(
|
||||
mock_socket: Mock, mock_file: io.BytesIO
|
||||
) -> None:
|
||||
"""ota_type values not in _SUPPORTED_OTA_TYPES are rejected before any I/O."""
|
||||
with pytest.raises(espota2.OTAError, match="Unsupported OTA type 0xFF"):
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
0xFF,
|
||||
)
|
||||
# No bytes should have been transmitted to the device.
|
||||
mock_socket.sendall.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_type", [-1, 256, 0x10000, "app", None, 1.5])
|
||||
def test_perform_ota_rejects_out_of_range_type(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, bad_type: object
|
||||
) -> None:
|
||||
"""Out-of-range or non-int ota_type must raise OTAError, not ValueError."""
|
||||
with pytest.raises(espota2.OTAError, match="Invalid ota_type"):
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
bad_type, # type: ignore[arg-type]
|
||||
)
|
||||
mock_socket.sendall.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_non_app_type_requires_extended_protocol(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Non-app OTA type must fail when device only supports the legacy protocol."""
|
||||
monkeypatch.setattr(
|
||||
espota2,
|
||||
"_SUPPORTED_OTA_TYPES",
|
||||
frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}),
|
||||
)
|
||||
recv_responses = [
|
||||
bytes([espota2.RESPONSE_OK]), # First byte of version response
|
||||
bytes([espota2.OTA_VERSION_2_0]), # Version number
|
||||
bytes([espota2.RESPONSE_HEADER_OK]), # Legacy single-byte feature ack
|
||||
]
|
||||
|
||||
mock_socket.recv.side_effect = recv_responses
|
||||
|
||||
with pytest.raises(
|
||||
espota2.OTAError, match="Device does not support extended OTA protocol"
|
||||
):
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
0xFF,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_non_app_type_requires_partition_access(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Non-app OTA type must fail when device advertises extended protocol but
|
||||
not the partition-access feature."""
|
||||
monkeypatch.setattr(
|
||||
espota2,
|
||||
"_SUPPORTED_OTA_TYPES",
|
||||
frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}),
|
||||
)
|
||||
recv_responses = [
|
||||
bytes([espota2.RESPONSE_OK]), # First byte of version response
|
||||
bytes([espota2.OTA_VERSION_2_0]), # Version number
|
||||
bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Extended protocol marker
|
||||
bytes(
|
||||
[espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION]
|
||||
), # Compression only, no partition access
|
||||
]
|
||||
|
||||
mock_socket.recv.side_effect = recv_responses
|
||||
|
||||
with pytest.raises(
|
||||
espota2.OTAError, match="Device does not support partition access"
|
||||
):
|
||||
espota2.perform_ota(
|
||||
mock_socket,
|
||||
"testpass",
|
||||
mock_file,
|
||||
"test.bin",
|
||||
0xFF,
|
||||
)
|
||||
|
||||
|
||||
def test_check_error_detects_errors_when_expect_is_none() -> None:
|
||||
"""check_error must surface device error bytes even when expect is None.
|
||||
|
||||
Regression test: previously, receive_exactly(..., expect=None) calls (used
|
||||
during feature negotiation and nonce reads) silently passed error bytes
|
||||
through, turning clean device errors into confusing later failures.
|
||||
"""
|
||||
with pytest.raises(espota2.OTAError, match="Error: Authentication invalid"):
|
||||
espota2.check_error([espota2.RESPONSE_ERROR_AUTH_INVALID], None)
|
||||
|
||||
|
||||
def test_check_error_detects_empty_when_expect_is_none() -> None:
|
||||
"""Empty data with expect=None must still raise (connection closed)."""
|
||||
with pytest.raises(
|
||||
espota2.OTAError, match="Device closed connection without responding"
|
||||
):
|
||||
espota2.check_error([], None)
|
||||
|
||||
|
||||
def test_check_error_passes_non_error_when_expect_is_none() -> None:
|
||||
"""Non-error bytes with expect=None must pass through silently."""
|
||||
espota2.check_error([espota2.RESPONSE_OK], None)
|
||||
espota2.check_error([espota2.RESPONSE_HEADER_OK], None)
|
||||
espota2.check_error([espota2.RESPONSE_FEATURE_FLAGS], None)
|
||||
Reference in New Issue
Block a user