Merge pull request #27 from cloudblue/feature-2/LITE-17209

LITE-17209 Added support for passing of Correlation ID in CQRS payload

Author: maxipavlovic

dj_cqrs/correlation.py

@@ -0,0 +1,22 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+from django.conf import settings
+
+
+_correlation_function = getattr(settings, 'CQRS', {}).get('master', {}).get('correlation_function')
+if _correlation_function and (not callable(_correlation_function)):
+    raise AttributeError('CQRS correlation_function must be callable.')
+
+
+def get_correlation_id(signal_type, cqrs_id, instance_pk, queue):
+    """
+    :param signal_type: Type of the signal for this message.
+    :type signal_type: dj_cqrs.constants.SignalType
+    :param cqrs_id: The unique CQRS identifier of the model.
+    :type cqrs_id: str
+    :param instance_pk: Primary key of the instance.
+    :param queue: Queue to synchronize, defaults to None
+    :type queue: str, optional
+    """
+    if _correlation_function:
+        return _correlation_function(signal_type, cqrs_id, instance_pk, queue)

dj_cqrs/dataclasses.py

@@ -1,4 +1,6 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+from dj_cqrs.correlation import get_correlation_id
 
 
 class TransportPayload:
@@ -11,23 +13,38 @@ class TransportPayload:
     :param instance_data: Serialized data of the instance that
                             generates the event.
     :type instance_data: dict
-    :param instance_pk: Primary key of the instance
-    :param queue: Queue to syncronize, defaults to None
+    :param instance_pk: Primary key of the instance.
+    :param queue: Queue to synchronize, defaults to None.
     :type queue: str, optional
     :param previous_data: Previous values for fields tracked for changes,
-                                defaults to None
+                                defaults to None.
     :type previous_data: dict, optional
+    :param correlation_id: Correlation ID of process, where this payload is used.
+    :type correlation_id: str, optional
     """
 
-    def __init__(self, signal_type, cqrs_id, instance_data, instance_pk, queue=None,
-                 previous_data=None):
+    def __init__(
+            self,
+            signal_type,
+            cqrs_id,
+            instance_data,
+            instance_pk,
+            queue=None,
+            previous_data=None,
+            correlation_id=None,
+    ):
         self.__signal_type = signal_type
         self.__cqrs_id = cqrs_id
         self.__instance_data = instance_data
         self.__instance_pk = instance_pk
         self.__queue = queue
         self.__previous_data = previous_data
 
+        if correlation_id:
+            self.__correlation_id = correlation_id
+        else:
+            self.__correlation_id = get_correlation_id(signal_type, cqrs_id, instance_pk, queue)
+
     @property
     def signal_type(self):
         return self.__signal_type
@@ -52,6 +69,10 @@ def queue(self):
     def previous_data(self):
         return self.__previous_data
 
+    @property
+    def correlation_id(self):
+        return self.__correlation_id
+
     def to_dict(self):
         """
         Return the payload as a dictionary.
@@ -65,4 +86,5 @@ def to_dict(self):
             'instance_data': self.__instance_data,
             'previous_data': self.__previous_data,
             'instance_pk': self.__instance_pk,
+            'correlation_id': self.__correlation_id,
         }

dj_cqrs/transport/kombu.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 import logging
 
@@ -124,8 +124,12 @@ def _consume_message(cls, body, message):
             return
 
         payload = TransportPayload(
-            dct['signal_type'], dct['cqrs_id'], dct['instance_data'], dct.get('instance_pk'),
+            dct['signal_type'],
+            dct['cqrs_id'],
+            dct['instance_data'],
+            dct.get('instance_pk'),
             previous_data=dct.get('previous_data'),
+            correlation_id=dct.get('correlation_id'),
         )
 
         cls.log_consumed(payload)

dj_cqrs/transport/rabbit_mq.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 import logging
 import time
@@ -108,8 +108,12 @@ def _consume_message(cls, ch, method, properties, body):
             return
 
         payload = TransportPayload(
-            dct['signal_type'], dct['cqrs_id'], dct['instance_data'], dct.get('instance_pk'),
+            dct['signal_type'],
+            dct['cqrs_id'],
+            dct['instance_data'],
+            dct.get('instance_pk'),
             previous_data=dct.get('previous_data'),
+            correlation_id=dct.get('correlation_id'),
         )
 
         cls.log_consumed(payload)

tests/test_controller.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 import pytest
 
@@ -19,6 +19,7 @@ def test_producer(mocker):
         'instance_data': {},
         'instance_pk': 'c',
         'previous_data': {'e': 'f'},
+        'correlation_id': None,
     }
 
 

tests/test_correlation.py

@@ -0,0 +1,35 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+from importlib import import_module, reload
+
+import pytest
+
+from dj_cqrs.correlation import get_correlation_id
+
+
+def test_default_correlation():
+    assert get_correlation_id(None, None, None, None) is None
+
+
+def test_wrong_correlation_type_in_settings(settings):
+    previous_cqrs_settings = settings.CQRS
+    settings.CQRS = {'master': {'correlation_function': 1}}
+
+    with pytest.raises(AttributeError) as e:
+        reload(import_module('dj_cqrs.correlation'))
+
+    assert str(e.value) == 'CQRS correlation_function must be callable.'
+
+    settings.CQRS = previous_cqrs_settings
+    reload(import_module('dj_cqrs.correlation'))
+
+
+def test_custom_correlation(settings):
+    previous_cqrs_settings = settings.CQRS
+    settings.CQRS = {'master': {'correlation_function': lambda *args: '1q2w3e'}}
+
+    reload(import_module('dj_cqrs.correlation'))
+    assert get_correlation_id(None, None, None, None) == '1q2w3e'
+
+    settings.CQRS = previous_cqrs_settings
+    reload(import_module('dj_cqrs.correlation'))

tests/test_transport/test_kombu.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 import logging
 import ujson
@@ -146,6 +146,7 @@ def test_produce_message_ok(mocker):
             'instance_data': {},
             'instance_pk': 'id',
             'previous_data': {'e': 'f'},
+            'correlation_id': None,
         }
 
     assert prepare_message_args[2] == 'text/plain'
@@ -174,6 +175,7 @@ def test_produce_sync_message_no_queue(mocker):
             'instance_data': {},
             'instance_pk': None,
             'previous_data': None,
+            'correlation_id': None,
         }
     assert basic_publish_kwargs['routing_key'] == 'cqrs_id'
 
@@ -195,6 +197,7 @@ def test_produce_sync_message_queue(mocker):
             'instance_data': {},
             'instance_pk': 'id',
             'previous_data': None,
+            'correlation_id': None,
         }
     assert basic_publish_kwargs['routing_key'] == 'cqrs.queue.cqrs_id'
 
@@ -206,7 +209,7 @@ def test_consume_message_ack(mocker, caplog):
 
     PublicKombuTransport.consume_message(
         '{"signal_type":"signal","cqrs_id":"cqrs_id","instance_data":{},'
-        '"instance_pk":1, "previous_data":{}}',
+        '"instance_pk":1, "previous_data":{}, "correlation_id":"zyx"}',
         message_mock,
     )
 
@@ -219,6 +222,7 @@ def test_consume_message_ack(mocker, caplog):
     assert payload.instance_data == {}
     assert payload.previous_data == {}
     assert payload.pk == 1
+    assert payload.correlation_id == 'zyx'
 
     assert 'CQRS is received: pk = 1 (cqrs_id).' in caplog.text
     assert 'CQRS is applied: pk = 1 (cqrs_id).' in caplog.text

tests/test_transport/test_rabbit_mq.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 import logging
 import ujson
@@ -182,6 +182,7 @@ def test_produce_message_ok(mocker):
             'instance_data': {},
             'instance_pk': 'id',
             'previous_data': {'e': 'f'},
+            'correlation_id': None,
         }
     assert basic_publish_kwargs['exchange'] == 'exchange'
     assert basic_publish_kwargs['mandatory']
@@ -204,6 +205,7 @@ def test_produce_sync_message_no_queue(mocker):
             'instance_data': {},
             'instance_pk': None,
             'previous_data': None,
+            'correlation_id': None,
         }
     assert basic_publish_kwargs['routing_key'] == 'cqrs_id'
 
@@ -222,6 +224,7 @@ def test_produce_sync_message_queue(mocker):
             'instance_data': {},
             'instance_pk': 'id',
             'previous_data': None,
+            'correlation_id': None,
         }
     assert basic_publish_kwargs['routing_key'] == 'cqrs.queue.cqrs_id'
 
@@ -259,7 +262,7 @@ def test_consume_message_ack(mocker, caplog):
         mocker.MagicMock(),
         None,
         '{"signal_type":"signal","cqrs_id":"cqrs_id","instance_data":{},'
-        '"instance_pk":1, "previous_data":{}}',
+        '"instance_pk":1, "previous_data":{}, "correlation_id":"abc"}',
     )
 
     assert consumer_mock.call_count == 1
@@ -270,6 +273,7 @@ def test_consume_message_ack(mocker, caplog):
     assert payload.instance_data == {}
     assert payload.previous_data == {}
     assert payload.pk == 1
+    assert payload.correlation_id == 'abc'
 
     assert 'CQRS is received: pk = 1 (cqrs_id).' in caplog.text
     assert 'CQRS is applied: pk = 1 (cqrs_id).' in caplog.text