Fix/code test issues (#210)

* fix daly problem test case
* fix problem test cases
* clean version output
* Update to HA 2025.2.2
* assignment expression
* skip fuzz tests for coverage
* OGT decode error test
* inline crc assignment
* run fuzz tests only without coverage
* add GHA for fuzz testing
* Update README.md

.github/workflows/fuzzing.yaml

@@ -0,0 +1,26 @@
+name: Run fuzz tests
+
+on:
+  pull_request:
+  workflow_dispatch:
+  schedule:
+    - cron:  '1 5 * * *'
+
+jobs:
+  validate:
+    runs-on: "ubuntu-latest"
+    steps:
+        - name: "Checkout the repository" 
+          uses: "actions/checkout@v4"
+
+        - name: "Set up Python"
+          uses: actions/setup-python@main
+          with:
+            python-version: "3.13"
+            cache: "pip"
+
+        - name: Install dependencies
+          run: pip install -r requirements_test.txt
+
+        - name: Run fuzz tests
+          run: pytest tests/test_fuzzing.py --no-cov

README.md

@@ -16,7 +16,7 @@ This integration allows to monitor Bluetooth Low Energy (BLE) battery management
 - Any number of batteries in parallel
 - Native Home Assistant integration (works with all [HA installation methods](https://www.home-assistant.io/installation/#advanced-installation-methods))
 - Readout of individual cell voltages to be able to judge battery health
-- 100% test coverage
+- 100% test coverage plus fuzz tests for BLE data
 
 ### Supported Devices
 - CBT Power BMS, Creabest batteries
@@ -203,6 +203,6 @@ for helping with making the integration better.
 [license-shield]: https://img.shields.io/github/license/patman15/BMS_BLE-HA.svg?style=for-the-badge
 [releases-shield]: https://img.shields.io/github/release/patman15/BMS_BLE-HA.svg?style=for-the-badge
 [releases]: https://github.com//patman15/BMS_BLE-HA/releases
-[effort-shield]: https://img.shields.io/badge/Effort%20spent-391_hours-gold?style=for-the-badge&cacheSeconds=86400
+[effort-shield]: https://img.shields.io/badge/Effort%20spent-395_hours-gold?style=for-the-badge&cacheSeconds=86400
 [install-shield]: https://img.shields.io/badge/dynamic/json?style=for-the-badge&color=green&label=Analytics&suffix=%20Installs&cacheSeconds=15600&url=https://analytics.home-assistant.io/custom_integrations.json&query=$.bms_ble.total
 [btproxy-url]: https://esphome.io/components/bluetooth_proxy

custom_components/bms_ble/plugins/cbtpwr_bms.py

@@ -120,8 +120,9 @@ def _notification_handler(
             self._log.debug("incorrect frame start/end: %s", data)
             return
 
-        crc = crc_sum(data[len(BMS.HEAD) : len(data) + BMS.CRC_POS])
-        if data[BMS.CRC_POS] != crc:
+        if (crc := crc_sum(data[len(BMS.HEAD) : len(data) + BMS.CRC_POS])) != data[
+            BMS.CRC_POS
+        ]:
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 data[len(data) + BMS.CRC_POS],

custom_components/bms_ble/plugins/dpwrcore_bms.py

@@ -147,13 +147,13 @@ async def _notification_handler(
         self._log.debug("(%s): %s", "start" if page == 1 else "cnt.", data)
 
         if page == maxpg:
-            if int.from_bytes(self._data[-4:-2], byteorder="big") != BMS._crc(
-                self._data[3:-4]
+            if (crc := BMS._crc(self._data[3:-4])) != int.from_bytes(
+                self._data[-4:-2], byteorder="big"
             ):
                 self._log.debug(
                     "incorrect checksum: 0x%X != 0x%X",
                     int.from_bytes(self._data[-4:-2], byteorder="big"),
-                    self._crc(self._data[3:-4]),
+                    crc,
                 )
                 self._data = bytearray()
                 self._data_final = bytearray()  # reset invalid data

custom_components/bms_ble/plugins/ecoworthy_bms.py

@@ -109,8 +109,7 @@ def _notification_handler(
             self._log.debug("Invalid frame type: 0x%X", data[0:1])
             return
 
-        crc: Final[int] = crc_modbus(data[:-2])
-        if int.from_bytes(data[-2:], "little") != crc:
+        if (crc := crc_modbus(data[:-2])) != int.from_bytes(data[-2:], "little"):
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 int.from_bytes(data[-2:], "little"),

custom_components/bms_ble/plugins/jbd_bms.py

@@ -134,8 +134,9 @@ def _notification_handler(
             self._log.debug("incorrect frame end (length: %i).", len(self._data))
             return
 
-        crc: Final[int] = BMS._crc(self._data[2 : frame_end - 2])
-        if int.from_bytes(self._data[frame_end - 2 : frame_end], "big") != crc:
+        if (crc := BMS._crc(self._data[2 : frame_end - 2])) != int.from_bytes(
+            self._data[frame_end - 2 : frame_end], "big"
+        ):
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 int.from_bytes(self._data[frame_end - 2 : frame_end], "big"),

custom_components/bms_ble/plugins/jikong_bms.py

@@ -151,8 +151,7 @@ def _notification_handler(
             self._log.debug("wrong data length (%i): %s", len(self._data), self._data)
             self._data = self._data[: BMS.INFO_LEN]
 
-        crc: Final[int] = crc_sum(self._data[:-1])
-        if self._data[-1] != crc:
+        if (crc := crc_sum(self._data[:-1])) != self._data[-1]:
             self._log.debug("invalid checksum 0x%X != 0x%X", self._data[-1], crc)
             return
 

custom_components/bms_ble/plugins/seplos_bms.py

@@ -158,8 +158,9 @@ def _notification_handler(
         if len(self._data) < self._pkglen:
             return
 
-        crc: Final[int] = crc_modbus(self._data[: self._pkglen - 2])
-        if int.from_bytes(self._data[self._pkglen - 2 : self._pkglen], "little") != crc:
+        if (crc := crc_modbus(self._data[: self._pkglen - 2])) != int.from_bytes(
+            self._data[self._pkglen - 2 : self._pkglen], "little"
+        ):
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 int.from_bytes(self._data[self._pkglen - 2 : self._pkglen], "little"),

custom_components/bms_ble/plugins/seplos_v2_bms.py

@@ -138,8 +138,7 @@ def _notification_handler(
             self._log.debug("BMS reported error code: 0x%X", self._data[4])
             return
 
-        crc: Final[int] = crc_xmodem(self._data[1:-3])
-        if int.from_bytes(self._data[-3:-1]) != crc:
+        if (crc := crc_xmodem(self._data[1:-3])) != int.from_bytes(self._data[-3:-1]):
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 int.from_bytes(self._data[-3:-1]),

custom_components/bms_ble/plugins/tdt_bms.py

@@ -147,8 +147,9 @@ def _notification_handler(
             self._log.debug("BMS reported error code: 0x%X", self._data[4])
             return
 
-        crc: Final[int] = crc_modbus(self._data[:-3])
-        if int.from_bytes(self._data[-3:-1], "big") != crc:
+        if (crc := crc_modbus(self._data[:-3])) != int.from_bytes(
+            self._data[-3:-1], "big"
+        ):
             self._log.debug(
                 "invalid checksum 0x%X != 0x%X",
                 int.from_bytes(self._data[-3:-1], "big"),

tests/test_fuzzing.py

@@ -6,6 +6,7 @@
 from hypothesis import HealthCheck, given, settings, strategies as st
 import pytest
 
+from custom_components.bms_ble.const import BMS_TYPES
 from custom_components.bms_ble.plugins.basebms import BaseBMS
 
 from .bluetooth import generate_ble_device
@@ -16,8 +17,9 @@
     data=st.binary(min_size=0, max_size=513)
 )  # ATT is not allowed larger than 512 bytes
 @settings(
-    max_examples=1000, suppress_health_check=[HealthCheck.function_scoped_fixture]
+    max_examples=5000, suppress_health_check=[HealthCheck.function_scoped_fixture]
 )
+@pytest.mark.parametrize("plugin_fixture", BMS_TYPES, indirect=True)
 async def test_notification_handler(
     monkeypatch,
     pytestconfig: pytest.Config,
@@ -26,7 +28,13 @@ async def test_notification_handler(
 ) -> None:
     """Test the notification handler."""
 
-    if pytestconfig.getoption("--cov") == ["custom_components.bms_ble"]:
+    # fuzzing can run from VScode (no coverage) or command line with option --no-cov
+    if {"vscode_pytest", "--cov=."}.issubset(
+        set(pytestconfig.invocation_params.args)
+    ) or (
+        "vscode_pytest" not in pytestconfig.invocation_params.args
+        and not pytestconfig.getoption("--no-cov")
+    ):
         pytest.skip("Skipping fuzzing tests due to coverage generation!")
 
     async def patch_init() -> None: