address bot nits

esphome/espota2.py

@@ -59,6 +59,11 @@ CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04
 SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01
 SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02
 
+# OTA types this client knows how to send. Future PRs that add bootloader/partition
+# updates extend this set. Anything outside the set is rejected up front so callers
+# of perform_ota/run_ota get a clear error instead of a post-auth 0x8E from the device.
+_SUPPORTED_OTA_TYPES: frozenset[int] = frozenset({OTA_TYPE_UPDATE_APP})
+
 UPLOAD_BLOCK_SIZE = 8192
 UPLOAD_BUFFER_SIZE = UPLOAD_BLOCK_SIZE * 8
 
@@ -192,8 +197,10 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None
     :param expect: Expected response code(s), None to skip validation.
     :raises OTAError: If an error code is detected or response doesn't match expected.
     """
-    if expect is None:
-        return
+    # Detect device errors and connection-closed cases regardless of `expect`. If we
+    # only ran these checks when expect was set, error bytes returned during
+    # accept-any-response reads (e.g. feature negotiation, auth nonces) would be
+    # silently passed through and surface later as cryptic decode/timeout failures.
     if not data:
         raise OTAError(
             "Error: Device closed connection without responding. "
@@ -204,6 +211,8 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None
     error_msg = _ERROR_MESSAGES.get(dat)
     if error_msg is not None:
         raise OTAError(f"Error: {error_msg}")
+    if expect is None:
+        return
     if not isinstance(expect, (list, tuple)):
         expect = [expect]
     if dat not in expect:
@@ -240,6 +249,19 @@ def perform_ota(
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
 ) -> None:
+    # Validate ota_type up front. It travels as a single byte on the wire, and
+    # passing an out-of-range value would only surface as a ValueError from
+    # bytes([ota_type]) deep inside send_check, bypassing OTAError handling.
+    if not isinstance(ota_type, int) or not 0 <= ota_type <= 0xFF:
+        raise OTAError(
+            f"Invalid ota_type {ota_type!r}; expected an integer in range 0-255"
+        )
+    if ota_type not in _SUPPORTED_OTA_TYPES:
+        supported = ", ".join(f"0x{t:02X}" for t in sorted(_SUPPORTED_OTA_TYPES))
+        raise OTAError(
+            f"Unsupported OTA type 0x{ota_type:02X}; this ESPHome supports: {supported}"
+        )
+
     file_contents = file_handle.read()
     file_size = len(file_contents)
     _LOGGER.info("Uploading %s (%s bytes)", filename, file_size)

tests/unit_tests/test_espota2.py

@@ -873,10 +873,48 @@ def test_perform_ota_device_rejects_with_unsupported_ota_type(
 
 
 @pytest.mark.usefixtures("mock_time")
-def test_perform_ota_non_app_type_requires_extended_protocol(
+def test_perform_ota_unsupported_type_rejected_early(
     mock_socket: Mock, mock_file: io.BytesIO
+) -> None:
+    """ota_type values not in _SUPPORTED_OTA_TYPES are rejected before any I/O."""
+    with pytest.raises(espota2.OTAError, match="Unsupported OTA type 0xFF"):
+        espota2.perform_ota(
+            mock_socket,
+            "testpass",
+            mock_file,
+            "test.bin",
+            0xFF,
+        )
+    # No bytes should have been transmitted to the device.
+    mock_socket.sendall.assert_not_called()
+
+
+@pytest.mark.parametrize("bad_type", [-1, 256, 0x10000, "app", None, 1.5])
+def test_perform_ota_rejects_out_of_range_type(
+    mock_socket: Mock, mock_file: io.BytesIO, bad_type: object
+) -> None:
+    """Out-of-range or non-int ota_type must raise OTAError, not ValueError."""
+    with pytest.raises(espota2.OTAError, match="Invalid ota_type"):
+        espota2.perform_ota(
+            mock_socket,
+            "testpass",
+            mock_file,
+            "test.bin",
+            bad_type,  # type: ignore[arg-type]
+        )
+    mock_socket.sendall.assert_not_called()
+
+
+@pytest.mark.usefixtures("mock_time")
+def test_perform_ota_non_app_type_requires_extended_protocol(
+    mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """Non-app OTA type must fail when device only supports the legacy protocol."""
+    monkeypatch.setattr(
+        espota2,
+        "_SUPPORTED_OTA_TYPES",
+        frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}),
+    )
     recv_responses = [
         bytes([espota2.RESPONSE_OK]),  # First byte of version response
         bytes([espota2.OTA_VERSION_2_0]),  # Version number
@@ -893,16 +931,21 @@ def test_perform_ota_non_app_type_requires_extended_protocol(
             "testpass",
             mock_file,
             "test.bin",
-            255,
+            0xFF,
         )
 
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_non_app_type_requires_partition_access(
-    mock_socket: Mock, mock_file: io.BytesIO
+    mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """Non-app OTA type must fail when device advertises extended protocol but
     not the partition-access feature."""
+    monkeypatch.setattr(
+        espota2,
+        "_SUPPORTED_OTA_TYPES",
+        frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}),
+    )
     recv_responses = [
         bytes([espota2.RESPONSE_OK]),  # First byte of version response
         bytes([espota2.OTA_VERSION_2_0]),  # Version number
@@ -922,5 +965,31 @@ def test_perform_ota_non_app_type_requires_partition_access(
             "testpass",
             mock_file,
             "test.bin",
-            255,
+            0xFF,
         )
+
+
+def test_check_error_detects_errors_when_expect_is_none() -> None:
+    """check_error must surface device error bytes even when expect is None.
+
+    Regression test: previously, receive_exactly(..., expect=None) calls (used
+    during feature negotiation and nonce reads) silently passed error bytes
+    through, turning clean device errors into confusing later failures.
+    """
+    with pytest.raises(espota2.OTAError, match="Error: Authentication invalid"):
+        espota2.check_error([espota2.RESPONSE_ERROR_AUTH_INVALID], None)
+
+
+def test_check_error_detects_empty_when_expect_is_none() -> None:
+    """Empty data with expect=None must still raise (connection closed)."""
+    with pytest.raises(
+        espota2.OTAError, match="Device closed connection without responding"
+    ):
+        espota2.check_error([], None)
+
+
+def test_check_error_passes_non_error_when_expect_is_none() -> None:
+    """Non-error bytes with expect=None must pass through silently."""
+    espota2.check_error([espota2.RESPONSE_OK], None)
+    espota2.check_error([espota2.RESPONSE_HEADER_OK], None)
+    espota2.check_error([espota2.RESPONSE_FEATURE_FLAGS], None)