SIENTIAPDE-1163

SIENTIAPDE-1163
Refactor Ingestor and DataManager classes to enhance metadata handling

- Updated Ingestor class to include additional metadata fields: 'workflow_name' and 'schema_name'.
- Modified DataManager to accept metadata during initialization.
- Adjusted unit tests to validate the new metadata structure and ensure proper functionality across various managers.
- Removed unused logger initialization in tests for cleaner code.

ingestor/ingestor.py

@@ -70,10 +70,10 @@ class Ingestor:
         )
 
         self.metadata = {
-            'schedule_name': 'OPC_INGESTOR',
+            'model_id': '-',
-            'trigger': 'INGESTOR',
             'model_name': '-',
-            'model_id': '-'
+            'workflow_name': 'OPC_INGESTOR',
+            'schema_name': 'OPC_INGESTOR',
         }
         self.ingestor_manager = None
 

ingestor/managers/data_manager.py

@@ -87,6 +87,8 @@ class DataManager(BaseActivity):
         self.mongo_client = MongoClient(self.connection_string)
         self.mongo_client.server_info()
 
+        self.metadata = metadata
+
         self.mongo_db = self.mongo_client[self.database]
 
         logger.info(

tests/unit/managers/test_data_manager.py

@@ -4,20 +4,35 @@ from kafka.errors import NoBrokersAvailable
 from sientia_do.notifications.models import NotificationLevel
 from ingestor.managers.data_manager import DataManager
 
+metadata = {
+    "metadata": {
+        "model_id": "test_model",
+        "model_name": "test_model",
+        "workflow_name": "test_workflow",
+        "schema_name": "test_schedule",
+    },
+}
+
 
 @fixture
 @patch("ingestor.managers.data_manager.KafkaProducer")
 @patch("ingestor.managers.data_manager.MongoClient")
 def data_manager(mongo, kafka):
-    return DataManager(
+
+    data_manager = DataManager(
         kafka_servers="localhost:9092",
         mongo_connection_string="mongodb://localhost:27017",
         mongo_database="sientia",
         export_to_kafka=True,
         logger=MagicMock(),
-        notification_handler=MagicMock()
+        notification_handler=MagicMock(),
+        metadata=metadata["metadata"],
     )
 
+    data_manager.send_notification = MagicMock()
+
+    return data_manager
+
 
 @patch("ingestor.managers.data_manager.KafkaProducer")
 @patch("ingestor.managers.data_manager.MongoClient")
@@ -25,6 +40,7 @@ def test___init___success(mongo, kafka):
     logger_mock = MagicMock()
 
     data_manager = DataManager(
+        metadata=metadata["metadata"],
         kafka_servers="localhost:9092",
         mongo_connection_string="mongodb://localhost:27017",
         mongo_database="sientia",
@@ -61,7 +77,8 @@ def test___init___second_attempt(mongo, kafka):
         mongo_database="sientia",
         export_to_kafka=True,
         logger=logger_mock,
-        notification_handler=MagicMock()
+        notification_handler=MagicMock(),
+        metadata=metadata["metadata"],
     )
 
     kafka.assert_any_call(
@@ -99,7 +116,8 @@ def test___init___failure_max_attempts(mongo, kafka):
             mongo_database="sientia",
             export_to_kafka=True,
             logger=logger_mock,
-            notification_handler=MagicMock()
+            notification_handler=MagicMock(),
+            metadata=metadata["metadata"],
         )
     except NoBrokersAvailable as e:
         assert str(
@@ -255,12 +273,13 @@ def test_publish_error(traceback, data_manager):
     )
 
     # Check if the error was logged
-    data_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    data_manager.send_notification.assert_called_once_with(
         notification_id=f"KAFKA_PRODUCER_ERROR_{topic}",
         message=f"Error publishing message to topic {topic}: Test error",
         block="kafka_producer",
         level=NotificationLevel.ERROR,
-        attachment_content=traceback.format_exc.return_value
+        attachment_content=traceback.format_exc.return_value,
+        metadata=metadata["metadata"]
     )
 
 
@@ -271,10 +290,11 @@ def test_publish_error_mongo(data_manager):
 
     data_manager.publish("test_topic", {"key": "value"})
 
-    data_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    data_manager.send_notification.assert_called_once_with(
         notification_id="MONGO_PRODUCER_ERROR_test_topic",
         message="Error inserting message to MongoDB: Test error",
         block="mongo_producer",
         level=NotificationLevel.ERROR,
-        attachment_content=ANY
+        attachment_content=ANY,
+        metadata=metadata["metadata"]
     )

tests/unit/managers/test_ingestor_manager.py

@@ -3,12 +3,21 @@ from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
 from ingestor.managers.ingestor_manager import IngestorManager
 
+metadata = {
+    "metadata": {
+        "model_id": "test_model",
+        "model_name": "test_model",
+        "workflow_name": "test_workflow",
+        "schema_name": "test_schedule",
+    },
+}
+
 
 @fixture
 @patch('ingestor.managers.ingestor_manager.DataManager')
 @patch('ingestor.managers.ingestor_manager.ResourceManager')
 def ingestor_manager(data_manager_mock, resource_manager_mock):
-    return IngestorManager(
+    ingestor = IngestorManager(
         kafka_servers="localhost:9092",
         redis_data={
             "host": "localhost",
@@ -22,9 +31,14 @@ def ingestor_manager(data_manager_mock, resource_manager_mock):
         mongo_database="sientia",
         export_to_kafka=False,
         logger=MagicMock(),
-        notification_handler=MagicMock()
+        notification_handler=MagicMock(),
+        metadata=metadata["metadata"],
     )
 
+    ingestor.send_notification = MagicMock()
+
+    return ingestor
+
 
 @patch('ingestor.managers.ingestor_manager.OpcManager')
 @patch('ingestor.managers.ingestor_manager.DataManager')
@@ -46,15 +60,32 @@ def test___init__(notification_handler_mock, resource_manager_mock, data_manager
         mongo_database="sientia",
         export_to_kafka=False,
         logger=MagicMock(),
-        notification_handler=MagicMock()
+        notification_handler=MagicMock(),
+        metadata=metadata["metadata"],
     )
 
     opc_manager_mock.assert_not_called()
     data_manager_mock.assert_called_once_with(
-        "localhost:9092", "mongodb://localhost:27017", "sientia", False,
+        kafka_servers="localhost:9092",
-        ingestor.logger, ingestor.notification_handler)
+        mongo_connection_string="mongodb://localhost:27017",
+        mongo_database="sientia",
+        export_to_kafka=False,
+        metadata=metadata["metadata"],
+        logger=ingestor.logger,
+        notification_handler=ingestor.notification_handler,
+    )
     resource_manager_mock.assert_called_once_with(
-        "localhost", 6379, 60, 60, "test_pod", None, None)
+        host="localhost",
+        port=6379,
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        pod_id="test_pod",
+        metadata=metadata["metadata"],
+        logger=ingestor.logger,
+        notification_handler=ingestor.notification_handler,
+        username=None,
+        password=None,
+    )
     assert ingestor.poll_interval == 5
     assert ingestor.managed_tags == {}
     assert ingestor.opc_servers == {}
@@ -77,14 +108,20 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager):
 
     opc_manager.return_value = MagicMock()
     result = ingestor_manager.initialize_opc_from_config(
-        server_config, ingestor_manager.data_manager, ingestor_manager.logger)
+        server_config)
 
     opc_manager.assert_called_once_with(
-        server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger,
+        name=server_config['name'],
-        server_config['server_uri'], ingestor_manager.notification_handler,
+        url=server_config['url'],
-        server_config['pod_id'],
+        data_manager=ingestor_manager.data_manager,
-        server_config['cert_path'], server_config['private_key_path'],
+        logger=ingestor_manager.logger,
-        server_config['server_cert_path']
+        server_uri=server_config['server_uri'],
+        notification_handler=ingestor_manager.notification_handler,
+        pod_id=server_config['pod_id'],
+        cert_path=server_config['cert_path'],
+        private_key_path=server_config['private_key_path'],
+        server_cert_path=server_config['server_cert_path'],
+        metadata=metadata["metadata"],
     )
 
     assert result == opc_manager.return_value
@@ -107,12 +144,13 @@ def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, inges
     opc_manager.side_effect = Exception("Initialization error")
 
     result = ingestor_manager.initialize_opc_from_config(
-        server_config, ingestor_manager.data_manager, ingestor_manager.logger)
+        server_config)
 
     assert result is None
 
     traceback_mock.format_exc.assert_called_once()
-    ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    ingestor_manager.send_notification.assert_called_once_with(
+        metadata=metadata["metadata"],
         notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
         message='Error initializing OPC manager: Initialization error',
         block="opc_manager",
@@ -131,7 +169,7 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
     manager3 = MagicMock(
         config={"config": "config3"})
 
-    def mock_initialize_from_config(config, data_manager, logger):
+    def mock_initialize_from_config(config):
         if config == {"config": "config1"}:
             return manager1
         elif config == {"config": "config2"}:
@@ -169,11 +207,11 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
     assert len(ingestor_manager.opc_managers) == 3
 
     ingestor_manager.initialize_opc_from_config.assert_any_call(
-        {"config": "config1"}, ingestor_manager.data_manager, ingestor_manager.logger)
+        {"config": "config1"})
     ingestor_manager.initialize_opc_from_config.assert_any_call(
-        {"config": "config2"}, ingestor_manager.data_manager, ingestor_manager.logger)
+        {"config": "config2"})
     ingestor_manager.initialize_opc_from_config.assert_any_call(
-        {"config": "config5"}, ingestor_manager.data_manager, ingestor_manager.logger)
+        {"config": "config5"})
     assert ingestor_manager.initialize_opc_from_config.call_count == 3
 
     assert ingestor_manager.opc_managers['server1'].config == {
@@ -470,7 +508,8 @@ def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager):
 
     traceback_mock.format_exc.assert_called_once()
 
-    ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    ingestor_manager.send_notification.assert_called_once_with(
+        metadata=metadata["metadata"],
         notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1',
         message='Failed to subscribe to tags from slot1:server1\n{\'tags\': \'config1\'}: Subscription error',
         block="opc_manager",

tests/unit/managers/test_opc_manager.py

@@ -33,6 +33,15 @@ tags = {
     },
 }
 
+metadata = {
+    "metadata": {
+        "model_id": "test_model",
+        "model_name": "test_model",
+        "workflow_name": "test_workflow",
+        "schema_name": "test_schedule",
+    },
+}
+
 
 @fixture
 def raw_opc_manager():
@@ -44,6 +53,7 @@ def raw_opc_manager():
         "opc.tcp://localhost:4840",
         MagicMock(),
         "localhost",
+        metadata=metadata["metadata"],
     )
 
 
@@ -53,6 +63,7 @@ def opc_manager(raw_opc_manager):
     raw_opc_manager.cert_path = "cert.pem"
     raw_opc_manager.private_key_path = "private_key.pem"
     raw_opc_manager.server_cert_path = "server_cert.pem"
+    raw_opc_manager.send_notification = MagicMock()
 
     return raw_opc_manager
 
@@ -414,7 +425,6 @@ def test_check_cycles_no_notification(opc_manager):
             "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 3.0},
         }
     }
-    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
     opc_manager.check_cycles()
 
@@ -422,7 +432,7 @@ def test_check_cycles_no_notification(opc_manager):
     assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][
         "cycle_count"
     ] == pytest.approx(4.0)
-    opc_manager.notification_handler.build_and_send_notification.assert_not_called()
+    opc_manager.send_notification.assert_not_called()
 
 
 def test_check_cycles_triggers_notification(opc_manager):
@@ -441,11 +451,12 @@ def test_check_cycles_triggers_notification(opc_manager):
     assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][
         "cycle_count"
     ] == pytest.approx(5.5)
-    opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    opc_manager.send_notification.assert_called_once_with(
         notification_id="TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED",
         message="5.5 cycles without receive from ns=3;i=1001:Counter",
         block="opc_manager",
         level=NotificationLevel.WARNING,
+        metadata=metadata["metadata"],
     )
 
 
@@ -473,16 +484,16 @@ def test_check_opc_listenning_no_notification(metrics, opc_manager):
 
 def test_check_opc_listenning_warning_notification(opc_manager):
     opc_manager.non_receive_count = 4
-    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
     result = opc_manager.check_opc_listenning()
 
     assert opc_manager.non_receive_count == 5
-    opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+    opc_manager.send_notification.assert_called_once_with(
         notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}",
         message=f"5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}",
         block="opc_manager",
         level=NotificationLevel.ERROR,
+        metadata=metadata["metadata"],
     )
     assert result is False
 
@@ -498,14 +509,15 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager)
 
     assert opc_manager.non_receive_count == 15
     # Should be called twice: once for 5, once for 15
-    assert opc_manager.notification_handler.build_and_send_notification.call_count == 2
+    assert opc_manager.send_notification.call_count == 2
-    calls = opc_manager.notification_handler.build_and_send_notification.call_args_list
+    calls = opc_manager.send_notification.call_args_list
     # First call: 5 cycles warning
     assert calls[0].kwargs == dict(
         notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}",
         message=f"15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}",
         block="opc_manager",
         level=NotificationLevel.ERROR,
+        metadata=metadata["metadata"],
     )
     # Second call: 15 cycles retry
     assert calls[1].kwargs == dict(
@@ -513,6 +525,7 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager)
         message=f"Retrying to connect to server {opc_manager.name}",
         block="opc_manager",
         level=NotificationLevel.ERROR,
+        metadata=metadata["metadata"],
     )
     assert result is True
 
@@ -540,6 +553,7 @@ def test_init_metrics_calls_correct_metric_methods(mock_metrics):
         server_uri="opc.tcp://init.test:4840/uri",
         notification_handler=MagicMock(),
         pod_id="init_pod_localhost",
+        metadata=metadata["metadata"],
     )
 
     mock_metrics.OPC_CONNECTION_STATUS.labels.assert_called_once_with(

tests/unit/managers/test_resource_manager.py

@@ -1,13 +1,36 @@
 from unittest.mock import MagicMock, patch
 from pytest import fixture, raises
+from sientia_do.notifications.models import NotificationLevel
 from ingestor.managers.resource_manager import ResourceManager
 
+metadata = {
+    "metadata": {
+        "model_id": "test_model",
+        "model_name": "test_model",
+        "workflow_name": "test_workflow",
+        "schema_name": "test_schedule",
+    },
+}
+
 
 @fixture
 @patch("ingestor.managers.resource_manager.Redis")
 def resource_manager(redis):
 
-    return ResourceManager("localhost", 6379, 10, 10, "pod_id")
+    resource_manager = ResourceManager(
+        host="localhost",
+        port=6379,
+        lease_ttl=10,
+        heartbeat_ttl=10,
+        pod_id="pod_id",
+        metadata=metadata["metadata"],
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+    )
+
+    resource_manager.send_notification = MagicMock()
+
+    return resource_manager
 
 
 def test_get_success(resource_manager):
@@ -54,7 +77,8 @@ def test_renew_tag_lease_success(resource_manager):
     result = resource_manager.renew_tag_lease("tag_id")
     assert result is True
     resource_manager.redis.get.assert_called_once_with("lease:opc_tags:tag_id")
-    resource_manager.redis.expire.assert_called_once_with("lease:opc_tags:tag_id", 10)
+    resource_manager.redis.expire.assert_called_once_with(
+        "lease:opc_tags:tag_id", 10)
 
 
 def test_renew_tag_lease_failure(resource_manager):
@@ -69,7 +93,8 @@ def test_renew_tag_lease_failure(resource_manager):
 def test_drop_tag_lease(resource_manager):
     resource_manager.redis.delete.return_value = True
     resource_manager.drop_tag_lease("tag_id")
-    resource_manager.redis.delete.assert_called_once_with("lease:opc_tags:tag_id")
+    resource_manager.redis.delete.assert_called_once_with(
+        "lease:opc_tags:tag_id")
 
 
 def test_get_all_ingestors(resource_manager):
@@ -106,7 +131,16 @@ def test_init_connection_failure(monkeypatch):
         mock_metrics.labels.return_value = mock_status
 
         with raises(Exception, match="Connection failed"):
-            ResourceManager("localhost", 6379, 10, 10, "pod_id")
+            ResourceManager(
+                host="localhost",
+                port=6379,
+                lease_ttl=10,
+                heartbeat_ttl=10,
+                pod_id="pod_id",
+                metadata=metadata["metadata"],
+                logger=MagicMock(),
+                notification_handler=MagicMock(),
+            )
 
         mock_metrics.labels.assert_called_once_with(pod_id="pod_id")
         mock_status.set.assert_called_once_with(0)
@@ -118,7 +152,8 @@ def test_init_ping_failure(monkeypatch):
     mock_redis_instance.ping.side_effect = Exception("Ping failed")
 
     mock_redis_class = MagicMock(return_value=mock_redis_instance)
-    monkeypatch.setattr("ingestor.managers.resource_manager.Redis", mock_redis_class)
+    monkeypatch.setattr(
+        "ingestor.managers.resource_manager.Redis", mock_redis_class)
 
     # Test that the exception is raised and metrics are set properly
     with patch("ingestor.metrics.REDIS_CONNECTION_STATUS") as mock_metrics:
@@ -126,7 +161,16 @@ def test_init_ping_failure(monkeypatch):
         mock_metrics.labels.return_value = mock_status
 
         with raises(Exception, match="Ping failed"):
-            ResourceManager("localhost", 6379, 10, 10, "pod_id")
+            ResourceManager(
+                host="localhost",
+                port=6379,
+                lease_ttl=10,
+                heartbeat_ttl=10,
+                pod_id="pod_id",
+                metadata=metadata["metadata"],
+                logger=MagicMock(),
+                notification_handler=MagicMock(),
+            )
 
         mock_metrics.labels.assert_called_once_with(pod_id="pod_id")
         mock_status.set.assert_called_once_with(0)
@@ -172,19 +216,23 @@ def test_execute_redis_op_exception(resource_manager):
 
     with patch("ingestor.managers.resource_manager.time", return_value=100):
         with patch("ingestor.metrics.REDIS_OPERATIONS_ERRORS") as mock_errors:
-            with patch("builtins.print") as mock_print:
+            mock_errors_labels = MagicMock()
-                mock_errors_labels = MagicMock()
+            mock_errors.labels.return_value = mock_errors_labels
-                mock_errors.labels.return_value = mock_errors_labels
 
-                # Execute the operation and expect an exception
+            # Execute the operation and expect an exception
-                with raises(Exception, match="Operation failed"):
+            with raises(Exception, match="Operation failed"):
-                    resource_manager._execute_redis_op("test_op", mock_func, "arg1")
+                resource_manager._execute_redis_op(
+                    "test_op", mock_func, "arg1")
 
-                # Verify metrics and error handling
+            # Verify metrics and error handling
-                mock_errors.labels.assert_called_once_with(
+            mock_errors.labels.assert_called_once_with(
-                    pod_id="pod_id", operation="test_op"
+                pod_id="pod_id", operation="test_op"
-                )
+            )
-                mock_errors_labels.inc.assert_called_once()
+            mock_errors_labels.inc.assert_called_once()
-                mock_print.assert_called_once_with(
+            resource_manager.send_notification.assert_called_once_with(
-                    "Error in Redis operation 'test_op': Operation failed"
+                metadata=metadata["metadata"],
-                )
+                notification_id="REDIS_OPERATION_ERROR_test_op",
+                message="Error in Redis operation 'test_op': Operation failed",
+                block="redis_manager",
+                level=NotificationLevel.ERROR,
+            )

tests/unit/test_ingestor.py

@@ -6,9 +6,8 @@ from ingestor.ingestor import Ingestor
 
 
 @patch("ingestor.ingestor.getenv")
-@patch("ingestor.ingestor.Ingestor.init_logger")
 @patch("ingestor.ingestor.NotificationHandler")
-def test___init__(notification_handler, init_logger, getenv):
+def test___init__(notification_handler, getenv):
     getenv.side_effect = [
         "localhost:9092,localhost:35",  # KAFKA_SERVERS
         "true",  # EXPORT_TO_KAFKA
@@ -20,7 +19,7 @@ def test___init__(notification_handler, init_logger, getenv):
         '200',               # HEARTBEAT_TTL
         "localhost1",      # HOSTNAME
         '50',                # POLL_INTERVAL
-        "mongodb://localhost:27017",  # MONGODB_URL
+        "localhost:27017",  # MONGODB_URL
         "sientia",            # MONGODB_USERNAME
         "sientia",            # MONGODB_PASSWORD
         "sientia"             # MONGODB_DATABASE
@@ -47,10 +46,16 @@ def test___init__(notification_handler, init_logger, getenv):
     assert ingestor.heartbeat_ttl == 200
     assert ingestor.pod_id == "localhost1"
     assert ingestor.poll_interval == 50
+    assert ingestor.metadata == {
+        "model_id": "-",
+        "model_name": "-",
+        "workflow_name": "OPC_INGESTOR",
+        "schema_name": "OPC_INGESTOR",
+    }
 
-    init_logger.assert_called_once()
     notification_handler.assert_called_once_with(
-        servers=["localhost:9092", "localhost:35"],
+        connection_string="mongodb://sientia:sientia@localhost:27017",
+        database="sientia",
         logger=ingestor.logger,
         project_name="OPC_INGESTOR"
     )
@@ -58,9 +63,8 @@ def test___init__(notification_handler, init_logger, getenv):
 
 @fixture
 @patch("ingestor.ingestor.getenv")
-@patch("ingestor.ingestor.Ingestor.init_logger")
 @patch("ingestor.ingestor.NotificationHandler")
-def ingestor(_notification_handler, _init_logger, _getenv):
+def ingestor(_notification_handler, _getenv):
     ing = Ingestor()
     ing.logger = MagicMock()
 
@@ -73,25 +77,6 @@ def ingestor_manager_started(ingestor):
     return ingestor
 
 
-@patch("ingestor.ingestor.getLogger")
-@patch("ingestor.ingestor.StreamHandler")
-@patch("ingestor.ingestor.Formatter")
-def test_init_logger(formatter, stream_handler, get_logger, ingestor):
-    ingestor.logger = None
-    ingestor.init_logger()
-
-    get_logger.assert_called_once_with('ingestor.ingestor')
-    stream_handler.assert_called_once()
-    formatter.assert_called_once_with(
-        '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-    ingestor.logger.setLevel.assert_called_once_with(
-        getenv("LOG_LEVEL", "INFO"))
-    ingestor.logger.addHandler.assert_called_once_with(
-        stream_handler.return_value)
-    stream_handler.return_value.setFormatter.assert_called_once_with(
-        formatter.return_value)
-
-
 def test_handle_acquired_tags_not_acquired(ingestor_manager_started):
     ingestor_manager_started.handle_acquired_tags([])
 
@@ -118,22 +103,23 @@ def test_prepare_ingestor(ingestor_manager_mock, ingestor):
     ingestor.prepare_ingestor()
 
     ingestor_manager_mock.assert_called_once_with(
-        ingestor.kafka_servers,
+        kafka_servers=ingestor.kafka_servers,
-        {
+        redis_data={
             "host": ingestor.redis_host,
             "port": ingestor.redis_port,
             "username": ingestor.redis_username,
             "password": ingestor.redis_password
         },
-        ingestor.lease_ttl,
+        lease_ttl=ingestor.lease_ttl,
-        ingestor.heartbeat_ttl,
+        heartbeat_ttl=ingestor.heartbeat_ttl,
-        ingestor.pod_id,
+        pod_id=ingestor.pod_id,
-        ingestor.poll_interval,
+        poll_interval=ingestor.poll_interval,
-        ingestor.mongo_connection_string,
+        mongo_connection_string=ingestor.mongo_connection_string,
-        ingestor.mongo_database,
+        mongo_database=ingestor.mongo_database,
-        ingestor.logger,
+        metadata=ingestor.metadata,
-        ingestor.notification_handler,
+        logger=ingestor.logger,
-        ingestor.export_to_kafka
+        notification_handler=ingestor.notification_handler,
+        export_to_kafka=ingestor.export_to_kafka,
     )
     ingestor_manager.declare_active.assert_called_once()
     ingestor_manager.get_slot_leases.assert_called_once()