.

Author: hampsterx

kinesis/consumer.py

@@ -928,9 +928,7 @@ async def __anext__(self):
                     log.debug("Skipping checkpoint for deallocated shard %s", cp_shard)
                 elif self.checkpointer:
                     # Don't execute now — defer to next __anext__ call
-                    self._deferred_checkpoints[cp_shard] = item[
-                        "__CHECKPOINT__"
-                    ]["SequenceNumber"]
+                    self._deferred_checkpoints[cp_shard] = item["__CHECKPOINT__"]["SequenceNumber"]
                 checkpoint_count += 1
                 if checkpoint_count >= max_checkpoints:
                     log.warning(f"Processed {max_checkpoints} checkpoints, stopping iteration")

tests/conftest.py

@@ -199,9 +199,7 @@ def _factory(**kwargs):
                 # Don't re-raise: the test itself is responsible for observing
                 # expected task failures (e.g. test_wait_ready_fetch_task_crash).
                 if not task.cancelled() and task.exception():
-                    log.warning(
-                        "Task finished with error during test: %s", task.exception(), exc_info=task.exception()
-                    )
+                    log.warning("Task finished with error during test: %s", task.exception(), exc_info=task.exception())
             else:
                 task.cancel()
                 try:

tests/test_checkpoint_ordering.py

@@ -278,10 +278,12 @@ async def test_shard_exhaustion_flushes_checkpoint_before_deallocate(self, mock_
         await consumer.fetch()
 
         # Checkpoint must have been called before deallocate
-        checkpointer.assert_has_calls([
-            call.checkpoint("shard-0", "100"),
-            call.deallocate("shard-0"),
-        ])
+        checkpointer.assert_has_calls(
+            [
+                call.checkpoint("shard-0", "100"),
+                call.deallocate("shard-0"),
+            ]
+        )
 
     @pytest.mark.asyncio
     async def test_shard_exhaustion_with_records_no_sentinel_enqueued(self, mock_consumer):
@@ -380,9 +382,7 @@ async def test_checkpoint_interval_debounces(self, mock_consumer):
 
         for i in range(5):
             await consumer.queue.put({"msg": f"r{i}"})
-            await consumer.queue.put(
-                {"__CHECKPOINT__": {"ShardId": "shard-0", "SequenceNumber": str((i + 1) * 100)}}
-            )
+            await consumer.queue.put({"__CHECKPOINT__": {"ShardId": "shard-0", "SequenceNumber": str((i + 1) * 100)}})
 
         for _ in range(5):
             await consumer.__anext__()