[esphome][ota][core] Compress OTA uploads on ESP32, RP2040, LibreTiny and host (#19037)

tests/unit_tests/test_espota2.py

@@ -12,6 +12,7 @@ from pathlib import Path
 import socket
 import struct
 from unittest.mock import Mock, call, patch
+import zlib
 
 import pytest
 from pytest import CaptureFixture
@@ -354,6 +355,7 @@ def test_perform_ota_successful_md5_auth(
                 espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
                 | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
                 | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+                | espota2.CLIENT_FEATURE_SUPPORTS_DEFLATE
             ]
         )
     )
@@ -601,12 +603,16 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N
         espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
-def _no_auth_handshake(version: int) -> list[bytes]:
+def _no_auth_handshake(version: int, server_features: int | None = None) -> list[bytes]:
     """Recv responses for a handshake without auth, up to the MD5 check."""
+    if server_features is None:
+        features = [bytes([espota2.RESPONSE_HEADER_OK])]
+    else:
+        features = [bytes([espota2.RESPONSE_FEATURE_FLAGS]), bytes([server_features])]
     return [
         bytes([espota2.RESPONSE_OK]),  # First byte of version response
         bytes([version]),  # Version number
-        bytes([espota2.RESPONSE_HEADER_OK]),  # Features response
+        *features,
         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
@@ -1054,6 +1060,7 @@ def test_perform_ota_successful_sha256_auth(
                 espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
                 | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
                 | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+                | espota2.CLIENT_FEATURE_SUPPORTS_DEFLATE
             ]
         )
     )
@@ -1110,6 +1117,7 @@ def test_perform_ota_sha256_fallback_to_md5(
                 espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
                 | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
                 | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+                | espota2.CLIENT_FEATURE_SUPPORTS_DEFLATE
             ]
         )
     )
@@ -1219,6 +1227,7 @@ def test_perform_ota_extended_protocol_app(
                 espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
                 | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
                 | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+                | espota2.CLIENT_FEATURE_SUPPORTS_DEFLATE
             ]
         )
     )
@@ -1279,6 +1288,7 @@ def test_perform_ota_successful_partition_table(
                 espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION
                 | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
                 | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+                | espota2.CLIENT_FEATURE_SUPPORTS_DEFLATE
             ]
         )
     )
@@ -1507,3 +1517,40 @@ def test_check_error_passes_non_error_when_expect_is_none() -> None:
     espota2.check_error([espota2.RESPONSE_OK], None)
     espota2.check_error([espota2.RESPONSE_HEADER_OK], None)
     espota2.check_error([espota2.RESPONSE_FEATURE_FLAGS], None)
+
+
+# Device replies after the MD5 check for a one-chunk upload
+_UPLOAD_TAIL = [
+    bytes([espota2.RESPONSE_CHUNK_OK]),
+    bytes([espota2.RESPONSE_RECEIVE_OK]),
+    bytes([espota2.RESPONSE_UPDATE_END_OK]),
+]
+
+
+@pytest.mark.usefixtures("mock_time")
+@pytest.mark.parametrize(
+    "server_features",
+    [
+        espota2.SERVER_FEATURE_SUPPORTS_DEFLATE,
+        # Binding offer: deflate wins over gzip
+        espota2.SERVER_FEATURE_SUPPORTS_DEFLATE
+        | espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION,
+    ],
+)
+def test_perform_ota_with_deflate(mock_socket: Mock, server_features: int) -> None:
+    """The device gets a raw deflate stream, both sizes and the image MD5."""
+    original_content = b"firmware" * 100
+    mock_socket.recv.side_effect = (
+        _no_auth_handshake(espota2.OTA_VERSION_2_0, server_features) + _UPLOAD_TAIL
+    )
+
+    espota2.perform_ota(mock_socket, None, io.BytesIO(original_content), "test.bin")
+
+    sent = [c[0][0] for c in mock_socket.sendall.call_args_list]
+    # magic, features, ota type, size, image size, md5, data, end ack
+    sent_size = struct.unpack(">I", sent[3])[0]
+    assert sent[4] == len(original_content).to_bytes(espota2.SIZE_FIELD_BYTES, "big")
+    payload = sent[6]
+    assert len(payload) == sent_size < len(original_content)
+    assert zlib.decompress(payload, -espota2.DEFLATE_WINDOW_BITS) == original_content
+    assert sent[5] == hashlib.md5(original_content).hexdigest().encode()