sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-09-22 05:02:23 +01:00
Code Issues Releases Wiki Activity

preen

Browse Source
This commit is contained in:
J. Nick Koston
2025-09-21 10:07:01 -06:00
parent 0d622fa268
commit 594c60a4a4
1 changed files with 235 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
tests/unit_tests/test_espota2.py
View File

@@ -216,16 +216,10 @@ def test_send_check_socket_error(mock_socket) -> None:
espota2.send_check(mock_socket, b"data", "test")
def test_perform_ota_successful_md5_auth(mock_socket) -> None:
def test_perform_ota_successful_md5_auth(
mock_socket, mock_file, mock_time, mock_random
) -> None:
"""Test successful OTA with MD5 authentication."""
mock_file = io.BytesIO(b"firmware content here")
# Mock random for predictable cnonce
with (
patch("random.random", return_value=0.123456),
patch("time.sleep"),
patch("time.perf_counter", side_effect=[0, 1]),
):
# Setup socket responses for recv calls
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
@@ -272,11 +266,8 @@ def test_perform_ota_successful_md5_auth(mock_socket) -> None:
assert mock_socket.sendall.call_args_list[3] == call(expected_result.encode())
def test_perform_ota_no_auth(mock_socket) -> None:
def test_perform_ota_no_auth(mock_socket, mock_file, mock_time) -> None:
"""Test OTA without authentication."""
mock_file = io.BytesIO(b"firmware")
with patch("time.sleep"), patch("time.perf_counter", side_effect=[0, 1]):
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
bytes([espota2.OTA_VERSION_1_0]), # Version number
@@ -301,18 +292,14 @@ def test_perform_ota_no_auth(mock_socket) -> None:
assert len(auth_calls) == 0
def test_perform_ota_with_compression(mock_socket) -> None:
def test_perform_ota_with_compression(mock_socket, mock_time) -> None:
"""Test OTA with compression support."""
original_content = b"firmware" * 100 # Repeating content for compression
mock_file = io.BytesIO(original_content)
with patch("time.sleep"), patch("time.perf_counter", side_effect=[0, 1]):
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
bytes([espota2.OTA_VERSION_2_0]), # Version number
bytes(
[espota2.RESPONSE_SUPPORTS_COMPRESSION]
), # Device supports compression
bytes([espota2.RESPONSE_SUPPORTS_COMPRESSION]), # Device supports compression
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
@@ -375,11 +362,8 @@ def test_perform_ota_unsupported_version(mock_socket) -> None:
espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
def test_perform_ota_upload_error(mock_socket) -> None:
def test_perform_ota_upload_error(mock_socket, mock_file, mock_time) -> None:
"""Test OTA handles upload errors."""
mock_file = io.BytesIO(b"firmware")
with patch("time.perf_counter", side_effect=[0, 1]):
# Setup responses - provide enough for the recv calls
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
@@ -394,28 +378,19 @@ def test_perform_ota_upload_error(mock_socket) -> None:
mock_socket.recv.side_effect = recv_responses
with pytest.raises(
espota2.OTAError, match="Error receiving acknowledge chunk OK"
):
with pytest.raises(espota2.OTAError, match="Error receiving acknowledge chunk OK"):
espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
def test_run_ota_impl_successful(mock_socket, tmp_path: Path) -> None:
def test_run_ota_impl_successful(
mock_socket, tmp_path: Path, mock_resolve_ip, mock_perform_ota
) -> None:
"""Test run_ota_impl_ with successful upload."""
# Create a real firmware file
firmware_file = tmp_path / "firmware.bin"
firmware_file.write_bytes(b"firmware content")
with (
patch("socket.socket", return_value=mock_socket),
patch("esphome.espota2.resolve_ip_address") as mock_resolve,
patch("esphome.espota2.perform_ota") as mock_perform,
):
# Setup mocks
mock_resolve.return_value = [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.100", 3232))
]
with patch("socket.socket", return_value=mock_socket):
# Run OTA with real file path
result_code, result_host = espota2.run_ota_impl_(
"test.local", 3232, "password", str(firmware_file)
@@ -431,8 +406,8 @@ def test_run_ota_impl_successful(mock_socket, tmp_path: Path) -> None:
mock_socket.close.assert_called_once()
# Verify perform_ota was called with real file
mock_perform.assert_called_once()
call_args = mock_perform.call_args[0]
mock_perform_ota.assert_called_once()
call_args = mock_perform_ota.call_args[0]
assert call_args[0] == mock_socket
assert call_args[1] == "password"
# The file object should be opened
@@ -440,7 +415,9 @@ def test_run_ota_impl_successful(mock_socket, tmp_path: Path) -> None:
assert call_args[3] == str(firmware_file)
def test_run_ota_impl_connection_failed(mock_socket, tmp_path: Path) -> None:
def test_run_ota_impl_connection_failed(
mock_socket, tmp_path: Path, mock_resolve_ip
) -> None:
"""Test run_ota_impl_ when connection fails."""
mock_socket.connect.side_effect = OSError("Connection refused")
@@ -448,14 +425,7 @@ def test_run_ota_impl_connection_failed(mock_socket, tmp_path: Path) -> None:
firmware_file = tmp_path / "firmware.bin"
firmware_file.write_bytes(b"firmware content")
with (
patch("socket.socket", return_value=mock_socket),
patch("esphome.espota2.resolve_ip_address") as mock_resolve,
):
mock_resolve.return_value = [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.100", 3232))
]
with patch("socket.socket", return_value=mock_socket):
result_code, result_host = espota2.run_ota_impl_(
"test.local", 3232, "password", str(firmware_file)
)
@@ -465,14 +435,13 @@ def test_run_ota_impl_connection_failed(mock_socket, tmp_path: Path) -> None:
mock_socket.close.assert_called_once()
def test_run_ota_impl_resolve_failed(tmp_path: Path) -> None:
def test_run_ota_impl_resolve_failed(tmp_path: Path, mock_resolve_ip) -> None:
"""Test run_ota_impl_ when DNS resolution fails."""
# Create a real firmware file
firmware_file = tmp_path / "firmware.bin"
firmware_file.write_bytes(b"firmware content")
with patch("esphome.espota2.resolve_ip_address") as mock_resolve:
mock_resolve.side_effect = EsphomeError("DNS resolution failed")
mock_resolve_ip.side_effect = EsphomeError("DNS resolution failed")
with pytest.raises(espota2.OTAError, match="DNS resolution failed"):
result_code, result_host = espota2.run_ota_impl_(
@@ -480,16 +449,15 @@ def test_run_ota_impl_resolve_failed(tmp_path: Path) -> None:
)
def test_run_ota_wrapper() -> None:
def test_run_ota_wrapper(mock_run_ota_impl) -> None:
"""Test run_ota wrapper function."""
with patch("esphome.espota2.run_ota_impl_") as mock_impl:
# Test successful case
mock_impl.return_value = (0, "192.168.1.100")
mock_run_ota_impl.return_value = (0, "192.168.1.100")
result = espota2.run_ota("test.local", 3232, "pass", "fw.bin")
assert result == (0, "192.168.1.100")
# Test error case
mock_impl.side_effect = espota2.OTAError("Test error")
mock_run_ota_impl.side_effect = espota2.OTAError("Test error")
result = espota2.run_ota("test.local", 3232, "pass", "fw.bin")
assert result == (1, None)
@@ -528,67 +496,109 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
assert captured.err.count("50%") == 1
# Tests for SHA256 authentication (for when PR is merged)
def test_perform_ota_successful_sha256_auth(mock_socket) -> None:
"""Test successful OTA with SHA256 authentication (future support)."""
# Mock random for predictable cnonce
with patch("random.random", return_value=0.123456):
# Constants for SHA256 auth (when implemented)
RESPONSE_REQUEST_SHA256_AUTH = 0x02 # From PR
# Setup socket responses
responses = [
# Version handshake
bytes([espota2.RESPONSE_OK, espota2.OTA_VERSION_2_0]),
# Features response
bytes([espota2.RESPONSE_HEADER_OK]),
# SHA256 Auth request
bytes([RESPONSE_REQUEST_SHA256_AUTH]),
# Nonce from device (64 chars for SHA256)
b"1234567890123456789012345678901234567890123456789012345678901234",
# Auth result
bytes([espota2.RESPONSE_AUTH_OK]),
# Binary size OK
bytes([espota2.RESPONSE_UPDATE_PREPARE_OK]),
# MD5 checksum OK
bytes([espota2.RESPONSE_BIN_MD5_OK]),
# Chunk OK
bytes([espota2.RESPONSE_CHUNK_OK]),
bytes([espota2.RESPONSE_CHUNK_OK]),
bytes([espota2.RESPONSE_CHUNK_OK]),
# Receive OK
bytes([espota2.RESPONSE_RECEIVE_OK]),
# Update end OK
bytes([espota2.RESPONSE_UPDATE_END_OK]),
# Tests for SHA256 authentication
def test_perform_ota_successful_sha256_auth(
mock_socket, mock_file, mock_time, mock_random
) -> None:
"""Test successful OTA with SHA256 authentication."""
# Setup socket responses for recv calls
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
bytes([espota2.OTA_VERSION_2_0]), # Version number
bytes([espota2.RESPONSE_HEADER_OK]), # Features response
bytes([espota2.RESPONSE_REQUEST_SHA256_AUTH]), # SHA256 Auth request
b"1234567890123456789012345678901234567890123456789012345678901234", # 64 char hex nonce
bytes([espota2.RESPONSE_AUTH_OK]), # Auth result
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 = responses
mock_socket.recv.side_effect = recv_responses
# When SHA256 is implemented, this test will verify:
# 1. Client sends FEATURE_SUPPORTS_SHA256_AUTH flag
# 2. Device responds with RESPONSE_REQUEST_SHA256_AUTH
# 3. Authentication uses SHA256 instead of MD5
# 4. Nonce is 64 characters instead of 32
# Run OTA
espota2.perform_ota(mock_socket, "testpass", mock_file, "test.bin")
# For now, this would raise an error since SHA256 isn't implemented
# Once implemented, uncomment to test:
# espota2.perform_ota(mock_socket, "testpass", mock_file, "test.bin")
# 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)
assert mock_socket.sendall.call_args_list[1] == call(
bytes(
[
espota2.FEATURE_SUPPORTS_COMPRESSION
| espota2.FEATURE_SUPPORTS_SHA256_AUTH
]
)
)
# Verify cnonce was sent (SHA256 of random.random())
cnonce = hashlib.sha256(b"0.123456").hexdigest()
assert mock_socket.sendall.call_args_list[2] == call(cnonce.encode())
# Verify auth result was computed correctly with SHA256
expected_hash = hashlib.sha256()
expected_hash.update(b"testpass")
expected_hash.update(
b"1234567890123456789012345678901234567890123456789012345678901234"
)
expected_hash.update(cnonce.encode())
expected_result = expected_hash.hexdigest()
assert mock_socket.sendall.call_args_list[3] == call(expected_result.encode())
def test_perform_ota_sha256_fallback_to_md5() -> None:
def test_perform_ota_sha256_fallback_to_md5(
mock_socket, mock_file, mock_time, mock_random
) -> None:
"""Test SHA256-capable client falls back to MD5 for compatibility."""
# This test verifies the temporary backward compatibility
# where a SHA256-capable client can still authenticate with MD5
# This compatibility will be removed in 2026.1.0 according to PR
pass # Implementation depends on final PR merge
# This compatibility will be removed in 2026.1.0
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
bytes([espota2.OTA_VERSION_2_0]), # Version number
bytes([espota2.RESPONSE_HEADER_OK]), # Features response
bytes(
[espota2.RESPONSE_REQUEST_AUTH]
), # MD5 Auth request (device doesn't support SHA256)
b"12345678901234567890123456789012", # 32 char hex nonce for MD5
bytes([espota2.RESPONSE_AUTH_OK]), # Auth result
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
# Run OTA - should work even though device requested MD5
espota2.perform_ota(mock_socket, "testpass", mock_file, "test.bin")
# Verify client still advertised SHA256 support
assert mock_socket.sendall.call_args_list[1] == call(
bytes(
[
espota2.FEATURE_SUPPORTS_COMPRESSION
| espota2.FEATURE_SUPPORTS_SHA256_AUTH
]
)
)
# But authentication was done with MD5
cnonce = hashlib.md5(b"0.123456").hexdigest()
expected_hash = hashlib.md5()
expected_hash.update(b"testpass")
expected_hash.update(b"12345678901234567890123456789012")
expected_hash.update(cnonce.encode())
expected_result = expected_hash.hexdigest()
assert mock_socket.sendall.call_args_list[3] == call(expected_result.encode())
def test_perform_ota_version_differences(mock_socket) -> None:
def test_perform_ota_version_differences(mock_socket, mock_file, mock_time) -> None:
"""Test OTA behavior differences between version 1.0 and 2.0."""
mock_file = io.BytesIO(b"firmware")
with patch("time.sleep"), patch("time.perf_counter", side_effect=[0, 1]):
# Test version 1.0 - no chunk acknowledgments
recv_responses = [
bytes([espota2.RESPONSE_OK]), # First byte of version response
@@ -607,3 +617,4 @@ def test_perform_ota_version_differences(mock_socket) -> None:
# Verify no chunk acknowledgments were expected
# (implementation detail - v1 doesn't wait for chunk OK)
assert True # Placeholder assertion