From 83bc96cbed7ef4ed3822be4d5a9ef4e477244e32 Mon Sep 17 00:00:00 2001 From: gabriel Date: Wed, 26 Aug 2026 14:47:39 +0200 Subject: [PATCH 1/3] feat: add host-side helpers for wire encodings, keys and sleep timing Hosts driving OpenDisplay devices have had to reimplement pieces of the protocol the library already knew: packing an LED colour, sizing an NDEF record, formatting a firmware version to match its release tag, parsing a stored encryption key, and working out when a sleeping device is reachable. Every reimplementation is somewhere a host can drift from the firmware. All additive; no existing signature changes meaning. - pack_led_color / unpack_led_color / ms_to_loop_delay_units / ms_to_inter_delay_units, plus LedFlashStep.from_rgb() and .rgb so callers can work in RGB and milliseconds instead of the packed wire encoding - build_nfc_payload(), the same payload write_nfc_* sends, so a record can be validated before a connection is spent on it - format_firmware_version(), the single definition of how a version maps to a release tag. A host that formats it differently from the tag it compares against shows a permanently pending update - supports_ble_ota_install(), deliberately narrower than "an asset exists": it answers whether a host should offer the install over a Bluetooth proxy, which nRF Legacy DFU cannot survive - parse_encryption_key(), accepting a key however it was pasted, raising a typed InvalidEncryptionKeyError - BinaryInputs.enabled_button_ids and DisplayConfig.canvas_size() - SleepModel: the post-wake window and whether a device has gone dark again, with host-side slack passed in rather than assumed - prepare_image(compress=None) derives compression from the device config Internal call sites now use these too, so the library no longer keeps its own copies of the NFC payload assembly or the compression capability check. Verified against firmware upstream/main: the 10 s default wake window is DEFAULT_IDLE_HOLD_MS in src/main.h, and the sleep-entry condition is the one in platformIdle(). The window is re-armed by activity and held open by a button wake, so probably_asleep() is documented as a prediction that errs towards asleep rather than a fact. --- src/opendisplay/__init__.py | 44 ++++++- src/opendisplay/crypto.py | 44 +++++++ src/opendisplay/device.py | 38 +++--- src/opendisplay/exceptions.py | 13 ++ src/opendisplay/models/config.py | 56 +++++++++ src/opendisplay/models/firmware.py | 46 +++++++ src/opendisplay/models/led_flash.py | 99 +++++++++++++++ src/opendisplay/protocol/__init__.py | 4 + src/opendisplay/protocol/commands.py | 59 +++++++++ src/opendisplay/sleep.py | 119 +++++++++++++++++++ tests/conftest.py | 60 ++++++++++ tests/unit/test_crypto.py | 71 +++++++++++ tests/unit/test_device_nfc_write.py | 87 ++++++++++++++ tests/unit/test_device_upload_compression.py | 38 +++++- tests/unit/test_models_config.py | 98 +++++++++++++++ tests/unit/test_models_firmware.py | 67 +++++++++++ tests/unit/test_models_led_flash.py | 113 +++++++++++++++++- tests/unit/test_sleep_model.py | 89 ++++++++++++++ 18 files changed, 1123 insertions(+), 22 deletions(-) create mode 100644 src/opendisplay/sleep.py create mode 100644 tests/unit/test_models_firmware.py create mode 100644 tests/unit/test_sleep_model.py diff --git a/src/opendisplay/__init__.py b/src/opendisplay/__init__.py index c7c8a11..ca09ff2 100644 --- a/src/opendisplay/__init__.py +++ b/src/opendisplay/__init__.py @@ -6,6 +6,7 @@ from epaper_dithering import ColorScheme, DitherMode from .battery import voltage_to_percent +from .crypto import KEY_LENGTH_BYTES, KEY_LENGTH_HEX, parse_encryption_key from .device import OpenDisplayDevice, prepare_image from .discovery import discover_devices, discover_devices_with_adv from .discovery_ip import IpDeviceInfo, discover_ip_devices @@ -18,6 +19,7 @@ ConfigParseError, ImageEncodingError, IntegrityCheckError, + InvalidEncryptionKeyError, InvalidResponseError, NfcNotSupportedError, NfcWriteError, @@ -88,12 +90,32 @@ get_board_type_name, get_manufacturer_name, ) -from .models.firmware import firmware_ota_asset, firmware_release_repo -from .models.led_flash import LedFlashConfig, LedFlashStep +from .models.firmware import ( + firmware_ota_asset, + firmware_release_repo, + format_firmware_version, + supports_ble_ota_install, +) +from .models.led_flash import ( + DELAY_UNIT_MS, + LedFlashConfig, + LedFlashStep, + ms_to_inter_delay_units, + ms_to_loop_delay_units, + pack_led_color, + unpack_led_color, +) from .ota import find_nrf_dfu_device, perform_nrf_dfu, perform_silabs_ota from .partial import PartialState -from .protocol import MANUFACTURER_ID, SERVICE_UUID +from .protocol import ( + MANUFACTURER_ID, + NFC_MIME_TYPE_MAX, + NFC_WRITE_MAX_TOTAL, + SERVICE_UUID, + build_nfc_payload, +) from .sensors import SensorReading, read_sensor_values +from .sleep import DEFAULT_WAKE_WINDOW_MS, SleepModel from .transport import BleTransport, TcpTransport, Transport __version__ = "0.1.0" @@ -127,6 +149,7 @@ "InvalidResponseError", "ImageEncodingError", "IntegrityCheckError", + "InvalidEncryptionKeyError", "NfcNotSupportedError", "NfcWriteError", "OTAError", @@ -147,7 +170,14 @@ "note_to_index", "LedFlashConfig", "LedFlashStep", + "DELAY_UNIT_MS", + "pack_led_color", + "unpack_led_color", + "ms_to_loop_delay_units", + "ms_to_inter_delay_units", "firmware_ota_asset", + "format_firmware_version", + "supports_ble_ota_install", "firmware_release_repo", "SensorData", "SensorReading", @@ -207,4 +237,12 @@ # Constants "SERVICE_UUID", "MANUFACTURER_ID", + "NFC_MIME_TYPE_MAX", + "NFC_WRITE_MAX_TOTAL", + "build_nfc_payload", + "SleepModel", + "DEFAULT_WAKE_WINDOW_MS", + "KEY_LENGTH_BYTES", + "KEY_LENGTH_HEX", + "parse_encryption_key", ] diff --git a/src/opendisplay/crypto.py b/src/opendisplay/crypto.py index 1e8a3c4..21508ad 100644 --- a/src/opendisplay/crypto.py +++ b/src/opendisplay/crypto.py @@ -7,17 +7,61 @@ from __future__ import annotations import os +from typing import Final from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.ciphers.aead import AESCCM from cryptography.hazmat.primitives.cmac import CMAC +from .exceptions import InvalidEncryptionKeyError + # Firmware placeholder device ID (hardcoded in firmware, never changes) _DEVICE_ID = bytes([0x00, 0x00, 0x00, 0x01]) # CCM auth tag length used by firmware _TAG_LEN = 12 +#: Length of the AES-128 master key, in bytes. +KEY_LENGTH_BYTES: Final = 16 +#: Length of the same key written as hex, which is how hosts store it. +KEY_LENGTH_HEX: Final = KEY_LENGTH_BYTES * 2 + + +def parse_encryption_key(raw: str | None) -> bytes | None: + """Parse a stored hex encryption key into the bytes the device API expects. + + ``raw`` is the form a host persists or a user pastes: 32 hex characters for + the 16-byte AES-128 master key, or None when the device is unencrypted. + + Normalization is deliberately forgiving, because a key is something a human + copies between a config tool, a shell and a settings field. Case is not + significant, and spaces and colons are separators rather than content, so + ``AA:BB:CC...``, ``aa bb cc...`` and ``aabbcc...`` are the same key. + + This is the single definition of the stored-key format. Hosts that + reimplement it drift from the library and from each other, and each such copy + is a place where a malformed key produces a different, less useful error. + + Returns: + The 16 key bytes, or None if ``raw`` is None. + + Raises: + InvalidEncryptionKeyError: If the key is the wrong length or not hex. + An empty string raises rather than being read as "no key" - absence + is expressed by None, so an empty value is a malformed key. + """ + if raw is None: + return None + candidate = raw.strip().replace(" ", "").replace(":", "") + if len(candidate) != KEY_LENGTH_HEX: + raise InvalidEncryptionKeyError( + f"encryption key must be {KEY_LENGTH_HEX} hex characters ({KEY_LENGTH_BYTES} bytes), got {len(candidate)}" + ) + try: + return bytes.fromhex(candidate) + except ValueError as err: + raise InvalidEncryptionKeyError("encryption key is not valid hexadecimal") from err + def aes_cmac(key: bytes, data: bytes) -> bytes: """Compute AES-128-CMAC.""" diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index cd5d494..0684f7e 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -100,6 +100,7 @@ build_direct_write_start_uncompressed, build_enter_dfu_command, build_led_activate_command, + build_nfc_payload, build_nfc_write_data_command, build_nfc_write_end_command, build_nfc_write_inline_command, @@ -280,7 +281,7 @@ def prepare_image( use_measured_palettes: bool = True, panel_ic_type: int | None = None, dither_mode: DitherMode = DitherMode.BURKES, - compress: bool = True, + compress: bool | None = True, serpentine: bool = True, exposure: float = 1.0, saturation: float = 1.0, @@ -306,7 +307,10 @@ def prepare_image( panel_ic_type: Panel IC type for palette lookup. If None, extracted from config. dither_mode: Dithering algorithm to use (default: BURKES) - compress: Whether to compress the image data (default: True) + compress: Whether to compress the image data (default: True). Pass None + to derive it from ``config``, matching what upload_image() does: + compress only when the panel advertises ZIP or streaming + decompression. serpentine: Alternate scan direction each row to reduce artifacts (default: True) exposure: Exposure multiplier, >1.0 brightens (default: 1.0) saturation: Saturation multiplier, >1.0 boosts (default: 1.0) @@ -385,7 +389,14 @@ def prepare_image( else: image_data = encode_image(dithered, color_scheme) - # Optionally compress + # Optionally compress. compress=None means "ask the config", which is what + # upload_image() does internally; a caller preparing an image up front (to + # move the CPU work off an event loop, say) would otherwise have to + # reimplement this check and keep it in sync by hand. + if compress is None: + display_cfg = config.displays[0] if config is not None and config.displays else None + compress = display_cfg.supports_compression if display_cfg else True + compressed_data = None if compress: # Current firmware compiles uzlib with a 9-bit window and hard-rejects any @@ -1548,7 +1559,7 @@ async def write_nfc_url(self, url: str, timeout: float | None = None) -> None: url: URL to write. timeout: Optional override; see write_nfc. """ - await self.write_nfc(NfcRecordType.URI, url.encode("utf-8"), timeout) + await self.write_nfc(NfcRecordType.URI, build_nfc_payload(NfcRecordType.URI, url), timeout) async def write_nfc_text(self, text: str, timeout: float | None = None) -> None: """Write a TEXT NDEF record containing the given text. @@ -1559,7 +1570,7 @@ async def write_nfc_text(self, text: str, timeout: float | None = None) -> None: text: Text to write. timeout: Optional override; see write_nfc. """ - await self.write_nfc(NfcRecordType.TEXT, text.encode("utf-8"), timeout) + await self.write_nfc(NfcRecordType.TEXT, build_nfc_payload(NfcRecordType.TEXT, text), timeout) async def write_nfc_mime( self, @@ -1580,13 +1591,10 @@ async def write_nfc_mime( timeout: Optional override; see write_nfc. Raises: - ValueError: If the encoded MIME type is outside 1..255 bytes. + ValueError: If the encoded MIME type is outside 1..255 bytes, or the + assembled payload exceeds NFC_WRITE_MAX_TOTAL. """ - mt = mime_type.encode("utf-8") - if not 1 <= len(mt) <= 255: - raise ValueError(f"mime_type must encode to 1..255 bytes, got {len(mt)}") - body_bytes = body.encode("utf-8") if isinstance(body, str) else body - payload = bytes([len(mt)]) + mt + body_bytes + payload = build_nfc_payload(NfcRecordType.MIME, body, mime_type) await self.write_nfc(NfcRecordType.MIME, payload, timeout) @_serialized @@ -1799,9 +1807,7 @@ async def upload_image( # compressed uploads either way (<=1.81 NACKs without the ZIP bit and # the upload falls back to uncompressed). display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None - supports_compression = ( - (display_cfg.supports_zip or display_cfg.supports_streaming_decompression) if display_cfg else True - ) + supports_compression = display_cfg.supports_compression if display_cfg else True # When a partial upload may succeed, defer full-frame compression: it is # pure waste if the partial path handles the update. _dispatch_upload @@ -1919,9 +1925,7 @@ async def _dispatch_upload( sent), False if the firmware auto-completed the upload. """ display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None - supports_compression = ( - (display_cfg.supports_zip or display_cfg.supports_streaming_decompression) if display_cfg else True - ) + supports_compression = display_cfg.supports_compression if display_cfg else True streaming_decompression = bool(display_cfg and display_cfg.supports_streaming_decompression) if ( compress diff --git a/src/opendisplay/exceptions.py b/src/opendisplay/exceptions.py index 4b7782c..8ec0ab8 100644 --- a/src/opendisplay/exceptions.py +++ b/src/opendisplay/exceptions.py @@ -159,6 +159,19 @@ def __init__(self, message: str = "Device may not support NFC write (no response super().__init__(message) +class InvalidEncryptionKeyError(OpenDisplayError): + """A stored encryption key is not a valid AES-128 key. + + Raised by ``parse_encryption_key`` for a key that is the wrong length or is + not hexadecimal. Deliberately *not* an ``AuthenticationError``: nothing has + been sent to a device, so this is a local configuration fault rather than a + rejection by the device, and a host should treat it as "ask the user for the + key again" rather than "the device said no". + """ + + pass + + class ImageEncodingError(OpenDisplayError): """Failed to encode image.""" diff --git a/src/opendisplay/models/config.py b/src/opendisplay/models/config.py index 73c3891..2f9d790 100644 --- a/src/opendisplay/models/config.py +++ b/src/opendisplay/models/config.py @@ -342,6 +342,21 @@ def supports_zip(self) -> bool: """ return bool(self.transmission_modes & 0x02) + @property + def supports_compression(self) -> bool: + """Whether the panel accepts a compressed upload by either mechanism. + + Post-2.0 configs may advertise only streaming decompression (bit 0x01, + historically ZIPXL) without the plain ZIP bit; pre-2.0 configs may + advertise only ZIP. Firmware 2.0 accepts a compressed upload either way, + and <= 1.81 NACKs one without the ZIP bit so the upload falls back to + uncompressed. + + This is the question every upload path actually asks, so it lives here + rather than being spelled out at each call site. + """ + return self.supports_zip or self.supports_streaming_decompression + @property def supports_g5(self) -> bool: """Check if display supports Group 5 compression (TRANSMISSION_MODE_G5).""" @@ -404,6 +419,34 @@ def rotation_enum(self) -> Rotation | int: except ValueError: return _INDEX_TO_ROTATION.get(self.rotation, self.rotation) + def canvas_size(self, extra_rotation: Rotation | int = Rotation.ROTATE_0) -> tuple[int, int]: + """Return the (width, height) a source image should be authored at. + + The device applies its configured ``rotation`` on top of any rotation the + caller asks for, then fits the result to the panel's native pixel grid. + When the combined rotation transposes the axes (90 or 270 degrees), a + canvas drawn at the panel's own width x height has the wrong aspect ratio + and the device-side fit scales or letterboxes it. Drawing at the + transposed size instead makes that fit a 1:1 no-op. + + Rotation itself is left to the device; this only answers what shape to + draw. An unknown stored rotation is treated as 0 degrees. + + Args: + extra_rotation: Additional rotation the caller will request, as a + ``Rotation`` or as degrees. + + Returns: + (width, height) in pixels: the panel's own dimensions, or those + swapped when the effective rotation is 90 or 270 degrees. + """ + base = self.rotation_enum + base_deg = base.value if isinstance(base, Rotation) else 0 + extra_deg = extra_rotation.value if isinstance(extra_rotation, Rotation) else int(extra_rotation) + if (base_deg + extra_deg) % 360 in (90, 270): + return self.pixel_height, self.pixel_width + return self.pixel_width, self.pixel_height + SIZE: ClassVar[int] = 46 @classmethod @@ -634,6 +677,19 @@ def published_button_byte_index(self) -> int | None: return None return self.button_data_byte_index + @property + def enabled_button_ids(self) -> tuple[int, ...]: + """Button ids this input actually has fitted, in ascending order. + + ``input_flags`` is a bitmask over the 8 pin slots: bit N set means slot N + carries a button. The bit position *is* the button id reported in the + advertisement's button byte, so this is also the set of ids a consumer + should expect to see events for. + + Returns an empty tuple when no slots are populated. + """ + return tuple(bit for bit in range(self.MAX_BUTTON_ID + 1) if self.input_flags & (1 << bit)) + @classmethod def adc_ladder( cls, diff --git a/src/opendisplay/models/firmware.py b/src/opendisplay/models/firmware.py index 1d15030..8b12f6e 100644 --- a/src/opendisplay/models/firmware.py +++ b/src/opendisplay/models/firmware.py @@ -35,6 +35,52 @@ def firmware_ota_asset(ic_type: int, tag: str) -> str | None: return None +# BLE OTA install is only advertised for ICs where the flash completes reliably +# over an ESPHome Bluetooth proxy, which is the common deployment. EFR32BG22 +# (Silabs AppLoader) does. nRF Legacy DFU does NOT: verified end to end, the +# device receives the full, CRC-valid image but the final activate/commit write +# is unreliable over a proxy and strands the device in the bootloader. It works +# over a *direct* connection, so nRF firmware must be flashed directly or via +# USB-UF2. Note this is narrower than firmware_ota_asset(), which answers "is +# there an asset for this IC" rather than "should a host offer to install it". +_BLE_OTA_INSTALL_IC_TYPES: Final[frozenset[int]] = frozenset({ICType.EFR32BG22}) + + +def supports_ble_ota_install(ic_type: int) -> bool: + """Return True if a host should offer a BLE OTA install for this IC type. + + Deliberately stricter than ``firmware_ota_asset`` being non-None: an asset + can exist for an IC whose over-the-proxy install path is not dependable + (nRF Legacy DFU), where offering the install strands the device in its + bootloader. Hosts that only surface release notes should use + ``firmware_release_repo`` instead. + """ + return ic_type in _BLE_OTA_INSTALL_IC_TYPES + + +def format_firmware_version(major: int, minor: int, patch: int | None = None) -> str: + """Format a firmware version to match the GitHub release tag convention. + + Firmware parses its own BUILD_VERSION string with a plain int conversion + (``atoi`` on the substring after the dot), so the minor byte already equals + the literal digits in the tag_name: 1.6 -> 6, 1.71 -> 71, 2.20 -> 20. No + scaling is applied. + + ``patch`` is None when the source predates the trailing patch byte of the + version response (for example a device dict cached by an older host), and + the two-part form is used. With a patch available the three-part form lets a + device on a patch release such as 2.25.1 match its tag instead of appearing + to have an update pending forever. + + Every consumer that displays a version must format it through here. A host + that formats the installed version one way and compares it against a tag + formatted another way shows a permanently pending update. + """ + if patch is None: + return f"{major}.{minor}" + return f"{major}.{minor}.{patch}" + + class FirmwareVersion(TypedDict): """Firmware version information. diff --git a/src/opendisplay/models/led_flash.py b/src/opendisplay/models/led_flash.py index bebdb75..c2fdf94 100644 --- a/src/opendisplay/models/led_flash.py +++ b/src/opendisplay/models/led_flash.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass, field +from typing import Final def _check_u8(name: str, value: int) -> None: @@ -15,6 +16,72 @@ def _check_nibble(name: str, value: int) -> None: raise ValueError(f"{name} out of range: {value} (must be 0-15)") +#: Milliseconds represented by one step of the firmware's delay units. +DELAY_UNIT_MS: Final = 100 + + +def pack_led_color(r: int, g: int, b: int) -> int: + """Pack 8-bit RGB into the firmware's single-byte LED colour (3R/3G/2B). + + The LED payload carries one byte per step: three bits of red, three of green, + two of blue. Each channel is rescaled proportionally and rounded, so 255 maps + to the channel maximum (7, 7, 3) and 0 maps to 0. + + Blue has one bit less than the others, which is a property of the wire format + rather than of any particular LED, so blues quantise more coarsely than reds + and greens. Callers wanting an exact colour should choose values that land on + the quantisation steps. + + Args: + r: Red, 0-255. + g: Green, 0-255. + b: Blue, 0-255. + + Raises: + ValueError: If any channel is outside 0-255. + """ + for name, value in (("r", r), ("g", g), ("b", b)): + _check_u8(name, value) + return ((round(r * 7 / 255)) << 5) | ((round(g * 7 / 255)) << 2) | (round(b * 3 / 255)) + + +def unpack_led_color(packed: int) -> tuple[int, int, int]: + """Expand a packed 3R/3G/2B colour byte back to approximate 8-bit RGB. + + The inverse of :func:`pack_led_color` up to quantisation: packing is lossy, + so this returns the representative full-range colour for each quantised + level, not the original input. + + Raises: + ValueError: If ``packed`` is outside 0-255. + """ + _check_u8("packed", packed) + r = (packed >> 5) & 0x07 + g = (packed >> 2) & 0x07 + b = packed & 0x03 + return round(r * 255 / 7), round(g * 255 / 7), round(b * 255 / 3) + + +def ms_to_loop_delay_units(ms: int) -> int: + """Convert a delay in milliseconds to the 4-bit loop-delay field. + + One unit is ``DELAY_UNIT_MS``. The field is a nibble, so the representable + range is 0 to 1500 ms and values outside it clamp rather than raise: the + delay is a presentation detail of a flash pattern, and refusing to blink at + all because a caller asked for 2 seconds would be the worse failure. + """ + return max(0, min(0x0F, round(ms / DELAY_UNIT_MS))) + + +def ms_to_inter_delay_units(ms: int) -> int: + """Convert a delay in milliseconds to the 8-bit inter-delay field. + + One unit is ``DELAY_UNIT_MS``, giving a representable range of 0 to 25500 ms. + Clamps rather than raises, for the same reason as :func:`ms_to_loop_delay_units`. + """ + return max(0, min(0xFF, round(ms / DELAY_UNIT_MS))) + + @dataclass(frozen=True, slots=True) class LedFlashStep: """One LED flash step used by firmware LED mode 1.""" @@ -30,6 +97,38 @@ def __post_init__(self) -> None: _check_nibble("loop_delay_units", self.loop_delay_units) _check_u8("inter_delay_units", self.inter_delay_units) + @classmethod + def from_rgb( + cls, + rgb: tuple[int, int, int], + *, + flash_count: int = 1, + loop_delay_ms: int = 0, + inter_delay_ms: int = 0, + ) -> LedFlashStep: + """Build a step from 8-bit RGB and millisecond delays. + + The human-facing constructor: it takes the units a caller actually has + and converts to the firmware's packed colour byte and 100 ms delay units, + so a caller never has to know the wire encoding to blink an LED. Use the + plain constructor when you already hold encoded values, such as when + round-tripping a payload read back from a device. + + Delays clamp to their representable range rather than raising; see + :func:`ms_to_loop_delay_units`. + """ + return cls( + color=pack_led_color(*rgb), + flash_count=flash_count, + loop_delay_units=ms_to_loop_delay_units(loop_delay_ms), + inter_delay_units=ms_to_inter_delay_units(inter_delay_ms), + ) + + @property + def rgb(self) -> tuple[int, int, int]: + """The step's colour as approximate 8-bit RGB (lossy; see unpack_led_color).""" + return unpack_led_color(self.color) + @dataclass(frozen=True, slots=True) class LedFlashConfig: diff --git a/src/opendisplay/protocol/__init__.py b/src/opendisplay/protocol/__init__.py index 687adbc..4b53057 100644 --- a/src/opendisplay/protocol/__init__.py +++ b/src/opendisplay/protocol/__init__.py @@ -10,6 +10,7 @@ MAX_START_PAYLOAD, NFC_CHUNK_SIZE, NFC_INLINE_MAX, + NFC_MIME_TYPE_MAX, NFC_SUB_READ, NFC_SUB_WRITE_DATA, NFC_SUB_WRITE_END, @@ -44,6 +45,7 @@ build_direct_write_start_uncompressed, build_enter_dfu_command, build_led_activate_command, + build_nfc_payload, build_nfc_write_data_command, build_nfc_write_end_command, build_nfc_write_inline_command, @@ -119,6 +121,7 @@ "NFC_SUB_WRITE_END", "NFC_INLINE_MAX", "NFC_CHUNK_SIZE", + "NFC_MIME_TYPE_MAX", "NFC_WRITE_MAX_TOTAL", "OD_LAN_TCP_PORT", "OD_LAN_TLS_PORT", @@ -126,6 +129,7 @@ "OD_LAN_MAX_PAYLOAD", "OD_LAN_MDNS_SERVICE", "OD_LAN_READ_TIMEOUT_S", + "build_nfc_payload", "build_nfc_write_inline_command", "build_nfc_write_start_command", "build_nfc_write_data_command", diff --git a/src/opendisplay/protocol/commands.py b/src/opendisplay/protocol/commands.py index 2e39704..2ddb087 100644 --- a/src/opendisplay/protocol/commands.py +++ b/src/opendisplay/protocol/commands.py @@ -7,6 +7,7 @@ from enum import IntEnum from ..models.buzzer_activate import BuzzerActivateConfig +from ..models.enums import NfcRecordType from ..models.led_flash import LedFlashConfig @@ -491,6 +492,64 @@ def build_pipe_write_end_command(refresh_mode: int, new_etag: int | None = None) return cmd + refresh_mode.to_bytes(1, byteorder="big") + new_etag.to_bytes(4, byteorder="big") +#: MIME record header: a single length byte, so the type is capped at 255 bytes. +NFC_MIME_TYPE_MAX = 255 + + +def build_nfc_payload( + record_type: NfcRecordType | int, + content: bytes | str, + mime_type: str | None = None, +) -> bytes: + """Build and validate the NDEF payload for an NFC write, without any I/O. + + This is the same payload ``OpenDisplayDevice.write_nfc_*`` sends, exposed + separately so a caller can validate before opening a connection. Rejecting an + oversized record costs nothing here; discovering it after waking a sleeping + device and negotiating a link costs a wake window. + + Sizes are measured in encoded bytes, not characters: a multi-byte UTF-8 + string is longer than it looks, and for MIME records the length-prefixed type + header counts against the same budget as the body. + + Args: + record_type: NDEF record type. MIME requires ``mime_type``; the others + reject it, since silently ignoring it would hide a caller's mistake. + content: Record content. A ``str`` is encoded as UTF-8; ``bytes`` is used + as-is. + mime_type: MIME type for a MIME record, for example "text/vcard". Must + encode to 1..255 bytes as UTF-8. + + Returns: + Payload bytes ready for ``write_nfc``. + + Raises: + ValueError: If ``mime_type`` is present on a non-MIME record or absent on + a MIME one, if the encoded MIME type is outside 1..255 bytes, or if + the total payload is empty or exceeds ``NFC_WRITE_MAX_TOTAL``. + """ + is_mime = int(record_type) == int(NfcRecordType.MIME) + if mime_type is not None and not is_mime: + raise ValueError("mime_type is only valid for a MIME record") + if is_mime and mime_type is None: + raise ValueError("a MIME record requires a mime_type") + + body = content.encode("utf-8") if isinstance(content, str) else content + + if is_mime: + assert mime_type is not None # narrowed by the checks above + mt = mime_type.encode("utf-8") + if not 1 <= len(mt) <= NFC_MIME_TYPE_MAX: + raise ValueError(f"mime_type must encode to 1..{NFC_MIME_TYPE_MAX} bytes, got {len(mt)}") + payload = bytes([len(mt)]) + mt + body + else: + payload = body + + if not 1 <= len(payload) <= NFC_WRITE_MAX_TOTAL: + raise ValueError(f"payload length must be 1..{NFC_WRITE_MAX_TOTAL}, got {len(payload)}") + return payload + + def build_nfc_write_inline_command(rec_type: int, payload: bytes) -> bytes: """Build an NFC_ENDPOINT inline write (sub-opcode 0x01). diff --git a/src/opendisplay/sleep.py b/src/opendisplay/sleep.py new file mode 100644 index 0000000..0020c64 --- /dev/null +++ b/src/opendisplay/sleep.py @@ -0,0 +1,119 @@ +"""When a deep-sleeping OpenDisplay device is reachable. + +A battery device configured for deep sleep is dark most of the time: it wakes on +a timer, advertises for a short window, and returns to sleep if nothing talks to +it. A host that wants to reach one has to know how long that window is and +whether it has already closed, and every host that reimplements those rules +drifts from the firmware independently. + +:class:`SleepModel` holds only what the *device* determines. Host policy - how +many missed wakes to tolerate before calling a device unavailable, how long to +keep queued work, how much slack to allow for scanner latency - stays with the +host, which is why :meth:`SleepModel.probably_asleep` takes ``slack`` as an +argument rather than baking a value in. + +Verified against firmware ``upstream/main``: ``DEFAULT_IDLE_HOLD_MS`` in +``src/main.h`` and the sleep-entry condition in ``platformIdle()`` +(``src/main.cpp``). Deep sleep is an ESP32 behaviour; nRF targets idle at their +configured cadence instead. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Final + +from .models.config import GlobalConfig, PowerOption + +#: Quiet window the firmware holds open when ``sleep_timeout_ms`` is 0, in +#: milliseconds (``DEFAULT_IDLE_HOLD_MS`` in the firmware's ``src/main.h``). +DEFAULT_WAKE_WINDOW_MS: Final = 10_000 + + +@dataclass(frozen=True, slots=True) +class SleepModel: + """Device-derived deep-sleep timing for one device. + + Build with :meth:`from_config` or :meth:`from_power` rather than + constructing directly, so the fields stay consistent with the device config + they came from. + """ + + #: Whether the device is configured to deep sleep at all. + is_deep_sleeping: bool + #: Timer-wake interval in seconds; 0 when deep sleep is off. + deep_sleep_time_seconds: int + #: Configured quiet window in milliseconds; 0 means "use the firmware default". + sleep_timeout_ms: int + + @classmethod + def from_power(cls, power: PowerOption) -> SleepModel: + """Build from a device's power configuration.""" + return cls( + is_deep_sleeping=power.deep_sleep_enabled, + deep_sleep_time_seconds=power.deep_sleep_time_seconds, + sleep_timeout_ms=power.sleep_timeout_ms, + ) + + @classmethod + def from_config(cls, config: GlobalConfig) -> SleepModel: + """Build from a full device config, read via ``OpenDisplayDevice.config``.""" + return cls.from_power(config.power) + + @property + def wake_window_s(self) -> float: + """Quiet window after a wake, in seconds, before the device sleeps again. + + Falls back to the firmware default when ``sleep_timeout_ms`` is 0, which + is what the firmware itself does. + """ + return (self.sleep_timeout_ms or DEFAULT_WAKE_WINDOW_MS) / 1000.0 + + def probably_asleep( + self, + last_seen: float | None, + now: float | None = None, + slack: float = 0.0, + ) -> bool: + """Return True if the device has almost certainly gone back to sleep. + + A sleeping device advertises only while its window is open, so an + advertisement older than one window means the window has closed and a + connect attempt would spend a full retry budget on a dark radio. + + This is a *prediction, not a fact*, and it errs towards "asleep" in two + known ways. The firmware measures the window from the last activity + rather than from the wake, so a client that connects and drops re-arms + the whole window; and a button wake holds the device up for at least + ``min_wake_time_seconds`` regardless. In both cases the device may still + be reachable when this returns True. Callers that can afford one cheap + connect attempt should prefer trying over trusting this. + + The test is pure freshness and does not consult ``is_deep_sleeping``: + callers combine the two where the distinction matters, since a mains + powered device is never "asleep" in this sense. + + Args: + last_seen: Wall-clock timestamp of the most recent advertisement, or + None if the device has never been seen - which counts as asleep. + now: Wall-clock override, for tests. + slack: Extra seconds to tolerate on top of the window, for host-side + latency between the device transmitting and the host recording + it (scanners, Bluetooth proxies). Defaults to none. + """ + if last_seen is None: + return True + current = time.time() if now is None else now + return (current - last_seen) > (self.wake_window_s + slack) + + def next_expected_wake(self, last_seen: float | None) -> float | None: + """Wall-clock time of the next timer wake, or None if not predictable. + + Returns None when the device does not deep sleep or has never been seen. + Assumes the device slept immediately after ``last_seen``, so one whose + window was extended by activity wakes slightly later than predicted. + """ + if last_seen is None or not self.is_deep_sleeping: + return None + return last_seen + self.deep_sleep_time_seconds diff --git a/tests/conftest.py b/tests/conftest.py index c7c342e..20e5a87 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,6 +8,9 @@ import pytest from PIL import Image +from opendisplay.models.config import GlobalConfig, ManufacturerData, PowerOption, SystemConfig +from opendisplay.models.enums import PowerMode + # Path to captured real protocol data FIXTURES_DIR = Path(__file__).parent / "fixtures/real_protocol_data" @@ -84,6 +87,63 @@ def _make(responses: list | None = None, **kwargs: object) -> FakeTransport: return _make +@pytest.fixture +def power_option(): + """Factory fixture returning a :class:`PowerOption` with test defaults. + + Defaults describe a battery device configured to deep sleep on a 5 minute + timer with the firmware's default wake window, which is the interesting case + for sleep behaviour; pass keyword overrides for anything else. Every field is + filled so callers only state what their test is actually about. + """ + + def _make(**overrides: object) -> PowerOption: + params: dict[str, object] = { + "power_mode": int(PowerMode.BATTERY), + "battery_capacity_mah": b"\x00\x00\x00", + "sleep_timeout_ms": 0, + "tx_power": 0, + "sleep_flags": 0, + "battery_sense_pin": 0xFF, + "battery_sense_enable_pin": 0xFF, + "battery_sense_flags": 0, + "capacity_estimator": 0, + "voltage_scaling_factor": 0, + "deep_sleep_current_ua": 0, + "deep_sleep_time_seconds": 300, + "charge_enable_pin": 0xFF, + "charge_state_pin": 0xFF, + "charger_flags": 0, + "min_wake_time_seconds": 0, + "screen_timeout_seconds": 0, + "reserved": b"\x00" * 10, + } + params.update(overrides) + return PowerOption(**params) # type: ignore[arg-type] + + return _make + + +@pytest.fixture +def global_config(power_option): + """Factory fixture returning a minimal :class:`GlobalConfig`. + + Carries only the three required sections; pass ``power=`` to supply a + specific :class:`PowerOption`, or any other field as a keyword override. + """ + + def _make(power: PowerOption | None = None, **overrides: object) -> GlobalConfig: + params: dict[str, object] = { + "system": SystemConfig(ic_type=2, communication_modes=0x05, device_flags=0, pwr_pin=0xFF, reserved=b""), + "manufacturer": ManufacturerData(manufacturer_id=1, board_type=1, board_revision=1, reserved=b""), + "power": power if power is not None else power_option(), + } + params.update(overrides) + return GlobalConfig(**params) # type: ignore[arg-type] + + return _make + + @pytest.fixture def small_test_image(): """Create a small RGB test image for encoding tests.""" diff --git a/tests/unit/test_crypto.py b/tests/unit/test_crypto.py index 95a2e69..11cc88e 100644 --- a/tests/unit/test_crypto.py +++ b/tests/unit/test_crypto.py @@ -3,6 +3,7 @@ import pytest from opendisplay.crypto import ( + KEY_LENGTH_BYTES, aes_cmac, aes_ecb_encrypt, compute_challenge_response, @@ -12,7 +13,9 @@ encrypt_command, generate_client_nonce, get_nonce, + parse_encryption_key, ) +from opendisplay.exceptions import AuthenticationError, InvalidEncryptionKeyError, OpenDisplayError _RFC4493_KEY = bytes.fromhex("2b7e151628aed2a6abf7158809cf4f3c") @@ -275,3 +278,71 @@ def test_random(self): n1 = generate_client_nonce() n2 = generate_client_nonce() assert n1 != n2 + + +class TestParseEncryptionKey: + """The stored-key format: 32 hex characters for a 16-byte AES-128 key.""" + + KEY_HEX = "aabbccddee112233aabbccddee112233" + + def test_none_means_no_key(self) -> None: + """An unencrypted device stores nothing, which is not an error.""" + assert parse_encryption_key(None) is None + + def test_parses_a_valid_key(self) -> None: + assert parse_encryption_key(self.KEY_HEX) == bytes.fromhex(self.KEY_HEX) + assert len(parse_encryption_key(self.KEY_HEX)) == KEY_LENGTH_BYTES + + def test_case_is_not_significant(self) -> None: + """A key pasted from a config tool round-trips whichever way it is cased.""" + assert parse_encryption_key(self.KEY_HEX.upper()) == parse_encryption_key(self.KEY_HEX) + + def test_surrounding_whitespace_is_ignored(self) -> None: + assert parse_encryption_key(f" {self.KEY_HEX}\n") == parse_encryption_key(self.KEY_HEX) + + @pytest.mark.parametrize( + "separated", + [ + "aa:bb:cc:dd:ee:11:22:33:aa:bb:cc:dd:ee:11:22:33", + "aa bb cc dd ee 11 22 33 aa bb cc dd ee 11 22 33", + "AA:BB:CC:DD:EE:11:22:33:AA:BB:CC:DD:EE:11:22:33", + ], + ) + def test_separators_are_not_content(self, separated: str) -> None: + """A key is copied between tools by humans; colons and spaces are noise.""" + assert parse_encryption_key(separated) == parse_encryption_key(self.KEY_HEX) + + @pytest.mark.parametrize( + "raw", + [ + "", # absence is expressed by None, so empty is malformed + "aabb", # too short + "aabbccddee112233aabbccddee1122", # 30 chars + "aabbccddee112233aabbccddee1122334", # 33 chars + "aabbccddee112233aabbccddee11223344", # 34 chars + ], + ) + def test_wrong_length_is_rejected(self, raw: str) -> None: + with pytest.raises(InvalidEncryptionKeyError): + parse_encryption_key(raw) + + @pytest.mark.parametrize( + "raw", + [ + "zzbbccddee112233aabbccddee112233", # not hex at all + "aabbccddee112233aabbccddee1122g3", # one bad nibble + "aabbccddee112233aabbccddee11223!", # punctuation + ], + ) + def test_non_hex_is_rejected(self, raw: str) -> None: + with pytest.raises(InvalidEncryptionKeyError): + parse_encryption_key(raw) + + def test_error_is_an_opendisplay_error_not_an_auth_error(self) -> None: + """Nothing was sent to a device, so this is local config, not a rejection. + + A host should react by asking for the key again, not by treating the + device as having refused it. + """ + assert issubclass(InvalidEncryptionKeyError, OpenDisplayError) + assert not issubclass(InvalidEncryptionKeyError, AuthenticationError) diff --git a/tests/unit/test_device_nfc_write.py b/tests/unit/test_device_nfc_write.py index 3deb1eb..b52a173 100644 --- a/tests/unit/test_device_nfc_write.py +++ b/tests/unit/test_device_nfc_write.py @@ -8,6 +8,7 @@ from opendisplay.crypto import encrypt_command from opendisplay.exceptions import BLETimeoutError, InvalidResponseError, NfcNotSupportedError, NfcWriteError from opendisplay.models.enums import NfcRecordType +from opendisplay.protocol.commands import NFC_MIME_TYPE_MAX, NFC_WRITE_MAX_TOTAL, build_nfc_payload class _FakeConnection: @@ -285,3 +286,89 @@ async def test_write_nfc_requires_connection() -> None: with pytest.raises(RuntimeError, match="not connected"): await device.write_nfc(NfcRecordType.TEXT, b"hello") + + +class TestBuildNfcPayload: + """Pure payload assembly, so a caller can validate before spending a connection.""" + + def test_text_and_uri_pass_content_through_as_utf8(self) -> None: + assert build_nfc_payload(NfcRecordType.TEXT, "hello") == b"hello" + assert build_nfc_payload(NfcRecordType.URI, "https://example.test") == b"https://example.test" + + def test_bytes_content_is_used_as_is(self) -> None: + assert build_nfc_payload(NfcRecordType.TEXT, b"\x01\x02\x03") == b"\x01\x02\x03" + + def test_mime_prefixes_a_length_byte_and_the_type(self) -> None: + payload = build_nfc_payload(NfcRecordType.MIME, "BEGIN:VCARD", "text/vcard") + assert payload == bytes([len(b"text/vcard")]) + b"text/vcard" + b"BEGIN:VCARD" + + def test_size_is_measured_in_bytes_not_characters(self) -> None: + """A multi-byte string is longer than it looks; the limit is on bytes.""" + content = "é" * 256 # 2 bytes each in UTF-8 = 512 + assert len(build_nfc_payload(NfcRecordType.TEXT, content)) == NFC_WRITE_MAX_TOTAL + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.TEXT, content + "x") + + @pytest.mark.parametrize("size", [1, 120, 121, NFC_WRITE_MAX_TOTAL - 1, NFC_WRITE_MAX_TOTAL]) + def test_accepts_sizes_up_to_the_firmware_limit(self, size: int) -> None: + assert len(build_nfc_payload(NfcRecordType.TEXT, "x" * size)) == size + + @pytest.mark.parametrize("size", [0, NFC_WRITE_MAX_TOTAL + 1, NFC_WRITE_MAX_TOTAL + 100]) + def test_rejects_empty_and_oversized_payloads(self, size: int) -> None: + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.TEXT, "x" * size) + + def test_mime_header_counts_against_the_same_budget(self) -> None: + """The type header is part of the payload, not free space alongside it.""" + mime = "text/vcard" # 10 bytes, plus 1 length byte + body_budget = NFC_WRITE_MAX_TOTAL - len(mime) - 1 + assert len(build_nfc_payload(NfcRecordType.MIME, "x" * body_budget, mime)) == NFC_WRITE_MAX_TOTAL + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.MIME, "x" * (body_budget + 1), mime) + + @pytest.mark.parametrize("length", [1, NFC_MIME_TYPE_MAX]) + def test_accepts_mime_types_at_the_header_bounds(self, length: int) -> None: + mime = "x" * length + assert build_nfc_payload(NfcRecordType.MIME, b"b", mime)[0] == length + + @pytest.mark.parametrize("length", [0, NFC_MIME_TYPE_MAX + 1]) + def test_rejects_mime_types_outside_the_header_bounds(self, length: int) -> None: + """The header length is a single byte, so 0 and 256 are unrepresentable.""" + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.MIME, b"b", "x" * length) + + def test_mime_type_on_a_non_mime_record_is_an_error(self) -> None: + """Ignoring it would silently drop something the caller asked for.""" + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.TEXT, "hello", "text/vcard") + + def test_mime_record_without_a_mime_type_is_an_error(self) -> None: + with pytest.raises(ValueError): + build_nfc_payload(NfcRecordType.MIME, "hello") + + def test_accepts_a_raw_int_record_type(self) -> None: + assert build_nfc_payload(int(NfcRecordType.TEXT), "hello") == b"hello" + + +@pytest.mark.asyncio +async def test_write_nfc_mime_emits_exactly_the_builder_payload() -> None: + """Anti-drift: the device path and the pre-flight validator must agree. + + If write_nfc_mime assembled its own payload, a host could validate a record + the device then rejects, or vice versa. Pinning the bytes keeps the two from + diverging silently. + """ + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(response=b"\x00\x83\x81") + device._connection = fake + + await device.write_nfc_mime("text/vcard", "BEGIN:VCARD") + + expected_payload = build_nfc_payload(NfcRecordType.MIME, "BEGIN:VCARD", "text/vcard") + expected_cmd = ( + b"\x00\x83" + + bytes([0x01, int(NfcRecordType.MIME)]) + + len(expected_payload).to_bytes(2, "big") + + expected_payload + ) + assert fake.written == [expected_cmd] diff --git a/tests/unit/test_device_upload_compression.py b/tests/unit/test_device_upload_compression.py index 4c46c1b..5ed01a8 100644 --- a/tests/unit/test_device_upload_compression.py +++ b/tests/unit/test_device_upload_compression.py @@ -6,7 +6,7 @@ from epaper_dithering import ColorScheme from PIL import Image -from opendisplay import OpenDisplayDevice +from opendisplay import OpenDisplayDevice, prepare_image from opendisplay.models.capabilities import DeviceCapabilities from opendisplay.models.config import ( DisplayConfig, @@ -268,3 +268,39 @@ def test_prepare_image_always_uses_9bit_zlib_window(transmission_modes: int) -> ) assert compressed is not None assert zlib_window_bits(compressed) == FIRMWARE_ZLIB_WINDOW_BITS + + +class TestPrepareImageCompressNone: + """compress=None means "ask the config", the same question upload_image asks.""" + + def _image(self) -> Image.Image: + return Image.new("RGB", (2, 2), color=(0, 0, 0)) + + @pytest.mark.parametrize( + ("transmission_modes", "label"), + [(0x02, "zip"), (0x01, "streaming decompression"), (0x03, "both bits")], + ) + def test_derives_true_on_a_compression_capable_panel(self, transmission_modes: int, label: str) -> None: + config = _config(transmission_modes=transmission_modes) + derived = prepare_image(self._image(), config=config, compress=None) + explicit = prepare_image(self._image(), config=config, compress=True) + assert derived[1] is not None, f"expected compression for {label}" + assert derived[1] == explicit[1] + + def test_derives_false_when_the_panel_advertises_neither_bit(self) -> None: + config = _config(transmission_modes=0x00) + derived = prepare_image(self._image(), config=config, compress=None) + assert derived[1] is None + assert derived[0] == prepare_image(self._image(), config=config, compress=False)[0] + + def test_explicit_values_still_win_over_the_config(self) -> None: + """None is opt-in; a caller that states a preference keeps it.""" + capable = _config(transmission_modes=0x02) + incapable = _config(transmission_modes=0x00) + assert prepare_image(self._image(), config=capable, compress=False)[1] is None + assert prepare_image(self._image(), config=incapable, compress=True)[1] is not None + + def test_default_is_unchanged_for_existing_callers(self) -> None: + """The default stays True, so nothing that omits the argument shifts.""" + config = _config(transmission_modes=0x00) + assert prepare_image(self._image(), config=config)[1] is not None diff --git a/tests/unit/test_models_config.py b/tests/unit/test_models_config.py index 40516c2..0ba5394 100644 --- a/tests/unit/test_models_config.py +++ b/tests/unit/test_models_config.py @@ -7,6 +7,7 @@ BoardManufacturer, DIYBoardType, PowerMode, + Rotation, SeeedBoardType, WaveshareBoardType, ) @@ -216,3 +217,100 @@ def test_explicit_offset_is_kept(self) -> None: sensor = SensorData(instance_number=0, sensor_type=4, bus_id=1, msd_data_start_byte=3) assert sensor.sht40_msd_start_byte == 3 + + +class TestBinaryInputsEnabledButtonIds: + """input_flags is a bitmask over 8 pin slots; the bit position is the button id.""" + + def _inputs(self, input_flags: int) -> BinaryInputs: + return BinaryInputs( + instance_number=0, + input_type=1, + display_as=1, + reserved_pins=b"\x00" * 8, + input_flags=input_flags, + invert=0, + pullups=0, + pulldowns=0, + ) + + def test_no_slots_populated(self) -> None: + assert self._inputs(0x00).enabled_button_ids == () + + def test_single_slot(self) -> None: + assert self._inputs(0x01).enabled_button_ids == (0,) + + def test_all_slots(self) -> None: + assert self._inputs(0xFF).enabled_button_ids == (0, 1, 2, 3, 4, 5, 6, 7) + + def test_sparse_mask_keeps_bit_positions(self) -> None: + """Ids are bit positions, not a count: gaps must not renumber the buttons.""" + assert self._inputs(0b1010_0100).enabled_button_ids == (2, 5, 7) + + def test_highest_bit_is_button_seven(self) -> None: + """The report byte carries a 3-bit id, so 7 is the last addressable slot.""" + assert self._inputs(0x80).enabled_button_ids == (7,) + + @pytest.mark.parametrize("flags", [0x00, 0x01, 0x0F, 0x55, 0xAA, 0xFF]) + def test_agrees_with_an_explicit_bit_scan(self, flags: int) -> None: + expected = tuple(bit for bit in range(8) if flags & (1 << bit)) + assert self._inputs(flags).enabled_button_ids == expected + + +class TestDisplayConfigCanvasSize: + """Axis order follows the combined device + caller rotation.""" + + def _display(self, rotation: int) -> DisplayConfig: + return DisplayConfig( + instance_number=0, + display_technology=1, + panel_ic_type=0, + pixel_width=296, + pixel_height=128, + active_width_mm=0, + active_height_mm=0, + tag_type=0, + rotation=rotation, + reset_pin=0xFF, + busy_pin=0xFF, + dc_pin=0xFF, + cs_pin=0xFF, + data_pin=0, + partial_update_support=1, + color_scheme=0, + transmission_modes=0, + clk_pin=0, + reserved_pins=b"\x00" * 7, + full_update_mC=0, + reserved=b"\x00" * 13, + ) + + def test_unrotated_panel_uses_its_own_dimensions(self) -> None: + assert self._display(0).canvas_size() == (296, 128) + + @pytest.mark.parametrize("degrees", [90, 270]) + def test_quarter_turns_transpose(self, degrees: int) -> None: + assert self._display(0).canvas_size(degrees) == (128, 296) + + @pytest.mark.parametrize("degrees", [0, 180, 360]) + def test_half_turns_do_not_transpose(self, degrees: int) -> None: + assert self._display(0).canvas_size(degrees) == (296, 128) + + def test_device_rotation_alone_transposes(self) -> None: + """A panel mounted sideways needs a transposed canvas with no caller rotation.""" + assert self._display(90).canvas_size() == (128, 296) + + def test_rotations_combine_rather_than_override(self) -> None: + """90 on top of 90 is 180, which is back to the panel's own axis order.""" + assert self._display(90).canvas_size(90) == (296, 128) + assert self._display(90).canvas_size(180) == (128, 296) + assert self._display(180).canvas_size(180) == (296, 128) + + def test_accepts_a_rotation_enum(self) -> None: + assert self._display(0).canvas_size(Rotation.ROTATE_90) == (128, 296) + assert self._display(0).canvas_size(Rotation.ROTATE_0) == (296, 128) + + def test_unknown_stored_rotation_is_treated_as_zero(self) -> None: + """A config carrying a rotation the library cannot decode must still render.""" + assert self._display(200).canvas_size() == (296, 128) + assert self._display(200).canvas_size(90) == (128, 296) diff --git a/tests/unit/test_models_firmware.py b/tests/unit/test_models_firmware.py new file mode 100644 index 0000000..423fc47 --- /dev/null +++ b/tests/unit/test_models_firmware.py @@ -0,0 +1,67 @@ +"""Test firmware version formatting and OTA install capability.""" + +import pytest + +from opendisplay.models.enums import ICType +from opendisplay.models.firmware import ( + firmware_ota_asset, + format_firmware_version, + supports_ble_ota_install, +) + + +@pytest.mark.parametrize( + ("major", "minor", "patch", "expected"), + [ + # The minor byte is the literal digits in the tag, never scaled. + (1, 6, None, "1.6"), + (1, 71, None, "1.71"), + (2, 20, None, "2.20"), + # A patch release must render three parts or it can never match its tag, + # which is what makes an update look permanently pending. + (2, 25, 1, "2.25.1"), + (1, 6, 0, "1.6.0"), + ], +) +def test_format_matches_the_release_tag_convention(major: int, minor: int, patch: int | None, expected: str) -> None: + assert format_firmware_version(major, minor, patch) == expected + + +def test_absent_patch_and_zero_patch_are_different() -> None: + """None means "this source predates the patch byte", 0 means "patch zero". + + Collapsing them would make a 1.6.0 device report 1.6 and stop matching the + 1.6.0 tag. + """ + assert format_firmware_version(1, 6, None) == "1.6" + assert format_firmware_version(1, 6, 0) == "1.6.0" + + +def test_patch_defaults_to_absent() -> None: + assert format_firmware_version(3, 4) == "3.4" + + +def test_ble_ota_install_is_offered_only_for_silabs() -> None: + assert supports_ble_ota_install(ICType.EFR32BG22) is True + # nRF Legacy DFU has an asset but strands the device when driven over a + # Bluetooth proxy, so it is deliberately not offered. + assert supports_ble_ota_install(ICType.NRF52811) is False + assert supports_ble_ota_install(ICType.NRF52840) is False + # ESP32 has no BLE OTA path at all. + assert supports_ble_ota_install(ICType.ESP32_C6) is False + + +def test_install_capability_is_stricter_than_asset_availability() -> None: + """Every installable IC has an asset, but not every IC with an asset is installable. + + Pinned because the two are easy to conflate, and conflating them is what + would re-enable the nRF-over-proxy path that bricks devices into their + bootloader. + """ + installable = [ic for ic in ICType if supports_ble_ota_install(ic)] + assert installable, "expected at least one installable IC type" + for ic in installable: + assert firmware_ota_asset(ic, "2.25.1") is not None + + has_asset = {ic for ic in ICType if firmware_ota_asset(ic, "2.25.1") is not None} + assert has_asset - set(installable), "expected an IC with an asset that is not installable" diff --git a/tests/unit/test_models_led_flash.py b/tests/unit/test_models_led_flash.py index 5fa344c..8acc327 100644 --- a/tests/unit/test_models_led_flash.py +++ b/tests/unit/test_models_led_flash.py @@ -2,7 +2,14 @@ import pytest -from opendisplay.models.led_flash import LedFlashConfig, LedFlashStep +from opendisplay.models.led_flash import ( + LedFlashConfig, + LedFlashStep, + ms_to_inter_delay_units, + ms_to_loop_delay_units, + pack_led_color, + unpack_led_color, +) def test_led_flash_config_to_bytes_and_from_bytes_roundtrip() -> None: @@ -89,3 +96,107 @@ def test_from_bytes_accepts_raw_0xff_without_raising() -> None: payload = bytes([0x70, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0]) cfg = LedFlashConfig.from_bytes(payload) # must not raise assert cfg.group_repeats is None + + +def test_pack_led_color_saturates_each_channel() -> None: + """Full-scale input maps to each channel's maximum: 3 bits, 3 bits, 2 bits.""" + assert pack_led_color(255, 0, 0) == 0b111_000_00 + assert pack_led_color(0, 255, 0) == 0b000_111_00 + assert pack_led_color(0, 0, 255) == 0b000_000_11 + assert pack_led_color(255, 255, 255) == 0xFF + assert pack_led_color(0, 0, 0) == 0x00 + + +def test_pack_led_color_rejects_out_of_range_channels() -> None: + for bad in ((256, 0, 0), (0, -1, 0), (0, 0, 999)): + with pytest.raises(ValueError): + pack_led_color(*bad) + + +def test_packed_colors_survive_a_round_trip_through_the_wire_format() -> None: + """Every representable colour byte must unpack and repack to itself. + + Exhaustive over all 256 values, which is the whole space: this pins the + channel bit widths and their order, so a change to either fails here rather + than silently lighting the wrong colour. + """ + for packed in range(256): + assert pack_led_color(*unpack_led_color(packed)) == packed + + +def test_packed_color_survives_a_step_round_trip() -> None: + """A packed colour must reach the device unchanged through the real payload.""" + packed = pack_led_color(255, 128, 64) + cfg = LedFlashConfig(step1=LedFlashStep(color=packed, flash_count=1)) + assert LedFlashConfig.from_bytes(cfg.to_bytes()).step1.color == packed + + +@pytest.mark.parametrize( + ("ms", "expected"), + [(0, 0), (100, 1), (150, 2), (1500, 15), (1600, 15), (99_999, 15)], +) +def test_ms_to_loop_delay_units_convert_and_clamp(ms: int, expected: int) -> None: + """The field is a nibble, so out-of-range delays clamp rather than raise.""" + assert ms_to_loop_delay_units(ms) == expected + + +@pytest.mark.parametrize( + ("ms", "expected"), + [(0, 0), (100, 1), (25_500, 255), (26_000, 255), (99_999_999, 255)], +) +def test_ms_to_inter_delay_units_convert_and_clamp(ms: int, expected: int) -> None: + assert ms_to_inter_delay_units(ms) == expected + + +def test_delay_units_are_accepted_by_the_step_model() -> None: + """The converters must land inside the ranges LedFlashStep validates.""" + step = LedFlashStep( + color=pack_led_color(0, 0, 255), + loop_delay_units=ms_to_loop_delay_units(99_999), + inter_delay_units=ms_to_inter_delay_units(99_999_999), + ) + assert step.loop_delay_units == 15 + assert step.inter_delay_units == 255 + + +class TestLedFlashStepFromRgb: + """The human-facing constructor: RGB and milliseconds, not wire encodings.""" + + def test_converts_colour_and_delays(self) -> None: + step = LedFlashStep.from_rgb((255, 0, 0), flash_count=2, loop_delay_ms=300, inter_delay_ms=1000) + assert step.color == pack_led_color(255, 0, 0) + assert step.flash_count == 2 + assert step.loop_delay_units == 3 + assert step.inter_delay_units == 10 + + def test_matches_the_equivalent_manual_construction(self) -> None: + """from_rgb must be sugar, not a second encoding path.""" + assert LedFlashStep.from_rgb( + (10, 200, 90), flash_count=3, loop_delay_ms=500, inter_delay_ms=200 + ) == LedFlashStep( + color=pack_led_color(10, 200, 90), + flash_count=3, + loop_delay_units=ms_to_loop_delay_units(500), + inter_delay_units=ms_to_inter_delay_units(200), + ) + + def test_out_of_range_delays_clamp_rather_than_raise(self) -> None: + """A too-long delay should still blink, not refuse to build a step.""" + step = LedFlashStep.from_rgb((0, 0, 255), loop_delay_ms=99_999, inter_delay_ms=99_999_999) + assert step.loop_delay_units == 15 + assert step.inter_delay_units == 255 + + def test_rejects_an_out_of_range_channel(self) -> None: + with pytest.raises(ValueError): + LedFlashStep.from_rgb((256, 0, 0)) + + def test_rgb_property_round_trips_through_the_wire_format(self) -> None: + step = LedFlashStep.from_rgb((255, 255, 255)) + assert step.rgb == (255, 255, 255) + assert LedFlashStep.from_rgb(step.rgb).color == step.color + + def test_survives_a_full_payload_round_trip(self) -> None: + cfg = LedFlashConfig(step1=LedFlashStep.from_rgb((255, 128, 0), flash_count=2, loop_delay_ms=200)) + decoded = LedFlashConfig.from_bytes(cfg.to_bytes()).step1 + assert decoded == cfg.step1 + assert decoded.rgb == cfg.step1.rgb diff --git a/tests/unit/test_sleep_model.py b/tests/unit/test_sleep_model.py new file mode 100644 index 0000000..050f0b9 --- /dev/null +++ b/tests/unit/test_sleep_model.py @@ -0,0 +1,89 @@ +"""Test the deep-sleep timing model.""" + +import pytest + +from opendisplay.models.enums import PowerMode +from opendisplay.sleep import DEFAULT_WAKE_WINDOW_MS, SleepModel + + +def test_deep_sleep_needs_battery_power_and_an_interval(power_option) -> None: + """Mirrors the firmware's platformIdle() gate: battery mode and a non-zero timer.""" + assert SleepModel.from_power(power_option()).is_deep_sleeping is True + assert SleepModel.from_power(power_option(power_mode=int(PowerMode.USB))).is_deep_sleeping is False + assert SleepModel.from_power(power_option(deep_sleep_time_seconds=0)).is_deep_sleeping is False + + +def test_wake_window_falls_back_to_the_firmware_default(power_option) -> None: + """sleep_timeout_ms == 0 means "use DEFAULT_IDLE_HOLD_MS", as the firmware does.""" + model = SleepModel.from_power(power_option(sleep_timeout_ms=0)) + assert model.wake_window_s == DEFAULT_WAKE_WINDOW_MS / 1000.0 + assert model.wake_window_s == 10.0 + + +def test_wake_window_honours_a_configured_value(power_option) -> None: + assert SleepModel.from_power(power_option(sleep_timeout_ms=40_000)).wake_window_s == 40.0 + + +def test_never_seen_counts_as_asleep(power_option) -> None: + model = SleepModel.from_power(power_option()) + assert model.probably_asleep(None) is True + assert model.probably_asleep(None, now=1_000.0) is True + + +def test_freshness_is_measured_against_the_wake_window(power_option) -> None: + model = SleepModel.from_power(power_option(sleep_timeout_ms=10_000)) + now = 1_000.0 + assert model.probably_asleep(now - 5.0, now=now) is False + assert model.probably_asleep(now - 9.9, now=now) is False + # Exactly one window still counts as awake; strictly older does not. + assert model.probably_asleep(now - 10.0, now=now) is False + assert model.probably_asleep(now - 10.1, now=now) is True + + +def test_slack_extends_the_freshness_horizon(power_option) -> None: + """Host-side latency is the caller's to declare, not the model's to assume.""" + model = SleepModel.from_power(power_option(sleep_timeout_ms=10_000)) + now = 1_000.0 + assert model.probably_asleep(now - 12.0, now=now) is True + assert model.probably_asleep(now - 12.0, now=now, slack=5.0) is False + assert model.probably_asleep(now - 15.1, now=now, slack=5.0) is True + + +def test_freshness_does_not_consult_is_deep_sleeping(power_option) -> None: + """probably_asleep is a pure freshness test; callers combine it themselves. + + A mains-powered device is never asleep in this sense, but the model does not + silently make that decision on the caller's behalf. + """ + mains = SleepModel.from_power(power_option(power_mode=int(PowerMode.USB))) + assert mains.is_deep_sleeping is False + assert mains.probably_asleep(1_000.0 - 60.0, now=1_000.0) is True + + +def test_next_expected_wake_is_one_interval_after_the_last_sighting(power_option) -> None: + model = SleepModel.from_power(power_option(deep_sleep_time_seconds=300)) + assert model.next_expected_wake(1_000.0) == 1_300.0 + + +@pytest.mark.parametrize( + ("power_mode", "interval"), + [ + (int(PowerMode.USB), 300), # not a deep sleeper + (int(PowerMode.BATTERY), 0), # no interval to predict from + ], +) +def test_next_expected_wake_is_unpredictable_without_deep_sleep(power_option, power_mode: int, interval: int) -> None: + model = SleepModel.from_power(power_option(power_mode=power_mode, deep_sleep_time_seconds=interval)) + assert model.next_expected_wake(1_000.0) is None + + +def test_next_expected_wake_needs_a_sighting(power_option) -> None: + assert SleepModel.from_power(power_option()).next_expected_wake(None) is None + + +def test_from_config_reads_the_power_section(power_option, global_config) -> None: + """from_config is sugar for from_power on config.power, not a second code path.""" + power = power_option(sleep_timeout_ms=5_000, deep_sleep_time_seconds=120) + config = global_config(power=power) + assert SleepModel.from_config(config) == SleepModel.from_power(power) + assert SleepModel.from_config(config).wake_window_s == 5.0 From fcea457ee74980111e3c8f56ad732d2dbff54bba Mon Sep 17 00:00:00 2001 From: gabriel Date: Wed, 26 Aug 2026 14:51:09 +0200 Subject: [PATCH 2/3] fix(cli): render firmware versions the way release tags are written `opendisplay info` always printed three version parts, so a device on the 1.6 release showed as 1.6.0 and never matched the tag it came from. It now formats through the library, so the CLI, a host's device registry and the GitHub tag all agree on what a device is running. The --json output is unchanged: it still reports separate major, minor and patch fields, which is a machine-readable contract rather than a rendering. --key now goes through the library's parser as well. It accepts everything it accepted before, including colon- and space-separated hex, and reports a malformed key with the same wording every other host will use. --- src/opendisplay/cli.py | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/src/opendisplay/cli.py b/src/opendisplay/cli.py index 73cc742..1b8a74f 100644 --- a/src/opendisplay/cli.py +++ b/src/opendisplay/cli.py @@ -23,6 +23,7 @@ from rich.tree import Tree from .battery import voltage_to_percent +from .crypto import parse_encryption_key from .device import OpenDisplayDevice from .discovery import discover_devices_with_adv from .exceptions import ( @@ -30,6 +31,7 @@ AuthenticationRequiredError, BLEConnectionError, BLETimeoutError, + InvalidEncryptionKeyError, OpenDisplayError, ) from .models.config import GlobalConfig, SensorData @@ -52,7 +54,7 @@ TouchIcType, WifiEncryption, ) -from .models.firmware import FirmwareVersion +from .models.firmware import FirmwareVersion, format_firmware_version from .partial import PartialState from .sensors import SensorReading @@ -99,16 +101,15 @@ def _handle_ble_error(exc: OpenDisplayError) -> NoReturn: def _parse_hex_key(hex_str: str | None) -> bytes | None: - """Convert hex string to 16-byte AES key, or None if not provided.""" - if hex_str is None: - return None - cleaned = hex_str.strip().replace(" ", "").replace(":", "") - if len(cleaned) != 32: - _error(f"--key must be exactly 32 hex characters (16 bytes), got {len(cleaned)}") + """Convert hex string to 16-byte AES key, or None if not provided. + + Delegates the format itself to the library so the CLI cannot drift from what + every other host accepts; only the argparse-flavoured error text is local. + """ try: - return bytes.fromhex(cleaned) - except ValueError as exc: - _error(f"--key contains invalid hex characters: {exc}") + return parse_encryption_key(hex_str) + except InvalidEncryptionKeyError as exc: + _error(f"--key is not a valid encryption key: {exc}") def _parse_compression_value(flag: str, value: str) -> float | str: @@ -943,7 +944,10 @@ def _build_info_tree(ctx: _InfoContext) -> Tree: group.add(list_section.line(item)) fw = ctx.fw - version = f"{fw['major']}.{fw['minor']}.{fw.get('patch', 0)}" + # Formatted through the library so the CLI, a host's device registry and a + # release tag all render the same version. Before this, the CLI always + # printed three parts and showed 1.6.0 where the tag is 1.6. + version = format_firmware_version(fw["major"], fw["minor"], fw.get("patch")) tree.add(f"[bold]Firmware[/bold] {version} [dim](sha: {fw['sha']})[/dim]") return tree From 7d3c1ba14cf8a71054f615f72e6841708ce47b8c Mon Sep 17 00:00:00 2001 From: gabriel Date: Wed, 26 Aug 2026 18:08:07 +0200 Subject: [PATCH 3/3] feat: accept an encryption key as hex, and reject a bad one up front OpenDisplayDevice took encryption_key as bytes. A hex string, which is how every host actually stores a key, was accepted silently and then failed inside authenticate() with "TypeError: key must be bytes-like" - after the connection had already been made. On a deep-sleeping device that cost a whole wake window to learn the key was the wrong type, and the error named neither the parameter nor the fix. encryption_key (and authenticate()) now take the 16 raw bytes or 32 hex characters, in any capitalization and with optional colon or space separators. The value is normalized and validated in the constructor, so a malformed key raises InvalidEncryptionKeyError before a single frame is sent and can never be mistaken for a device refusing a good key. parse_encryption_key() widens the same way, so a host holding either form can pass it straight through instead of branching. Raw bytes of the wrong length are now rejected too, rather than reaching AES and failing there. --- src/opendisplay/crypto.py | 12 +++++++--- src/opendisplay/device.py | 27 +++++++++++++++++----- tests/unit/test_crypto.py | 47 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 8 deletions(-) diff --git a/src/opendisplay/crypto.py b/src/opendisplay/crypto.py index 21508ad..c06dda2 100644 --- a/src/opendisplay/crypto.py +++ b/src/opendisplay/crypto.py @@ -27,11 +27,13 @@ KEY_LENGTH_HEX: Final = KEY_LENGTH_BYTES * 2 -def parse_encryption_key(raw: str | None) -> bytes | None: - """Parse a stored hex encryption key into the bytes the device API expects. +def parse_encryption_key(raw: bytes | str | None) -> bytes | None: + """Normalize an encryption key into the bytes the device API expects. ``raw`` is the form a host persists or a user pastes: 32 hex characters for - the 16-byte AES-128 master key, or None when the device is unencrypted. + the 16-byte AES-128 master key, the 16 raw bytes themselves, or None when the + device is unencrypted. Bytes are validated and returned unchanged, so a + caller holding either form can pass it straight through without branching. Normalization is deliberately forgiving, because a key is something a human copies between a config tool, a shell and a settings field. Case is not @@ -52,6 +54,10 @@ def parse_encryption_key(raw: str | None) -> bytes | None: """ if raw is None: return None + if isinstance(raw, bytes): + if len(raw) != KEY_LENGTH_BYTES: + raise InvalidEncryptionKeyError(f"encryption key must be {KEY_LENGTH_BYTES} bytes, got {len(raw)}") + return raw candidate = raw.strip().replace(" ", "").replace(":", "") if len(candidate) != KEY_LENGTH_HEX: raise InvalidEncryptionKeyError( diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index 0684f7e..7009004 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -25,6 +25,7 @@ derive_session_key, encrypt_command, generate_client_nonce, + parse_encryption_key, ) from .display_palettes import PANELS_4GRAY, get_bwry_codes, get_gray4_codes, get_palette_for_display from .encoding import ( @@ -45,6 +46,7 @@ BLETimeoutError, ImageEncodingError, IntegrityCheckError, + InvalidEncryptionKeyError, InvalidResponseError, NfcNotSupportedError, ProtocolError, @@ -515,7 +517,7 @@ def __init__( max_attempts: int = 4, use_services_cache: bool = True, use_measured_palettes: bool = True, - encryption_key: bytes | None = None, + encryption_key: bytes | str | None = None, blocks_per_ack: int = 8, max_queue_size: int = 16, ): @@ -543,7 +545,11 @@ def __init__( max_attempts: Maximum connection attempts for bleak-retry-connector (default: 4) use_services_cache: Enable GATT service caching for faster reconnections (default: True) use_measured_palettes: Use measured color palettes when available (default: True) - encryption_key: 16-byte AES-128 master key for encrypted devices (optional). + encryption_key: AES-128 master key for encrypted devices (optional), + as the 16 raw bytes or as 32 hex characters. A malformed key is + rejected here rather than at the first command, so a bad key + cannot cost a connection (or a sleeping device's wake window) + before it is reported. blocks_per_ack: Requested PIPE_WRITE ACK cadence N (blocks per ack), 1..32 (default: 8). Negotiated down to the device maximum. max_queue_size: Requested PIPE_WRITE window W (tokens in flight), 1..32 @@ -591,7 +597,10 @@ def __init__( self._fw_version: FirmwareVersion | None = None # Encryption session state (populated by authenticate()) - self._encryption_key = encryption_key + # Normalize (and validate) up front: a str key used to be accepted + # silently and fail with an opaque TypeError inside authenticate(), + # after the connect had already happened. + self._encryption_key = parse_encryption_key(encryption_key) self._session_key: bytes | None = None self._session_id: bytes | None = None self._nonce_counter: int = 0 @@ -861,19 +870,27 @@ async def _read(self, timeout: float) -> bytes: return raw @_serialized - async def authenticate(self, key: bytes) -> None: + async def authenticate(self, key: bytes | str) -> None: """Perform two-step challenge-response authentication with the device. After successful authentication, all subsequent commands and responses are transparently encrypted/decrypted via _write() and _read(). Args: - key: 16-byte AES-128 master key + key: AES-128 master key, as the 16 raw bytes or 32 hex characters Raises: AuthenticationFailedError: If the device rejects the key or is rate-limited + InvalidEncryptionKeyError: If the key is malformed. Raised before any + frame is sent, so a bad key is never mistaken for a device that + refused a good one. InvalidResponseError: If device sends malformed response """ + parsed_key = parse_encryption_key(key) + if parsed_key is None: # pragma: no cover - excluded by the signature + raise InvalidEncryptionKeyError("authenticate() requires an encryption key") + key = parsed_key + _LOGGER.debug("Authenticating with device %s", self.mac_address) # Step 1: Request server nonce (retry once if device reports existing session) diff --git a/tests/unit/test_crypto.py b/tests/unit/test_crypto.py index 11cc88e..c9b1214 100644 --- a/tests/unit/test_crypto.py +++ b/tests/unit/test_crypto.py @@ -2,6 +2,7 @@ import pytest +from opendisplay import OpenDisplayDevice from opendisplay.crypto import ( KEY_LENGTH_BYTES, aes_cmac, @@ -338,6 +339,20 @@ def test_non_hex_is_rejected(self, raw: str) -> None: with pytest.raises(InvalidEncryptionKeyError): parse_encryption_key(raw) + def test_bytes_pass_through_validated(self) -> None: + """A caller already holding the key should not have to branch on its form.""" + raw = bytes.fromhex(self.KEY_HEX) + assert parse_encryption_key(raw) is raw + + @pytest.mark.parametrize("size", [0, 1, 15, 17, 32]) + def test_bytes_of_the_wrong_length_are_rejected(self, size: int) -> None: + """AES-128 needs exactly 16 bytes; 32 is the hex length, not a key.""" + with pytest.raises(InvalidEncryptionKeyError): + parse_encryption_key(b"\x00" * size) + + def test_both_forms_of_the_same_key_agree(self) -> None: + assert parse_encryption_key(self.KEY_HEX) == parse_encryption_key(bytes.fromhex(self.KEY_HEX)) + def test_error_is_an_opendisplay_error_not_an_auth_error(self) -> None: """Nothing was sent to a device, so this is local config, not a rejection. @@ -346,3 +361,35 @@ def test_error_is_an_opendisplay_error_not_an_auth_error(self) -> None: """ assert issubclass(InvalidEncryptionKeyError, OpenDisplayError) assert not issubclass(InvalidEncryptionKeyError, AuthenticationError) + + +class TestDeviceAcceptsEitherKeyForm: + """The constructor normalizes, so hosts never parse a key themselves.""" + + KEY_HEX = "aabbccddee112233aabbccddee112233" + + def test_a_hex_string_is_accepted_and_normalized(self) -> None: + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", encryption_key=self.KEY_HEX) + assert device._encryption_key == bytes.fromhex(self.KEY_HEX) + + def test_separated_hex_is_accepted(self) -> None: + separated = "aa:bb:cc:dd:ee:11:22:33:aa:bb:cc:dd:ee:11:22:33" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", encryption_key=separated) + assert device._encryption_key == bytes.fromhex(self.KEY_HEX) + + def test_raw_bytes_still_work(self) -> None: + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", encryption_key=bytes.fromhex(self.KEY_HEX)) + assert device._encryption_key == bytes.fromhex(self.KEY_HEX) + + def test_no_key_stays_none(self) -> None: + assert OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF")._encryption_key is None + + @pytest.mark.parametrize("bad", ["nothex", "aabb", b"short", b""]) + def test_a_malformed_key_is_rejected_before_any_connection(self, bad: bytes | str) -> None: + """The point of the change: this used to surface as a TypeError mid-connect. + + A device that is asleep most of the time would have spent a wake window + discovering the caller had a typo. + """ + with pytest.raises(InvalidEncryptionKeyError): + OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", encryption_key=bad)