mirror of
https://github.com/esphome/esphome.git
synced 2026-10-04 02:05:59 +00:00
Trim comments
This commit is contained in:
@@ -3,10 +3,9 @@ from tests.testing_helpers import ComponentManifestOverride
|
||||
|
||||
|
||||
def override_manifest(manifest: ComponentManifestOverride) -> None:
|
||||
# to_code must run: it emits the component count the application needs
|
||||
# to_code emits the component count the application needs
|
||||
manifest.enable_codegen()
|
||||
# The deflate decoder lives with the ota platform, which is not part of
|
||||
# this build; only the decoder itself is under test here
|
||||
# Only the decoder is under test; its ota platform is not in this build
|
||||
manifest.resources = manifest.resources + [
|
||||
FileResource("esphome.components.esphome", "ota/ota_esphome_inflate.c"),
|
||||
FileResource("esphome.components.esphome", "ota/ota_esphome_inflate.h"),
|
||||
|
||||
@@ -8,11 +8,7 @@
|
||||
|
||||
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. Regenerate with the
|
||||
// window the CLI uses (espota2.DEFLATE_WINDOW_BITS):
|
||||
// plain = build_plain() written out by the same recipe in Python
|
||||
// build_plain() compressed with the CLI's window (espota2.DEFLATE_WINDOW_BITS):
|
||||
// DEFLATED = zlib.compress(plain, 9, wbits=-12)
|
||||
// STORED = zlib.compress(plain[:300], 0, wbits=-12)
|
||||
static const uint8_t DEFLATED[] = {
|
||||
@@ -185,7 +181,6 @@ static const uint8_t DEFLATED[] = {
|
||||
0x90, 0x3b, 0x14, 0x38, 0xe0, 0x80, 0x03, 0x0e, 0x38, 0xe0, 0x80, 0xfb, 0xff, 0xba, 0xff, 0x00,
|
||||
};
|
||||
|
||||
// The first 300 bytes of the same plaintext as a stored (uncompressed) block
|
||||
static const uint8_t STORED[] = {
|
||||
0x01, 0x2c, 0x01, 0xd3, 0xfe, 0x65, 0x73, 0x70, 0x68, 0x6f, 0x6d, 0x65, 0x20, 0x6f, 0x74, 0x61, 0x20, 0x64,
|
||||
0x65, 0x66, 0x6c, 0x61, 0x74, 0x65, 0x20, 0x65, 0x73, 0x70, 0x68, 0x6f, 0x6d, 0x65, 0x20, 0x6f, 0x74, 0x61,
|
||||
@@ -209,7 +204,6 @@ static const uint8_t STORED[] = {
|
||||
static constexpr size_t WINDOW = 4096;
|
||||
static constexpr size_t PLAIN_SIZE = 16000;
|
||||
|
||||
// Pseudo random bytes reproducible from Python for the vectors above
|
||||
static uint8_t lcg_next(uint32_t &x) {
|
||||
x = (x * 1103515245u + 12345u) & 0x7fffffffu;
|
||||
return (x >> 16) & 0xff;
|
||||
@@ -229,8 +223,7 @@ static std::vector<uint8_t> build_plain() {
|
||||
return plain;
|
||||
}
|
||||
|
||||
// Mirrors the OTA session: the state is the base, input arrives through the
|
||||
// read callback in chunks, the window doubles as the output buffer.
|
||||
// Mirrors the OTA session: chunked input through the read callback, window as output
|
||||
struct Session : OtaInflateState {
|
||||
const uint8_t *in;
|
||||
size_t in_len;
|
||||
@@ -310,9 +303,7 @@ TEST(OtaInflate, TruncatedStoredBlockFails) {
|
||||
}
|
||||
|
||||
TEST(OtaInflate, CorruptStreamsNeverEscapeTheWindow) {
|
||||
// Every third byte of the stream flipped in turn, plus pseudo random garbage:
|
||||
// the sanitizers check that the decoder never reads or writes out of bounds
|
||||
// whatever it returns.
|
||||
// Flipped bytes and garbage; the sanitizers check the decoder stays in bounds
|
||||
auto s = std::make_unique<Session>();
|
||||
std::vector<uint8_t> bad(DEFLATED, DEFLATED + sizeof(DEFLATED));
|
||||
for (size_t i = 0; i < bad.size(); i += 3) {
|
||||
|
||||
@@ -185,7 +185,7 @@ async def test_host_ota_self_update(
|
||||
def on_log(line: str) -> None:
|
||||
if "OTA staged at" in line:
|
||||
staged.set()
|
||||
# The host backend has no gzip support, so the upload negotiates deflate
|
||||
# The host backend cannot store gzip, so the upload negotiates deflate
|
||||
if "Inflated " in line and " bytes from " in line:
|
||||
inflated.set()
|
||||
dev.on_log(line)
|
||||
|
||||
@@ -601,10 +601,7 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N
|
||||
|
||||
|
||||
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.
|
||||
|
||||
With server_features the device answers with the extended feature flags.
|
||||
"""
|
||||
"""Recv responses for a handshake without auth, up to the MD5 check."""
|
||||
if server_features is None:
|
||||
features = [bytes([espota2.RESPONSE_HEADER_OK])]
|
||||
else:
|
||||
@@ -1532,13 +1529,13 @@ _UPLOAD_TAIL = [
|
||||
"server_features",
|
||||
[
|
||||
espota2.SERVER_FEATURE_SUPPORTS_DEFLATE,
|
||||
# A deflate offer is binding, so it wins should a device set both bits
|
||||
# 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:
|
||||
"""A device that inflates on the fly gets a raw deflate stream, both sizes and the image MD5."""
|
||||
"""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
|
||||
@@ -1552,6 +1549,5 @@ def test_perform_ota_with_deflate(mock_socket: Mock, server_features: int) -> No
|
||||
assert sent[4] == len(original_content).to_bytes(espota2.SIZE_FIELD_BYTES, "big")
|
||||
payload = sent[6]
|
||||
assert len(payload) == sent_size < len(original_content)
|
||||
# The device decodes through a window of 1 << DEFLATE_WINDOW_BITS bytes
|
||||
assert zlib.decompress(payload, -espota2.DEFLATE_WINDOW_BITS) == original_content
|
||||
assert sent[5] == hashlib.md5(original_content).hexdigest().encode()
|
||||
|
||||
Reference in New Issue
Block a user