diff --git a/esphome/espota2.py b/esphome/espota2.py index d9b09ef534..b9553e9237 100644 --- a/esphome/espota2.py +++ b/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 diff --git a/tests/components/esphome/test_ota_inflate.cpp b/tests/components/esphome/test_ota_inflate.cpp index 1beb99fa59..a2ad39abba 100644 --- a/tests/components/esphome/test_ota_inflate.cpp +++ b/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, diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index a9b9d30943..1ee954981c 100644 --- a/tests/unit_tests/test_espota2.py +++ b/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