Honour a deflate offer over gzip on the client and record how the test vectors are made

esphome/espota2.py

@@ -655,17 +655,18 @@ def perform_ota(
             )
 
     deflate = False
-    if features & SERVER_FEATURE_SUPPORTS_COMPRESSION:
-        # The device stores the gzip file and inflates it when it reboots
-        upload_contents = gzip.compress(file_contents, compresslevel=COMPRESS_LEVEL)
-        _LOGGER.info("Compressed to %s bytes", len(upload_contents))
-    elif extended_proto and features & SERVER_FEATURE_SUPPORTS_DEFLATE:
-        # The device inflates while receiving through a small ring window
+    if extended_proto and features & SERVER_FEATURE_SUPPORTS_DEFLATE:
+        # The device inflates while receiving through a small ring window; the
+        # offer is binding, so it wins over gzip should a device set both bits
         upload_contents = zlib.compress(
             file_contents, COMPRESS_LEVEL, wbits=-DEFLATE_WINDOW_BITS
         )
         deflate = True
         _LOGGER.info("Compressed to %s bytes (deflate)", len(upload_contents))
+    elif features & SERVER_FEATURE_SUPPORTS_COMPRESSION:
+        # The device stores the gzip file and inflates it when it reboots
+        upload_contents = gzip.compress(file_contents, compresslevel=COMPRESS_LEVEL)
+        _LOGGER.info("Compressed to %s bytes", len(upload_contents))
     else:
         upload_contents = file_contents
 

tests/components/esphome/test_ota_inflate.cpp

@@ -10,7 +10,11 @@ namespace esphome::testing {
 
 // The plaintext below, as built by build_plain(): repeated text, a pseudo random
 // run, a zero run and the text again, so literals, short and long back references
-// and stored data are all exercised across the 4 KB window.
+// and stored data are all exercised across the 4 KB window. Regenerate with the
+// window the CLI uses (espota2.DEFLATE_WINDOW_BITS):
+//   plain = build_plain() written out by the same recipe in Python
+//   DEFLATED = zlib.compress(plain, 9, wbits=-12)
+//   STORED = zlib.compress(plain[:300], 0, wbits=-12)
 static const uint8_t DEFLATED[] = {
     0xed, 0xc8, 0xf7, 0x3f, 0xd4, 0x0f, 0x03, 0x00, 0x70, 0x67, 0xaf, 0x4b, 0x67, 0x66, 0x9f, 0x90, 0x91, 0x11, 0xc2,
     0x11, 0x91, 0xb8, 0xb3, 0xf7, 0x3a, 0xd9, 0x5f, 0x4e, 0x99, 0x67, 0x1e, 0xce, 0x8a, 0xac, 0xec, 0x59, 0xb8, 0xc2,

tests/unit_tests/test_espota2.py

@@ -1552,8 +1552,8 @@ def test_perform_ota_with_deflate(mock_socket: Mock) -> None:
 
 
 @pytest.mark.usefixtures("mock_time")
-def test_perform_ota_gzip_wins_over_deflate(mock_socket: Mock) -> None:
-    """A device that can store gzip keeps getting gzip even when it also offers deflate."""
+def test_perform_ota_deflate_wins_over_gzip(mock_socket: Mock) -> None:
+    """A deflate offer is binding on the device, so it wins should a device set both bits."""
     original_content = b"firmware" * 100
     mock_socket.recv.side_effect = (
         _no_auth_handshake(
@@ -1567,8 +1567,5 @@ def test_perform_ota_gzip_wins_over_deflate(mock_socket: Mock) -> None:
     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]
-    # gzip output embeds the time of compression, so compare what it holds
-    payload = sent[5]
-    assert gzip.decompress(payload) == original_content
-    assert sent[3] == len(payload).to_bytes(4, "big")
-    assert sent[4] == hashlib.md5(payload).hexdigest().encode()
+    assert sent[4] == len(original_content).to_bytes(4, "big")
+    assert zlib.decompress(sent[6], -espota2.DEFLATE_WINDOW_BITS) == original_content