Use strict zip

bellows/ash.py

@@ -147,8 +147,9 @@ class DataFrame(AshFrame):
 
     @staticmethod
     def _randomize(data: bytes) -> bytes:
-        assert len(data) <= len(PSEUDO_RANDOM_DATA_SEQUENCE)
-        return bytes([a ^ b for a, b in zip(data, PSEUDO_RANDOM_DATA_SEQUENCE)])
+        return bytes(
+            [a ^ b for a, b in zip(data, PSEUDO_RANDOM_DATA_SEQUENCE, strict=True)]
+        )
 
     @classmethod
     def from_bytes(cls, data: bytes) -> DataFrame:

bellows/types/__init__.py

@@ -12,4 +12,4 @@ def deserialize(data, schema):
 
 
 def serialize(data, schema):
-    return b"".join(t(v).serialize() for t, v in zip(schema, data))
+    return b"".join(t(v).serialize() for t, v in zip(schema, data, strict=True))

bellows/zigbee/application.py

@@ -1000,7 +1000,9 @@ async def _watchdog_feed(self):
                 else:
                     (res,) = await self._ezsp.readAndClearCounters()
 
-                for cnt_type, value in zip(self._ezsp.types.EmberCounterType, res):
+                for cnt_type, value in zip(
+                    self._ezsp.types.EmberCounterType, res, strict=True
+                ):
                     counters[cnt_type.name[8:]].update(value)
 
                 if remainder == 0:

tests/test_application.py

@@ -1592,7 +1592,7 @@ async def test_startup_new_coordinator_no_groups_joined(app, ieee):
 )
 async def test_energy_scanning(app, scan_results):
     app._ezsp.startScan = AsyncMock(
-        return_value=list(zip(range(11, 26 + 1), scan_results))
+        return_value=list(zip(range(11, 26 + 1), scan_results, strict=True))
     )
 
     results = await app.energy_scan(