diff --git a/tests/test_ack_policy.py b/tests/test_ack_policy.py index bd8b551..0e39993 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,55 @@ 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() + + 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,