diff --git a/pyproject.toml b/pyproject.toml index de547fb..e9ae8b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -114,9 +114,6 @@ load-plugins = [ [tool.pylint."MESSAGES CONTROL"] disable = [ "format", - "missing-module-docstring", # no docs yet - "missing-class-docstring", - "missing-function-docstring", "not-context-manager", "too-few-public-methods", "too-many-ancestors", @@ -142,9 +139,11 @@ expected-line-ending-format = "LF" [tool.ruff] target-version = "py310" -exclude = ["docs"] +exclude = ["docs", "setup.py"] + +# Let's tone this down once we release the first version select = ["ALL"] -ignore = ["D","ANN101", "I001", "COM"] # no docs yet +ignore = ["ANN101", "I001", "COM"] unfixable = ["ERA"] # Same as Black. diff --git a/src/bonaparte/__init__.py b/src/bonaparte/__init__.py index 0c4e320..a401464 100644 --- a/src/bonaparte/__init__.py +++ b/src/bonaparte/__init__.py @@ -1,3 +1,4 @@ +"""The Bonaparte Library.""" from __future__ import annotations __version__ = "0.1.0" diff --git a/src/bonaparte/const.py b/src/bonaparte/const.py index 3ff7887..fc448f9 100644 --- a/src/bonaparte/const.py +++ b/src/bonaparte/const.py @@ -1,3 +1,5 @@ +"""Constants for Napoleon eFIRE devices.""" + from __future__ import annotations from enum import IntEnum @@ -30,6 +32,8 @@ class Feature(StrEnum): + """Enum encapsulating possible device features.""" + AUX = "aux" BLOWER = "blower" LED_LIGHTS = "led_lights" @@ -39,7 +43,7 @@ class Feature(StrEnum): class EfireCommand(IntEnum): - """Command names to hex value Enum Class.""" + """Enum encapsulating device commands and their raw hex values.""" SET_IFC_CMD1 = 0x27 SET_IFC_CMD2 = 0x28 @@ -71,26 +75,36 @@ class EfireCommand(IntEnum): class ReturnCode(IntEnum): + """Enum encapsulating command return codes.""" + SUCCESS = 0x00 FAILURE = 0x01 class PowerState(IntEnum): + """Enum encapsulating device power states.""" + OFF = 0x00 ON = 0xFF class AuxControlState(IntEnum): + """Enum encapsulating return values for remote control use.""" + USED = 0x00 NOT_USED = 0xFF class PasswordAction(IntEnum): + """Enum encapsulating the values for password actions.""" + RESET = 0x3F SET = 0xF5 class PasswordCommandResult(IntEnum): + """Enum encapsulating possible return values for password actions.""" + SET_SUCCESS = 0x00 SET_FAILED = 0x01 INVALID_PASSWORD = 0x19 @@ -98,11 +112,15 @@ class PasswordCommandResult(IntEnum): class PasswordSetResult(IntEnum): + """Enum encapsulating return values for authentication requests.""" + FAILED = 0x25 SUCCESS = 0x53 class LedState(MultiValueEnum): + """Enum encapsulating LED controller state.""" + _init_ = "short long" OFF = 0x00, "0x000000" @@ -110,6 +128,8 @@ class LedState(MultiValueEnum): class LedMode(MultiValueEnum): + """Enum encapsulating LED controller modes.""" + _init_ = "short long setvalue" CYCLE = 0x01, 0x010101, 0x20 diff --git a/src/bonaparte/device.py b/src/bonaparte/device.py index d6d432b..35af23a 100644 --- a/src/bonaparte/device.py +++ b/src/bonaparte/device.py @@ -1,3 +1,5 @@ +"""A generic device class for Napoleon eFIRE devices.""" + from __future__ import annotations import asyncio @@ -65,6 +67,8 @@ async def _raise_if_not_connected( class EfireDevice: + """A generic eFIRE Device.""" + _ble_device: BLEDevice _client: BleakClientWithServiceCache | None = None _connect_lock: asyncio.Lock @@ -80,6 +84,7 @@ class EfireDevice: def __init__( self, ble_device: BLEDevice, advertisement_data: AdvertisementData | None = None ) -> None: + """Initialize the eFIRE Device.""" self._address = ble_device.address self._advertisement_data = advertisement_data self._ble_device = ble_device @@ -94,15 +99,17 @@ def __init__( @property def name(self) -> str: + """Device name or, if unavailable, the device address.""" return self._ble_device.name or self._ble_device.address @property def address(self) -> str: + """The device's Bluetooth MAC address.""" return self._address @property def rssi(self) -> int | None: - """Get the rssi of the device.""" + """Get the RSSI of the device.""" if self._advertisement_data: return self._advertisement_data.rssi return None @@ -110,13 +117,12 @@ def rssi(self) -> int | None: def set_ble_device_and_advertisement_data( self, ble_device: BLEDevice, advertisement_data: AdvertisementData ) -> None: - """Set the ble device and advertisement data.""" + """Set the BLE Device and advertisement data.""" self._ble_device = ble_device self._advertisement_data = advertisement_data async def _ensure_connected(self) -> None: - """Connects to the device.""" - + """Connect to the device and ensure we stay connected.""" if self._connect_lock.locked(): _LOGGER.debug( ( @@ -169,7 +175,7 @@ def _reset_disconnect_timer(self) -> None: if not self._loop: self._loop = asyncio.get_running_loop() self._disconnect_timer = self._loop.call_later( - DISCONNECT_DELAY, self._disconnect + DISCONNECT_DELAY, self._timed_disconnect ) def _disconnected(self, _client: BleakClientWithServiceCache) -> None: @@ -196,12 +202,7 @@ def unregister() -> None: self._disconnect_callbacks.append(callback) return unregister - async def disconnect(self) -> None: - """Disconnect the Fireplace.""" - _LOGGER.debug("%s: Stop", self.name) - await self._execute_disconnect() - - def _disconnect(self) -> Task[None]: + def _timed_disconnect(self) -> Task[None]: """Disconnect from device.""" self._disconnect_timer = None return asyncio.create_task(self._execute_timed_disconnect()) @@ -213,10 +214,10 @@ async def _execute_timed_disconnect(self) -> None: self.name, DISCONNECT_DELAY, ) - await self._execute_disconnect() + await self.disconnect() - async def _execute_disconnect(self) -> None: - """Execute disconnection.""" + async def disconnect(self) -> None: + """Disconnect from device.""" async with self._connect_lock: read_char = self._read_char client = self._client @@ -290,7 +291,6 @@ async def _notification_handler( @retry_bluetooth_connection_error(DEFAULT_ATTEMPTS) async def _execute_locked(self, message: bytes | bytearray) -> bytes: """Send command to device and read response.""" - if not self._write_char: msg = "Write characteristic missing" raise CharacteristicMissingError(msg) @@ -315,14 +315,14 @@ async def _execute_locked(self, message: bytes | bytearray) -> bytes: BLEAK_BACKOFF_TIME, ex, ) - await self._execute_disconnect() + await self.disconnect() raise except BleakError as ex: # Disconnect so we can reset state and try again _LOGGER.debug( "%s: RSSI: %s; Disconnecting due to error: %s", self.name, self.rssi, ex ) - await self._execute_disconnect() + await self.disconnect() raise return result @@ -331,7 +331,6 @@ async def _execute( message: bytes | bytearray, ) -> bytes: """Send command to device and read response.""" - await self._ensure_connected() _LOGGER.debug( @@ -375,6 +374,7 @@ async def _execute( async def execute_command( self, command: int, parameter: int | bytes | bytearray | None = None ) -> bytes: + """Execute a command on the device.""" payload = bytearray([command]) if parameter: diff --git a/src/bonaparte/exceptions.py b/src/bonaparte/exceptions.py index 570fd4e..41dc864 100644 --- a/src/bonaparte/exceptions.py +++ b/src/bonaparte/exceptions.py @@ -1,33 +1,30 @@ +"""Exceptions used in the Bonaparte library.""" from __future__ import annotations class EfireException(Exception): # noqa: N818 - pass + """Generic eFIRE device exception.""" class AuthError(EfireException): - pass + """For when authentication fails.""" class CommandFailedException(EfireException): - pass + """For when a command returns a failure.""" class DisconnectedException(EfireException): - pass + """For when the device is disconnected and we try to use it.""" class FeatureNotSupported(EfireException): - pass - - -class ResponseError(EfireException): - pass + """For when a feature is accessed that the device is not set up for.""" class EfireMessageValueError(ValueError): - pass + """For when an invalid message is received or generated.""" class CharacteristicMissingError(EfireException): - pass + """For when a required BLE GATT characteristic is missing.""" diff --git a/src/bonaparte/fireplace.py b/src/bonaparte/fireplace.py index ccab081..40310c4 100644 --- a/src/bonaparte/fireplace.py +++ b/src/bonaparte/fireplace.py @@ -1,3 +1,4 @@ +"""Representation of a Fireplace.""" from __future__ import annotations from dataclasses import dataclass @@ -22,11 +23,11 @@ from .exceptions import AuthError, CommandFailedException, FeatureNotSupported from .parser import ( parse_ble_version, + parse_ifc_cmd1_state, + parse_ifc_cmd2_state, parse_led_color, parse_led_controller_state, parse_mcu_version, - parse_off_state, - parse_on_state, parse_timer, ) @@ -71,6 +72,8 @@ async def _authenticated_operation( @dataclass class FireplaceFeatures: + """The set of features selected for the fireplace.""" + aux: bool = False blower: bool = False led_lights: bool = False @@ -81,6 +84,8 @@ class FireplaceFeatures: @dataclass class FireplaceState: + """State of each component in the fireplace.""" + aux: bool = False blower_speed: int = 0 flame_height: int = 0 @@ -100,6 +105,8 @@ class FireplaceState: class Fireplace(EfireDevice): + """A class representing the fireplace with state and actions.""" + _features: FireplaceFeatures _is_authenticated: bool _state: FireplaceState @@ -110,6 +117,7 @@ def __init__( ble_device: BLEDevice, features: FireplaceFeatures | None = None, ) -> None: + """Initialize a fireplace.""" super().__init__(ble_device) self._features = features if features else FireplaceFeatures() @@ -124,29 +132,36 @@ def disconnected_callback(self: Fireplace) -> None: @property def has_aux(self) -> bool: + """Whether the AUX relay control is enabled.""" return self._features.aux @property def has_blower(self) -> bool: + """Whether the blower speed control is enabled.""" return self._features.blower @property def has_led_lights(self) -> bool: + """Whether the LED controller is enabled.""" return self._features.led_lights @property def has_night_light(self) -> bool: + """Whether the Night Light control is enabled.""" return self._features.night_light @property def has_split_flow(self) -> bool: + """Whether the split flow valve control is enabled.""" return self._features.split_flow @property def state(self) -> FireplaceState: + """The state of this fireplace.""" return self._state def set_features(self, features: set[str]) -> FireplaceFeatures: + """Set all features from a list of feature strings.""" feature_set = {field.name for field in dc_fields(self._features)} if not feature_set >= features: invalid_feature = features - feature_set @@ -162,11 +177,13 @@ def set_features(self, features: set[str]) -> FireplaceFeatures: async def _simple_command( self, command: int, parameter: int | bytes | bytearray | None = None ) -> bool: + """Execute a command that returns only success of failure.""" result = await self.execute_command(command, parameter) return result[0] == ReturnCode.SUCCESS async def _ifc_cmd1(self) -> bool: + """Call the IFC CMD1 function with the current fireplace state.""" payload = bytearray( [ 0x0, @@ -181,6 +198,7 @@ async def _ifc_cmd1(self) -> bool: return result async def _ifc_cmd2(self) -> bool: + """Call the IFC CMD2 function with the current fireplace state.""" data = ( (self._state.split_flow << 7) | (self._state.blower_speed << 4) @@ -194,6 +212,7 @@ async def _ifc_cmd2(self) -> bool: return result async def authenticate(self, password: str) -> bool: + """Authenticate with the fireplace.""" response = await self.execute_command( EfireCommand.SEND_PASSWORD, password.encode("ascii") ) @@ -209,6 +228,7 @@ async def authenticate(self, password: str) -> bool: @needs_auth async def power(self, *, on: bool) -> bool: + """Set the power state on the fireplace.""" result = await self._simple_command( EfireCommand.SET_POWER, PowerState.ON if on else PowerState.OFF ) @@ -221,14 +241,17 @@ async def power(self, *, on: bool) -> bool: @needs_auth async def power_on(self) -> bool: + """Power on the fireplace.""" return await self.power(on=True) @needs_auth async def power_off(self) -> bool: + """Power off the fireplace.""" return await self.power(on=False) @needs_auth async def set_night_light_brightness(self, brightness: int) -> bool: + """Set the Night Light brightness.""" if not 0 <= brightness <= MAX_NIGHT_LIGHT_BRIGHTNESS: msg = "Night Light brightness must be between 0 and 6." raise ValueError(msg) @@ -237,11 +260,13 @@ async def set_night_light_brightness(self, brightness: int) -> bool: @needs_auth async def set_continuous_pilot(self, *, enabled: bool = True) -> bool: + """Enable or disable the continuous pilot.""" self._state.pilot = enabled return await self._ifc_cmd1() @needs_auth async def set_aux(self, *, enabled: bool) -> bool: + """Enable or disable the AUX relay.""" if not self._features.aux: msg = f"Fireplace {self.name} does not support AUX relais control" raise FeatureNotSupported(msg) @@ -250,6 +275,7 @@ async def set_aux(self, *, enabled: bool) -> bool: @needs_auth async def set_flame_height(self, flame_height: int) -> bool: + """Set the flame height.""" if not 0 <= flame_height <= MAX_FLAME_HEIGHT: msg = "Flame height must be between 0 and 6." raise ValueError(msg) @@ -258,6 +284,7 @@ async def set_flame_height(self, flame_height: int) -> bool: @needs_auth async def set_blower_speed(self, blower_speed: int) -> bool: + """Set the blower speed.""" if not self._features.aux: msg = f"Fireplace {self.name} does not have a blower" raise FeatureNotSupported(msg) @@ -270,6 +297,7 @@ async def set_blower_speed(self, blower_speed: int) -> bool: @needs_auth async def set_split_flow(self, *, enabled: bool) -> bool: + """Set the split flow valve state.""" if not self._features.split_flow: msg = f"Fireplace {self.name} does not have split flow valve" raise FeatureNotSupported(msg) @@ -279,6 +307,7 @@ async def set_split_flow(self, *, enabled: bool) -> bool: @needs_auth async def set_led_mode(self, light_mode: LedMode, *, on: bool = False) -> bool: + """Set the LED mode/effect.""" if not self._features.led_lights: msg = f"Fireplace {self.name}does not have LED controller" raise FeatureNotSupported(msg) @@ -293,6 +322,7 @@ async def set_led_mode(self, light_mode: LedMode, *, on: bool = False) -> bool: @needs_auth async def set_timer(self, hours: int, minutes: int, *, enabled: bool) -> bool: + """Set the timer.""" ret_timer = await self._simple_command( EfireCommand.SET_TIMER, bytes([hours, minutes, enabled]) ) @@ -306,6 +336,7 @@ async def set_timer(self, hours: int, minutes: int, *, enabled: bool) -> bool: @needs_auth async def set_led_color(self, color: tuple[int, int, int]) -> bool: + """Set the LED color.""" if not self._features.led_lights: msg = f"Fireplace {self.name} does not have LED controller" raise FeatureNotSupported(msg) @@ -316,6 +347,7 @@ async def set_led_color(self, color: tuple[int, int, int]) -> bool: @needs_auth async def set_led_state(self, *, on: bool) -> bool: + """Set the LED power state.""" if not self._features.led_lights: msg = f"Fireplace {self.name} does not have LED controller" raise FeatureNotSupported(msg) @@ -329,20 +361,24 @@ async def set_led_state(self, *, on: bool) -> bool: @needs_auth async def led_on(self) -> bool: + """Turn LEDs on.""" return await self.set_led_state(on=True) @needs_auth async def led_off(self) -> bool: + """Turn LEDs off.""" return await self.set_led_state(on=False) @needs_auth async def query_aux_control(self) -> AuxControlState: + """Query whether the remote is currently overriding Bluetooth control.""" result = await self.execute_command(EfireCommand.GET_AUX_CTRL) _LOGGER.debug("[%s]: Aux Control State: %x", self.name, result) return AuxControlState(int.from_bytes(result, "big")) @needs_auth async def set_password(self, password: str) -> bool: + """Set the password for authentication.""" # Enter Password management enable_password_mgmt = await self._simple_command( EfireCommand.PASSWORD_MGMT, PasswordAction.SET @@ -357,6 +393,8 @@ async def set_password(self, password: str) -> bool: return False async def reset_password(self) -> bool: + # untested + """Reset the password on the controller.""" return await self._simple_command( EfireCommand.PASSWORD_MGMT, PasswordAction.RESET ) @@ -364,6 +402,7 @@ async def reset_password(self) -> bool: # E0 @needs_auth async def update_led_state(self) -> None: + """Update the power state of the LED Controller.""" result = await self.execute_command(EfireCommand.GET_LED_STATE) self._state.led = ( @@ -374,6 +413,7 @@ async def update_led_state(self) -> None: # E1 @needs_auth async def update_led_color(self) -> None: + """Update the state of the LED colors.""" result = await self.execute_command(EfireCommand.GET_LED_COLOR) self._state.led_color = parse_led_color(result) @@ -381,13 +421,15 @@ async def update_led_color(self) -> None: # E2 @needs_auth async def update_led_controller_mode(self) -> None: + """Update the mode of the LED colors.""" result = await self.execute_command(EfireCommand.GET_LED_MODE) self._state.led_mode = LedMode(int.from_bytes(result, "big")) # pyright: ignore # E3 @needs_auth - async def update_off_state_settings(self) -> None: + async def update_ifc_cmd1_state(self) -> None: + """Update the state of the IFC CMD1 functions.""" result = await self.execute_command(EfireCommand.GET_IFC_CMD1_STATE) if result[0] == ReturnCode.FAILURE: @@ -397,28 +439,31 @@ async def update_off_state_settings(self) -> None: self._state.pilot, self._state.night_light_brightness, self._state.main_mode, - ) = parse_off_state(result) + ) = parse_ifc_cmd1_state(result) # E4 @needs_auth - async def update_on_state_settings(self) -> None: + async def update_ifc_cmd2_state(self) -> None: + """Update the state of the IFC CMD1 functions.""" result = await self.execute_command(EfireCommand.GET_IFC_CMD2_STATE) ( self._state.split_flow, self._state.aux, self._state.blower_speed, self._state.flame_height, - ) = parse_on_state(result) + ) = parse_ifc_cmd2_state(result) # E6 @needs_auth async def update_timer_state(self) -> None: + """Update the state of the timer.""" result = await self.execute_command(EfireCommand.GET_TIMER) self._state.time_left, self._state.timer = parse_timer(result) # E7 @needs_auth async def update_power_state(self) -> None: + """Update the power state of the fireplace.""" result = await self.execute_command(EfireCommand.GET_POWER_STATE) self._state.power = result[0] == PowerState.ON @@ -426,6 +471,7 @@ async def update_power_state(self) -> None: # EB @needs_auth async def update_led_controller_state(self) -> None: + """Update the state of the LED Controller.""" result = await self.execute_command(EfireCommand.GET_LED_CONTROLLER_STATE) ( @@ -437,6 +483,7 @@ async def update_led_controller_state(self) -> None: # EE @needs_auth async def update_remote_usage(self) -> None: + """Update the remote control override state.""" result = await self.execute_command(EfireCommand.GET_REMOTE_USAGE) self._state.remote_in_use = result == AuxControlState.USED @@ -444,6 +491,7 @@ async def update_remote_usage(self) -> None: # F2 @needs_auth async def query_ble_version(self) -> str: + """Update the BLE version information.""" result = await self.execute_command(EfireCommand.GET_BLE_VERSION) return parse_ble_version(result) @@ -451,13 +499,15 @@ async def query_ble_version(self) -> str: # F3 @needs_auth async def query_mcu_version(self) -> str: + """Update the MCU version information.""" result = await self.execute_command(EfireCommand.GET_MCU_VERSION) return parse_mcu_version(result) async def update_state(self) -> None: - await self.update_off_state_settings() - await self.update_on_state_settings() + """Update all state, depending on selected features.""" + await self.update_ifc_cmd1_state() + await self.update_ifc_cmd2_state() await self.update_power_state() if self._features.timer: await self.update_timer_state() @@ -467,5 +517,6 @@ async def update_state(self) -> None: await self.update_led_controller_mode() async def update_firmware_version(self) -> None: + """Update firmware version strings.""" self._state.mcu_version = await self.query_mcu_version() self._state.ble_version = await self.query_ble_version() diff --git a/src/bonaparte/parser.py b/src/bonaparte/parser.py index dab8414..dfac14d 100644 --- a/src/bonaparte/parser.py +++ b/src/bonaparte/parser.py @@ -1,24 +1,29 @@ +"""Parsers used in the Bonaparte library.""" from __future__ import annotations from .const import LedMode, LedState def parse_ble_version(payload: bytes | bytearray) -> str: + """Return a string of the BLE version.""" return str(payload[1]) def parse_mcu_version(payload: bytes | bytearray) -> str: + """Return a string of the MCU version.""" return f"{payload[0]}.{payload[1]}{payload[2]}" -def parse_off_state(payload: bytes | bytearray) -> tuple[bool, int, int]: +def parse_ifc_cmd1_state(payload: bytes | bytearray) -> tuple[bool, int, int]: + """Parse the fireplace state from an IFC CMD1 response.""" pilot = bool((payload[1] >> 7) & 1) night_light = (payload[1] >> 4) & 7 main_mode = payload[1] & 2 return pilot, night_light, main_mode -def parse_on_state(payload: bytes | bytearray) -> tuple[bool, bool, int, int]: +def parse_ifc_cmd2_state(payload: bytes | bytearray) -> tuple[bool, bool, int, int]: + """Parse the fireplace state from an IFC CMD2 response.""" split_flow = bool((payload[1] >> 7) & 1) aux = bool((payload[1] >> 3) & 1) blower_speed = (payload[1] >> 4) & 7 @@ -27,6 +32,7 @@ def parse_on_state(payload: bytes | bytearray) -> tuple[bool, bool, int, int]: def parse_timer(payload: bytes | bytearray) -> tuple[tuple[int, int, int], bool]: + """Parse timer state.""" hours = payload[0] minutes = payload[1] seconds = payload[3] if len(payload) > 3 else 0 # noqa: PLR2004 @@ -36,18 +42,20 @@ def parse_timer(payload: bytes | bytearray) -> tuple[tuple[int, int, int], bool] def parse_led_color(payload: bytes | bytearray) -> tuple[int, int, int]: + """Parse the LED color as RGB.""" return int(payload[0] & 0xFF), int(payload[1] & 0xFF), int(payload[2] & 0xFF) def parse_led_controller_state( payload: bytes | bytearray, ) -> tuple[bool, tuple[int, int, int], LedMode]: + """Parse the LED Controller state.""" light_state = payload[0] == LedState.ON.short # type: ignore[attr-defined] # noqa: E501 pylint: disable=no-member light_color = ( int(payload[1] & 0xFF), int(payload[2] & 0xFF), int(payload[3] & 0xFF), ) - light_mode = LedMode(payload[4]) + light_mode = LedMode(payload[4]) # pyright: ignore return light_state, light_color, light_mode diff --git a/src/bonaparte/utils.py b/src/bonaparte/utils.py index 751c0d6..afa10e1 100644 --- a/src/bonaparte/utils.py +++ b/src/bonaparte/utils.py @@ -1,3 +1,4 @@ +"""Helper functions and utilities used in the Bonaparte library.""" from __future__ import annotations from functools import reduce @@ -6,16 +7,19 @@ def checksum(payload: bytearray | bytes) -> int: + """Calculate the checksum for a command payload.""" # checksum is a single byte XOR of all bytes in the payload return reduce(lambda x, y: x ^ y, payload) def checksum_message(message: bytearray | bytes) -> int: + """Calculate the checksum for a raw message.""" # strip the two header bytes, checksum and footer to get the payload return checksum(message[2:-2]) def build_message(payload: bytearray | bytes) -> bytes: + """Put together a raw message based on a command payload.""" # add the length of the message to the beginning of the payload # it needs to be part of the payload for checksum calculation payload = bytes([len(payload) + 2]) + payload diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..19b95eb 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the Bonaparte library.""" diff --git a/tests/mock_messages.py b/tests/mock_messages.py index cc4b4a8..3854bd3 100644 --- a/tests/mock_messages.py +++ b/tests/mock_messages.py @@ -1,3 +1,5 @@ +"""Collection of raw messages for testing.""" + request = { "login_1234": bytes.fromhex("ab aa 07 c5 31 32 33 34 c6 55"), "query_timer": bytes.fromhex("ab aa 03 e6 e5 55"), diff --git a/tests/test_messages.py b/tests/test_messages.py index ca7673b..6e2d97d 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -1,14 +1,18 @@ +"""Message related tests.""" + from bonaparte.utils import build_message, checksum_message from .mock_messages import request def test_checksum() -> None: + """Test checksum generation using known good messages.""" for message in request.values(): assert checksum_message(message) == message[-2] def test_build_message() -> None: + """Test building a message from a known good message.""" # we use a known message, remove header, length, checksum and footer for message in request.values(): assert build_message(message[3:-2]) == message diff --git a/tests/test_misc.py b/tests/test_misc.py index 2b6995d..caa5176 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -1,3 +1,4 @@ +"""Miscellaneous tests.""" from bleak.backends.device import BLEDevice import pytest @@ -5,7 +6,7 @@ fp = Fireplace(BLEDevice("aa:bb:cc:dd:ee:ff", "Fireplace", delegate="")) -full_valid_set = {"aux", "blower", "led_lights", "night_light", "split_flow"} +full_valid_set = {"aux", "blower", "led_lights", "night_light", "split_flow", "timer"} partial_valid_set = {"blower", "night_light"} full_invalid_set = { @@ -16,11 +17,13 @@ "night_light", "bar", "split_flow", + "timer", } partial_invalid_set = {"blower", "foo", "night_light"} def test_full_valid_featureset() -> None: + """Test a set of all features that is valid.""" fireplace_features = FireplaceFeatures() fireplace_features.aux = True fireplace_features.blower = True @@ -31,6 +34,7 @@ def test_full_valid_featureset() -> None: def test_partial_valid_featureset() -> None: + """Test a partial set of features that is valid.""" fireplace_features = FireplaceFeatures() fireplace_features.aux = False fireplace_features.blower = True @@ -41,6 +45,7 @@ def test_partial_valid_featureset() -> None: def test_full_invalid_featureset() -> None: + """Test a set of all features that also has invalid entries.""" with pytest.raises( ValueError, match=r"Invalid feature values found in input set: {'(foo|bar)', '(foo|bar)'}", @@ -49,6 +54,7 @@ def test_full_invalid_featureset() -> None: def test_partial_invalid_featureset() -> None: + """Test a partial set of features that also has invalid entries.""" with pytest.raises( ValueError, match="Invalid feature value found in input set: {'foo'}" ): diff --git a/tests/test_parsers.py b/tests/test_parsers.py index 2888a44..463bd97 100644 --- a/tests/test_parsers.py +++ b/tests/test_parsers.py @@ -1,11 +1,13 @@ +"""Parsing related tests.""" + from bonaparte.const import LedMode from bonaparte.parser import ( parse_ble_version, + parse_ifc_cmd1_state, + parse_ifc_cmd2_state, parse_led_color, parse_led_controller_state, parse_mcu_version, - parse_off_state, - parse_on_state, parse_timer, ) @@ -13,14 +15,17 @@ def test_ble_version_parser() -> None: + """Test parsing the BLE version from a known message.""" assert parse_ble_version(response["ble_version"][4:-2]) == "8" def test_led_color_parser() -> None: + """Test parsing the LED colors from a known message.""" assert parse_led_color(response["led_color_0000ff"][4:-2]) == (0, 2, 255) def test_led_controller_parser() -> None: + """Test parsing the LED controller state from a known message.""" assert parse_led_controller_state(response["led_controller_state"][4:-2]) == ( True, (0, 0, 255), @@ -29,16 +34,25 @@ def test_led_controller_parser() -> None: def test_mcu_version_parser() -> None: + """Test parsing the MCU version from a known message.""" assert parse_mcu_version(response["mcu_version"][4:-2]) == "1.14" def test_off_state_parser() -> None: - assert parse_off_state(response["off_state_all_off"][4:-2]) == (False, 0, 0) + """Test parsing the power off state from a known message.""" + assert parse_ifc_cmd1_state(response["off_state_all_off"][4:-2]) == (False, 0, 0) def test_on_state_parser() -> None: - assert parse_on_state(response["on_state_all_off"][4:-2]) == (False, False, 0, 0) + """Test parsing the on state from a known message.""" + assert parse_ifc_cmd2_state(response["on_state_all_off"][4:-2]) == ( + False, + False, + 0, + 0, + ) def test_timer_parser() -> None: + """Test parsing the timer state from a known message.""" assert parse_timer(response["timer_201455_on"][4:-2]) == ((20, 14, 55), True)