From b4732a075dd1598b3878eb869eb173441c0a5056 Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Wed, 17 Jun 2026 01:41:56 +0300 Subject: [PATCH 1/2] test: add manual nack tests --- tests/test_ack_policy.py | 65 ++++++++++++++++++++++++++++++++++++---- tests/test_consume.py | 30 +++++++++++++++++-- 2 files changed, 87 insertions(+), 8 deletions(-) diff --git a/tests/test_ack_policy.py b/tests/test_ack_policy.py index bd8b551..dcc2560 100644 --- a/tests/test_ack_policy.py +++ b/tests/test_ack_policy.py @@ -31,7 +31,7 @@ async def test_consume_nack_on_error( event: asyncio.Event, broker: SqlBroker, ) -> None: - """Message was Nack'ed.""" + """Message was Nacked.""" @broker.subscriber( queues=["default1"], @@ -89,7 +89,7 @@ async def test_consume_reject_on_error( event: asyncio.Event, broker: SqlBroker, ) -> None: - """Message was Reject'ed despite the retry strategy.""" + """Message was Rejected despite the retry strategy.""" @broker.subscriber( queues=["default1"], @@ -151,7 +151,7 @@ async def test_consume_ack_and_ack_first( ack_policy: AckPolicy, broker: SqlBroker, ) -> None: - """Message was Ack'ed despite the error.""" + """Message was Acked despite the error.""" @broker.subscriber( queues=["default1"], @@ -209,7 +209,7 @@ async def test_consume_manual( event: asyncio.Event, broker: SqlBroker, ) -> None: - """Message was manually Ack'ed.""" + """Message was manually Acked.""" @broker.subscriber( queues=["default1"], @@ -270,7 +270,7 @@ async def test_consume_manual_overrides_policy( event: asyncio.Event, broker: SqlBroker, ) -> None: - """Message was manually Ack'ed.""" + """Message was manually Acked.""" @broker.subscriber( queues=["default1"], @@ -328,7 +328,7 @@ async def test_consume_manual_no_manual_ack( self, engine: AsyncEngine, recreate_tables: None, event: asyncio.Event ) -> None: """Error was logged because the message was neither manually nor automatically acknowledged. - The message was Reject'ed. + The message was Rejected. """ logger = MagicMock() async with self.get_broker(engine=engine, logger=logger) as broker: @@ -367,3 +367,56 @@ async def handler(msg: SqlBrokerMessageAnnotation) -> None: assert result["queue"] == "default1" assert json.loads(result["payload"]) == {"message": "hello1"} assert result["state"] == SqlBrokerMessageState.FAILED.name + + @pytest.mark.asyncio() + async def test_consume_manual_nack_respects_policy( + self, + engine: AsyncEngine, + recreate_tables: None, + event: asyncio.Event, + ) -> None: + async with self.get_broker(engine=engine) as broker: + attempted = [] + + @broker.subscriber( + queues=["default1"], + max_workers=1, + retry_strategy=ConstantRetryStrategy( + delay_seconds=0, max_total_delay_seconds=None, max_attempts=3 + ), + max_fetch_interval=0.01, + min_fetch_interval=0, + fetch_batch_size=5, + overfetch_factor=1, + flush_interval=0.1, + release_stuck_interval=10, + release_stuck_timeout=10, + max_deliveries=20, + ack_policy=AckPolicy.NACK_ON_ERROR, + ) + async def handler(msg: Any) -> None: + nonlocal attempted + attempted.append(msg) + await msg.nack() + return 1 / 0 + + await broker.publish({"message": "hello1"}, queue="default1") + await broker.start() + + await asyncio.sleep(0.5) + + assert len(attempted) == 3 + + async with engine.begin() as conn: + result = await conn.execute(text("SELECT * FROM message_archive;")) + + result = result.mappings().one() + assert result["queue"] == "default1" + assert json.loads(result["payload"]) == {"message": "hello1"} + assert result["state"] == SqlBrokerMessageState.FAILED.name + assert result["attempts_count"] == 3 + assert result["deliveries_count"] == 3 + + async with engine.begin() as conn: + result = await conn.execute(text("SELECT * FROM message;")) + assert len(result.all()) == 0 diff --git a/tests/test_consume.py b/tests/test_consume.py index ad359be..2ab262d 100644 --- a/tests/test_consume.py +++ b/tests/test_consume.py @@ -706,7 +706,7 @@ async def test_consume_manual_nack_takes_precedence( queues=["default1"], max_workers=2, retry_strategy=ConstantRetryStrategy( - delay_seconds=0, max_total_delay_seconds=None, max_attempts=3 + delay_seconds=10, max_total_delay_seconds=None, max_attempts=3 ), max_fetch_interval=10, min_fetch_interval=10, @@ -716,7 +716,7 @@ async def test_consume_manual_nack_takes_precedence( release_stuck_interval=10, release_stuck_timeout=10, max_deliveries=20, - ack_policy=AckPolicy.NACK_ON_ERROR, + ack_policy=AckPolicy.ACK, ) async def handler(msg: SqlBrokerMessageAnnotation, msg_body: dict) -> None: await msg.nack() @@ -734,6 +734,32 @@ async def handler(msg: SqlBrokerMessageAnnotation, msg_body: dict) -> None: assert result[0]["state"] == SqlBrokerMessageState.RETRYABLE.name assert result[1]["state"] == SqlBrokerMessageState.RETRYABLE.name + result = result[0] + assert result["queue"] == "default1" + assert json.loads(result["payload"]) == {"message": "hello1"} + assert result["state"] == SqlBrokerMessageState.RETRYABLE.name + assert result["attempts_count"] == 1 + assert result["deliveries_count"] == 1 + assert as_datetime(result["created_at"]) < datetime.now(tz=timezone.utc).replace( + tzinfo=None + ).replace(tzinfo=None) + assert as_datetime(result["first_attempt_at"]) < datetime.now( + tz=timezone.utc + ).replace(tzinfo=None).replace(tzinfo=None) + assert as_datetime(result["first_attempt_at"]) > as_datetime(result["created_at"]) + assert as_datetime(result["last_attempt_at"]) == as_datetime( + result["first_attempt_at"] + ) + + assert as_datetime(result["next_attempt_at"]) >= as_datetime( + result["first_attempt_at"] + ) + timedelta(seconds=10) + assert result["acquired_at"] is None + + async with engine.begin() as conn: + result = await conn.execute(text("SELECT * FROM message_archive;")) + assert len(result.all()) == 0 + @pytest.mark.asyncio() async def test_consume_manual_reject_takes_precedence( self, From 47bb08510c6410ba022ccba1b39f81cf4f71fe2c Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Wed, 17 Jun 2026 16:52:27 +0300 Subject: [PATCH 2/2] test: add manual nack tests --- tests/test_ack_policy.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_ack_policy.py b/tests/test_ack_policy.py index dcc2560..0e39993 100644 --- a/tests/test_ack_policy.py +++ b/tests/test_ack_policy.py @@ -398,7 +398,6 @@ async def handler(msg: Any) -> None: nonlocal attempted attempted.append(msg) await msg.nack() - return 1 / 0 await broker.publish({"message": "hello1"}, queue="default1") await broker.start()