Merge pull request #67 from cloudblue/bugfix/LITE-22673

LITE-22673 Fixed MESSAGE_TTL in cqrs_dead_letters retry command

Author: maxipavlovic

dj_cqrs/management/commands/cqrs_dead_letters.py

@@ -1,5 +1,6 @@
-#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2022 Ingram Micro Inc. All rights reserved.
 
+from dj_cqrs.constants import DEFAULT_MASTER_MESSAGE_TTL
 from dj_cqrs.dataclasses import TransportPayload
 from dj_cqrs.registries import ReplicaRegistry
 from dj_cqrs.transport import current_transport
@@ -105,7 +106,7 @@ def handle_retry(self, channel, consumer_generator, dead_letters_count):
             dct['retries'] = 0
             if dct.get('expires'):
                 # Message could expire already
-                expires = get_message_expiration_dt()
+                expires = get_message_expiration_dt(DEFAULT_MASTER_MESSAGE_TTL)
                 dct['expires'] = expires.replace(microsecond=0).isoformat()
             payload = TransportPayload.from_message(dct)
             payload.is_requeue = True

dj_cqrs/management/commands/cqrs_sync.py

@@ -66,10 +66,10 @@ def handle(self, *args, **options):
         if progress:
             print('Processing {0} records with batch size {1}'.format(db_count, batch_size))
 
-        for qs in batch_qs(model.relate_cqrs_serialization(qs), batch_size=batch_size):
+        for qs_ in batch_qs(model.relate_cqrs_serialization(qs), batch_size=batch_size):
             ts = time.time()
             cs = counter
-            for instance in qs:
+            for instance in qs_:
                 counter += 1
                 try:
                     instance.cqrs_sync(queue=options['queue'])

dj_cqrs/utils.py

@@ -1,4 +1,4 @@
-#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2022 Ingram Micro Inc. All rights reserved.
 
 import logging
 from datetime import timedelta
@@ -10,13 +10,14 @@
 logger = logging.getLogger('django-cqrs')
 
 
-def get_message_expiration_dt():
+def get_message_expiration_dt(message_ttl=None):
     """Calculates when message should expire.
 
+    :param int or None message_ttl:
     :return: Expiration datetime or None if infinite
     :rtype: datetime.datetime or None
     """
-    message_ttl = settings.CQRS['master']['CQRS_MESSAGE_TTL']
+    message_ttl = message_ttl or settings.CQRS['master']['CQRS_MESSAGE_TTL']
     if message_ttl is None:
         # Infinite
         return

tests/test_commands/test_dead_letters.py

@@ -1,4 +1,4 @@
-#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2022 Ingram Micro Inc. All rights reserved.
 
 from datetime import datetime
 
@@ -61,7 +61,6 @@ def test_handle_retry(settings, capsys, mocker):
     method_frame = mocker.MagicMock()
     method_frame.delivery_tag = 12
 
-    settings.CQRS['master']['CQRS_MESSAGE_TTL'] = 3600
     fake_now = datetime(2020, 1, 1, second=0, tzinfo=timezone.utc)
     mocker.patch('django.utils.timezone.now', return_value=fake_now)
     message = {
@@ -86,15 +85,15 @@ def test_handle_retry(settings, capsys, mocker):
 
     produce_message = ujson.loads(produce_kwargs['body'])
     assert produce_message['instance_data'] == message['instance_data']
-    assert produce_message['expires'] == '2020-01-01T01:00:00+00:00'
+    assert produce_message['expires'] == '2020-01-02T00:00:00+00:00'
     assert produce_message['retries'] == 0
 
     captured = capsys.readouterr()
     total_msg, retrying_msg, body_msg = captured.out.strip().split('\n')
 
     assert total_msg == 'Total dead letters: 1'
     assert retrying_msg == 'Retrying: 1/1'
-    assert '2020-01-01T01:00:00+00:00' in body_msg
+    assert '2020-01-02T00:00:00+00:00' in body_msg
 
     assert channel.basic_nack.call_count == 1
     assert channel.basic_nack.call_args[0][0] == 12

tests/test_utils.py

@@ -1,4 +1,4 @@
-#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2022 Ingram Micro Inc. All rights reserved.
 
 from datetime import datetime, timedelta, timezone
 
@@ -18,6 +18,16 @@ def test_get_message_expiration_dt_fixed(mocker, settings):
     assert result == expected_result
 
 
+def test_get_message_expiration_dt_fixed_from_parameter(mocker, settings):
+    fake_now = datetime(2020, 1, 1, second=0, tzinfo=timezone.utc)
+    mocker.patch('django.utils.timezone.now', return_value=fake_now)
+
+    result = get_message_expiration_dt(message_ttl=2200)
+
+    expected_result = fake_now + timedelta(seconds=2200)
+    assert result == expected_result
+
+
 def test_get_message_expiration_dt_infinite(mocker, settings):
     settings.CQRS['master']['CQRS_MESSAGE_TTL'] = None
     fake_now = datetime(2020, 1, 1, second=0, tzinfo=timezone.utc)