SIENTIAPDE-988

Enhance ingestor and manager classes with notification handling

- Integrated NotificationHandler into Ingestor, DataManager, IngestorManager, and OpcManager for improved error reporting and monitoring.
- Updated methods to send notifications on critical events such as Kafka publishing errors, OPC connection issues, and cycle count warnings.
- Refactored related tests to ensure coverage of new notification functionalities and validate integration with existing components.
- Improved logging and error handling across the system to enhance traceability and operational insights.

ingestor/ingestor.py

@@ -2,6 +2,7 @@ from logging import Formatter, StreamHandler, getLogger
 from os import getenv
 
 from ingestor.managers.ingestor_manager import IngestorManager
+from sientia_do.notifications.handlers import NotificationHandler
 
 
 class Ingestor:
@@ -37,14 +38,27 @@ class Ingestor:
         self.poll_interval = int(getenv("POLL_INTERVAL", 5))
 
         self.kafka_servers = kafka_servers.split(",")
+        self.logger = None
         self.init_logger()
+        self.notification_handler = NotificationHandler(
+            servers=self.kafka_servers,
+            logger=self.logger,
+            project_name="OPC_INGESTOR",
+            pipeline_name="-",
+            trigger_name="-",
+            model_name="-",
+            model="-"
+        )
+        # build args for build notificarions components
+
+        # call build notifications components
 
     def init_logger(self):
         """
         Initializes a logger instance for the class.
-        This method sets up a logger with a specified log level, a stream handler, 
-        and a formatter. The log level is determined by the environment variable 
-        "LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached 
+        This method sets up a logger with a specified log level, a stream handler,
+        and a formatter. The log level is determined by the environment variable
+        "LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached
         to the instance for use throughout the class.
         Attributes:
             self.logger (logging.Logger): The configured logger instance.
@@ -105,7 +119,8 @@ class Ingestor:
 
         self.ingestor_manager = IngestorManager(
             self.kafka_servers, self.redis_host, self.redis_port, self.lease_ttl,
-            self.heartbeat_ttl, self.pod_id, self.poll_interval, self.logger, self.redis_username, self.redis_password
+            self.heartbeat_ttl, self.pod_id, self.poll_interval, self.logger,
+            self.notification_handler, self.redis_username, self.redis_password
         )
 
         # Declare ingestor ative
@@ -120,9 +135,9 @@ class Ingestor:
     def manage_no_slots(self, number_of_slots: int):
         """
         Manages the scenario where there are no slots assigned to the ingestor.
-        This method checks if the ingestor is active (i.e., has no managed tags) 
-        and if the number of available slots is greater than zero. If both 
-        conditions are met, it attempts to acquire a slot lease and handles 
+        This method checks if the ingestor is active (i.e., has no managed tags)
+        and if the number of available slots is greater than zero. If both
+        conditions are met, it attempts to acquire a slot lease and handles
         the acquired tags accordingly.
         Args:
             number_of_slots (int): The number of available slots.
@@ -138,16 +153,16 @@ class Ingestor:
 
     def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int):
         """
-        Manages the allocation and deallocation of slot leases for ingestors based on 
+        Manages the allocation and deallocation of slot leases for ingestors based on
         the number of available slots, lacking ingestors, and slot differences.
         Args:
             available_slots (int): The number of slots currently available for allocation.
             lacking_ingestors (int): The number of ingestors that are active and without slots.
             slot_diff (int): The difference between the total slots and the required slots.
         Behavior:
-            - If there are available slots and lacking ingestors, attempts to acquire slot leases 
+            - If there are available slots and lacking ingestors, attempts to acquire slot leases
               for the available slots and processes the acquired tags.
-            - If there are no lacking ingestors but there are extra slots (slot_diff > 0), 
+            - If there are no lacking ingestors but there are extra slots (slot_diff > 0),
               releases the extra slot leases to ensure proper allocation.
         Logs:
             - Logs the number of available slots when attempting to acquire leases.
@@ -223,3 +238,6 @@ class Ingestor:
 
         # Update opc servers
         self.ingestor_manager.update_slot_config()
+
+        # Check OPC cycles
+        self.ingestor_manager.check_opc_servers_integrity()

ingestor/managers/data_manager.py

@@ -3,10 +3,14 @@ from logging import Logger
 from time import sleep
 from kafka import KafkaProducer
 from kafka.errors import NoBrokersAvailable
+from sientia_do.notifications.handlers import NotificationHandler
+from sientia_do.notifications.models import NotificationLevel
+import traceback
 
 
 class DataManager():
-    def __init__(self, kafka_servers: str, logger: Logger) -> None:
+    def __init__(self, kafka_servers: str, logger: Logger,
+                 notification_handler: NotificationHandler) -> None:
         """
         Initializes the DataManager instance with a Kafka producer.
         This constructor attempts to establish a connection to the specified Kafka servers
@@ -45,6 +49,7 @@ class DataManager():
         logger.info(
             f"DataManager initialized with Kafka servers: {kafka_servers}")
         self.logger = logger
+        self.notification_handler = notification_handler
 
     def __del__(self):
         """Destructor to close the producer connection."""
@@ -91,4 +96,12 @@ class DataManager():
             self.kafka_producer.flush(timeout=10)
 
         except Exception as e:
-            self.logger.error(f"Failed to publish message: {e}")
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"KAFKA_PRODUCER_ERROR_{topic}",
+                message=f"Error publishing message to topic {topic}: {e}",
+                block="kafka_producer",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)

ingestor/managers/ingestor_manager.py

@@ -5,16 +5,19 @@ from copy import deepcopy
 from ingestor.managers.data_manager import DataManager
 from ingestor.managers.opc_manager import OpcManager
 from ingestor.managers.resource_manager import ResourceManager
+from sientia_do.notifications.models import Notification, NotificationLevel
+from sientia_do.notifications.handlers import NotificationHandler
 
 
 class IngestorManager():
     def __init__(self,
                  kafka_servers: str, redis_host: str, redis_port: int,
                  lease_ttl: int, heartbeat_ttl: int,  pod_id: str,
-                 poll_interval: int, logger: Logger,
+                 poll_interval: int, logger: Logger, notification_handler: NotificationHandler,
                  redis_username: str = None, redis_password: str = None):
 
-        self.data_manager = DataManager(kafka_servers, logger)
+        self.data_manager = DataManager(
+            kafka_servers, logger, notification_handler)
         self.opc_managers = {}
         self.resource_manager = ResourceManager(
             redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id, redis_username, redis_password
@@ -25,6 +28,8 @@ class IngestorManager():
         self.managed_tags = {}
         self.opc_servers = {}
 
+        self.notification_handler = notification_handler
+
     def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None:
         """
         Initializes an OPC Manager instance using the provided server configuration.
@@ -56,8 +61,17 @@ class IngestorManager():
             manager.config = server_config
             manager.connect()
         except Exception as e:
-            logger.error(f"Failed to initialize OpcManager: {e}")
-            traceback.print_exc()
+
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
+                message=f'Error initializing OPC manager: {e}',
+                block="opc_manager",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+
             return None
 
         return manager
@@ -100,6 +114,7 @@ class IngestorManager():
                     )
 
                 elif self.opc_managers[server].config != server_config:
+                    self.opc_managers[server].disconnect()
                     del self.opc_managers[server]
                     server_instance = self.initialize_opc_from_config(
                         server_config, self.data_manager, self.logger
@@ -117,6 +132,30 @@ class IngestorManager():
                 self.opc_managers[server].disconnect()
                 self.opc_managers.pop(server, None)
 
+    def check_opc_servers_integrity(self):
+        """
+        Checks the integrity of the OPC servers and updates the OPC servers if necessary.
+        """
+        for server, opc_manager in self.opc_managers.items():
+            opc_manager.check_cycles()
+
+            is_lost = opc_manager.check_opc_listenning()
+            if is_lost:
+                self.logger.warning(
+                    f"OPC server {server} is lost. "
+                    f"Desconnecting from server."
+                )
+                opc_manager.disconnect()
+                self.opc_managers[server] = self.initialize_opc_from_config(
+                    opc_manager.config, self.data_manager, self.logger
+                )
+                self.update_opc_servers()
+                for slot, slot_config in self.managed_tags.items():
+                    if server in slot_config:
+                        self.manage_server(
+                            slot, server, slot_config[server], slot_config[server]['tags']
+                        )
+
     def declare_active(self):
         """
         Declares the ingestor as active by sending a heartbeat signal to the resource manager.
@@ -342,10 +381,16 @@ class IngestorManager():
                 tags_to_sub
             )
         except Exception as e:
-            self.logger.error(
-                f"Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}"
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f'OPC_SUBSCRIPTION_ERROR_{slot}:{server}',
+                message=f'Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}',
+                block="opc_manager",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
             )
-            self.logger.error(traceback.format_exc())
+            self.logger.error(trace)
+
             self.logger.warning(
                 "Removing subscription from server "
                 f"{server} for slot {slot}"

ingestor/managers/opc_manager.py

@@ -5,14 +5,14 @@ from typing import Callable
 from asyncua.crypto.security_policies import SecurityPolicyBasic256
 from asyncua.sync import Client
 from sientia_do.notifications.models import NotificationLevel
-
+from sientia_do.notifications.handlers import NotificationHandler
 from ingestor.managers.data_manager import DataManager
 
 
 class OpcManager():
     def __init__(self, name: str, url: str, data_manager: DataManager,
-                 logger: Logger, server_uri: str, cert_path: str = None,
-                 private_key_path: str = None, server_cert_path: str = None):
+                 logger: Logger, server_uri: str, notification_handler: NotificationHandler,
+                 cert_path: str = None, private_key_path: str = None, server_cert_path: str = None):
         self.url = url
         self.name = name
         self.server_uri = server_uri
@@ -27,6 +27,8 @@ class OpcManager():
         self.subscriptions = {}
         self.data_manager = data_manager
 
+        self.notification_handler = notification_handler
+
     def __str__(self):
         return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
             f"nodes={self.nodes}, subscriptions={self.subscriptions}"
@@ -233,70 +235,48 @@ class OpcManager():
         [self.data_manager.publish(e, data)
          for e in self.nodes[tag]['topics']]
 
-    def check_cycles(self, removed_data: dict, node: str, config: dict,
-                     handle_listen_events: Callable[[str, str, NotificationLevel], None]):
-        """
-        Checks the cycle count for a specific node and triggers a notification if the cycle count exceeds a threshold.
-        Args:
-            removed_data (dict): A dictionary containing data that has been removed.
-                                    Used to check if the node is present.
-            node (str): The identifier of the node being checked.
-            config (dict): Configuration dictionary containing metadata such as the tag name.
-            handle_listen_events (Callable[[str, str, NotificationLevel], None]):
-                A callback function to handle notification events. It takes three arguments:
-                - A message string describing the event.
-                - A tag string identifying the event.
-                - A NotificationLevel enum indicating the severity of the event.
-        Behavior:
-            - If the node is not in `removed_data`, the cycle count for the node is incremented.
-            - If the cycle count reaches or exceeds 5, the `handle_listen_events` callback is invoked
-                with a warning message, a tag, and a notification level.
-        Notification Example:
-            If the cycle count exceeds the threshold, a warning message is generated in the format:
-            "{cycles} cycles without receive from {node}:{name}"
-            where `cycles` is the current cycle count, `node` is the node identifier, and `name` is the tag name
-            from the `config` dictionary.
+    def check_cycles(self):
         """
+        Checks the cycle counts for all monitored nodes and sends notifications if thresholds are exceeded.
 
-        if node not in removed_data.keys():
+        This method iterates through all monitored nodes and updates their cycle counts based on 
+        configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
+        a warning notification.
+
+        """
+        for node, config in self.nodes.items():
             self.nodes[node]['cycle_rule']['cycle_count'] += self.nodes[node]['cycle_rule']['cycle_increment']
-        if self.nodes[node]['cycle_rule']['cycle_count'] >= 5 and handle_listen_events:
-            name = config['tag_name']
-            cycles = self.nodes[node]['cycle_rule']['cycle_count']
-            handle_listen_events(
-                f'{cycles} cycles without receive from {node}:{name}',
-                f'TAG_{node}:{name}_LISTENNING_STOPPED',
-                NotificationLevel.WARNING
-            )
+            if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
+                name = config['tag_name']
+                cycles = self.nodes[node]['cycle_rule']['cycle_count']
+                self.notification_handler.build_and_send_notification(
+                    notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
+                    message=f'{cycles} cycles without receive from {node}:{name}',
+                    block="opc_manager",
+                    level=NotificationLevel.WARNING
+                )
 
-    def check_opc_listenning(self, handle_listen_events: Callable[[str, str, NotificationLevel], None]) -> None:
+    def check_opc_listenning(self) -> bool:
         """
-        Monitors the OPC connection and triggers events based on the number of cycles
-        without receiving data from the OPC server.
-        Args:
-            handle_listen_events (Callable[[str, str, NotificationLevel], None]):
-                A callback function to handle notification events. It takes three arguments:
-                - A message string describing the event.
-                - An event code string.
-                - A NotificationLevel indicating the severity of the event.
-        Behavior:
-            - Increments the non-receive count each time the method is called.
-            - If the non-receive count reaches 5, triggers a notification event indicating
-              that the OPC server has stopped sending data.
-            - If the non-receive count reaches 15, triggers a notification event indicating
-              a retry to connect to the OPC server, disconnects the current session, and
-              reinitializes the collector with the existing configuration.
+        Checks the OPC connection and triggers notifications if the connection is lost.
+        Returns:
+            bool: True if the connection is lost, False otherwise.
         """
 
         self.non_receive_count += 1
-        if self.non_receive_count >= 5 and handle_listen_events:
-            handle_listen_events(
-                f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
-                f'OPC_LISTENNING_STOPPED__{self.name}', NotificationLevel.ERROR)
-        if self.non_receive_count >= 15 and handle_listen_events:
-            handle_listen_events(
-                f'Retrying to connect to server {self.name}',
-                f'OPC_CONNECTION_RETRY__{self.name}', NotificationLevel.ERROR)
-            self.disconnect()
-            self.init_collector(
-                self.nodes, self.collect_period, self.period)
+        if self.non_receive_count >= 5:
+            self.notification_handler.build_and_send_notification(
+                notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
+                message=f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
+                block="opc_manager",
+                level=NotificationLevel.ERROR
+            )
+        if self.non_receive_count >= 15:
+            self.notification_handler.build_and_send_notification(
+                notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
+                message=f'Retrying to connect to server {self.name}',
+                block="opc_manager",
+                level=NotificationLevel.ERROR
+            )
+            return True
+        return False

tests/unit/managers/test_data_manager.py

@@ -1,7 +1,7 @@
 from unittest.mock import ANY, MagicMock, patch
 from pytest import fixture
 from kafka.errors import NoBrokersAvailable
-
+from sientia_do.notifications.models import NotificationLevel
 from ingestor.managers.data_manager import DataManager
 
 
@@ -10,7 +10,8 @@ from ingestor.managers.data_manager import DataManager
 def data_manager(kafka):
     return DataManager(
         kafka_servers="localhost:9092",
-        logger=MagicMock()
+        logger=MagicMock(),
+        notification_handler=MagicMock()
     )
 
 
@@ -20,7 +21,8 @@ def test___init___success(kafka):
 
     data_manager = DataManager(
         kafka_servers="localhost:9092",
-        logger=logger_mock
+        logger=logger_mock,
+        notification_handler=MagicMock()
     )
 
     kafka.assert_called_once_with(
@@ -46,7 +48,8 @@ def test___init___second_attempt(kafka):
 
     data_manager = DataManager(
         kafka_servers="localhost:9092",
-        logger=logger_mock
+        logger=logger_mock,
+        notification_handler=MagicMock()
     )
 
     kafka.assert_any_call(
@@ -79,7 +82,8 @@ def test___init___failure_max_attempts(kafka):
     try:
         DataManager(
             kafka_servers="localhost:9092",
-            logger=logger_mock
+            logger=logger_mock,
+            notification_handler=MagicMock()
         )
     except NoBrokersAvailable as e:
         assert str(
@@ -172,7 +176,8 @@ def test_publish(data_manager):
     data_manager.kafka_producer.flush.assert_called_once()
 
 
-def test_publish_error(data_manager):
+@patch("ingestor.managers.data_manager.traceback")
+def test_publish_error(traceback, data_manager):
     topic = "test_topic"
     data = {"key": "value"}
 
@@ -189,6 +194,10 @@ def test_publish_error(data_manager):
     )
 
     # Check if the error was logged
-    data_manager.logger.error.assert_called_once_with(
-        "Failed to publish message: Test error"
+    data_manager.notification_handler.build_and_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
     )

tests/unit/managers/test_ingestor_manager.py

@@ -1,6 +1,6 @@
 from unittest.mock import MagicMock, patch
 from pytest import fixture
-
+from sientia_do.notifications.models import NotificationLevel
 from ingestor.managers.ingestor_manager import IngestorManager
 
 
@@ -16,14 +16,16 @@ def ingestor_manager(data_manager_mock, resource_manager_mock):
         heartbeat_ttl=60,
         pod_id="test_pod",
         poll_interval=5,
-        logger=MagicMock()
+        logger=MagicMock(),
+        notification_handler=MagicMock()
     )
 
 
 @patch('ingestor.managers.ingestor_manager.OpcManager')
 @patch('ingestor.managers.ingestor_manager.DataManager')
 @patch('ingestor.managers.ingestor_manager.ResourceManager')
-def test___init__(resource_manager_mock, data_manager_mock, opc_manager_mock):
+@patch('ingestor.managers.ingestor_manager.NotificationHandler')
+def test___init__(notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock):
 
     ingestor = IngestorManager(
         kafka_servers="localhost:9092",
@@ -33,14 +35,15 @@ def test___init__(resource_manager_mock, data_manager_mock, opc_manager_mock):
         heartbeat_ttl=60,
         pod_id="test_pod",
         poll_interval=5,
-        logger=MagicMock()
+        logger=MagicMock(),
+        notification_handler=MagicMock()
     )
 
     opc_manager_mock.assert_not_called()
     data_manager_mock.assert_called_once_with(
         "localhost:9092", ingestor.logger)
     resource_manager_mock.assert_called_once_with(
-        "localhost", 6379, 60, 60, "test_pod")
+        "localhost", 6379, 60, 60, "test_pod", None, None)
 
     assert ingestor.poll_interval == 5
     assert ingestor.managed_tags == {}
@@ -76,7 +79,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager):
 
 
 @patch('ingestor.managers.ingestor_manager.OpcManager')
-def test_initialize_opc_from_config_exception(opc_manager, ingestor_manager):
+@patch('ingestor.managers.ingestor_manager.traceback')
+def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager):
     server_config = {
         'name': 'server1',
         'url': 'opc.tcp://localhost:4840',
@@ -93,8 +97,15 @@ def test_initialize_opc_from_config_exception(opc_manager, ingestor_manager):
         server_config, ingestor_manager.data_manager, ingestor_manager.logger)
 
     assert result is None
-    ingestor_manager.logger.error.assert_called_once_with(
-        "Failed to initialize OpcManager: Initialization error")
+
+    traceback_mock.format_exc.assert_called_once()
+    ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+        notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
+        message='Error initializing OPC manager: Initialization error',
+        block="opc_manager",
+        level=NotificationLevel.ERROR,
+        attachment_content=traceback_mock.format_exc.return_value
+    )
 
 
 @patch('ingestor.managers.ingestor_manager.OpcManager')
@@ -406,7 +417,8 @@ def test_manage_server(ingestor_manager):
         'slot1', 'config1', ingestor_manager.poll_interval)
 
 
-def test_manage_server_subscribe_failure(ingestor_manager):
+@patch('ingestor.managers.ingestor_manager.traceback')
+def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager):
     ingestor_manager.opc_managers = {
         "server1": MagicMock(),
         "server2": MagicMock()
@@ -431,9 +443,17 @@ def test_manage_server_subscribe_failure(ingestor_manager):
         'slot1', 'config1', ingestor_manager.poll_interval)
     ingestor_manager.opc_managers["server1"].unsubscribe.assert_called_once_with(
         'slot1')
-    ingestor_manager.logger.error.assert_any_call(
-        "Failed to subscribe to tags from slot1:server1\n{'tags': 'config1'}: Subscription error"
+
+    traceback_mock.format_exc.assert_called_once()
+
+    ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with(
+        notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1',
+        message='Failed to subscribe to tags from slot1:server1\n{\'tags\': \'config1\'}: Subscription error',
+        block="opc_manager",
+        level=NotificationLevel.ERROR,
+        attachment_content=traceback_mock.format_exc.return_value
     )
+
     ingestor_manager.logger.warning.assert_any_call(
         "Removing subscription from server server1 for slot slot1"
     )
@@ -475,3 +495,129 @@ def test_subscribe_to_tags(ingestor_manager):
 
     ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with(
         'server3', None)
+
+
+def test_check_opc_servers_integrity_all_healthy(ingestor_manager):
+    # Setup mock OPC managers
+    opc_manager1 = MagicMock()
+    opc_manager1.check_cycles.return_value = None
+    opc_manager1.check_opc_listenning.return_value = False
+    opc_manager1.config = {"config": "config1"}
+
+    opc_manager2 = MagicMock()
+    opc_manager2.check_cycles.return_value = None
+    opc_manager2.check_opc_listenning.return_value = False
+    opc_manager2.config = {"config": "config2"}
+
+    ingestor_manager.opc_managers = {
+        "server1": opc_manager1,
+        "server2": opc_manager2
+    }
+
+    # Mock the initialize_opc_from_config method
+    ingestor_manager.initialize_opc_from_config = MagicMock()
+
+    # Call the method
+    ingestor_manager.check_opc_servers_integrity()
+
+    # Verify that check_cycles and check_opc_listenning were called for each server
+    opc_manager1.check_cycles.assert_called_once()
+    opc_manager1.check_opc_listenning.assert_called_once()
+    opc_manager2.check_cycles.assert_called_once()
+    opc_manager2.check_opc_listenning.assert_called_once()
+
+    # Verify that no reinitialization was needed
+    ingestor_manager.initialize_opc_from_config.assert_not_called()
+
+
+def test_check_opc_servers_integrity_server_lost(ingestor_manager):
+    # Setup mock OPC manager that will be lost
+    opc_manager = MagicMock()
+    opc_manager.check_cycles.return_value = None
+    opc_manager.check_opc_listenning.return_value = True  # Server is lost
+    opc_manager.config = {"config": "config1"}
+
+    ingestor_manager.opc_managers = {
+        "server1": opc_manager
+    }
+
+    # Mock the initialize_opc_from_config method to return a new manager
+    new_manager = MagicMock()
+    ingestor_manager.initialize_opc_from_config = MagicMock(
+        return_value=new_manager)
+
+    # Mock update_opc_servers and manage_server
+    ingestor_manager.update_opc_servers = MagicMock()
+    ingestor_manager.manage_server = MagicMock()
+
+    # Call the method
+    ingestor_manager.check_opc_servers_integrity()
+
+    # Verify that the lost server was disconnected
+    opc_manager.disconnect.assert_called_once()
+
+    # Verify that a new manager was initialized
+    ingestor_manager.initialize_opc_from_config.assert_called_once_with(
+        opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger
+    )
+
+    # Verify that the new manager was assigned
+    assert ingestor_manager.opc_managers["server1"] == new_manager
+
+    # Verify that update_opc_servers was called
+    ingestor_manager.update_opc_servers.assert_called_once()
+
+
+def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager):
+    # Setup mock OPC manager that will be lost
+    opc_manager = MagicMock()
+    opc_manager.check_cycles.return_value = None
+    opc_manager.check_opc_listenning.return_value = True  # Server is lost
+    opc_manager.config = {"config": "config1"}
+
+    ingestor_manager.opc_managers = {
+        "server1": opc_manager
+    }
+
+    # Setup managed tags
+    ingestor_manager.managed_tags = {
+        "slot1": {
+            "server1": {
+                "config": "config1",
+                "tags": {"tag1": "value1"}
+            }
+        }
+    }
+
+    # Mock the initialize_opc_from_config method to return a new manager
+    new_manager = MagicMock()
+    ingestor_manager.initialize_opc_from_config = MagicMock(
+        return_value=new_manager)
+
+    # Mock update_opc_servers and manage_server
+    ingestor_manager.update_opc_servers = MagicMock()
+    ingestor_manager.manage_server = MagicMock()
+
+    # Call the method
+    ingestor_manager.check_opc_servers_integrity()
+
+    # Verify that the lost server was disconnected
+    opc_manager.disconnect.assert_called_once()
+
+    # Verify that a new manager was initialized
+    ingestor_manager.initialize_opc_from_config.assert_called_once_with(
+        opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger
+    )
+
+    # Verify that the new manager was assigned
+    assert ingestor_manager.opc_managers["server1"] == new_manager
+
+    # Verify that update_opc_servers was called
+    ingestor_manager.update_opc_servers.assert_called_once()
+
+    # Verify that manage_server was called with the correct tags
+    ingestor_manager.manage_server.assert_called_once_with(
+        "slot1", "server1",
+        {"config": "config1", "tags": {"tag1": "value1"}},
+        {"tag1": "value1"}
+    )

tests/unit/managers/test_opc_manager.py

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch
 from pytest import fixture
 
 from asyncua.crypto.security_policies import SecurityPolicyBasic256
+import pytest
 from ingestor.managers.opc_manager import OpcManager
 from sientia_do.notifications.models import NotificationLevel
 
@@ -36,7 +37,7 @@ tags = {
 def raw_opc_manager():
     return OpcManager(
         'TestConnector', 'opc.tcp://localhost:4840', MagicMock(),
-        MagicMock(), 'opc.tcp://localhost:4840'
+        MagicMock(), 'opc.tcp://localhost:4840', MagicMock()
     )
 
 
@@ -240,82 +241,102 @@ def test_datachange_notification(opc_manager_subscribed):
     assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0
 
 
-def test_check_cycles(opc_manager_subscribed):
-    handler = MagicMock()
-    opc_manager_subscribed.nodes = tags
-    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule'] = {
-        'cycle_increment': 1.0,
-        'cycle_count': 0
+def test_check_cycles_no_notification(opc_manager):
+    # Setup: node with cycle_count just below threshold
+    opc_manager.nodes = {
+        'ns=3;i=1001': {
+            'tag_name': 'Counter',
+            'cycle_rule': {
+                'cycle_increment': 1.0,
+                'cycle_count': 3.0
+            }
+        }
     }
-    opc_manager_subscribed.check_cycles(
-        {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler)
+    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
-    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 1
+    opc_manager.check_cycles()
 
-    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 0
-    opc_manager_subscribed.check_cycles({'ns=3;i=1001': {}},
-                                        'ns=3;i=1001', tags['ns=3;i=1001'], handler)
-
-    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0
-
-    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 4
-    opc_manager_subscribed.check_cycles(
-        {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler)
-
-    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 5
-    handler.assert_called_once_with('5.0 cycles without receive from ns=3;i=1001:Counter',
-                                    'TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', NotificationLevel.WARNING)
+    # After one increment, cycle_count = 4.0, still below threshold
+    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()
 
 
-def test_check_opc_listenning_5_cycles(opc_manager_subscribed):
+def test_check_cycles_triggers_notification(opc_manager):
+    # Setup: node with cycle_count just below threshold, increment will cross threshold
+    opc_manager.nodes = {
+        'ns=3;i=1001': {
+            'tag_name': 'Counter',
+            'cycle_rule': {
+                'cycle_increment': 2.5,
+                'cycle_count': 3.0
+            }
+        }
+    }
+    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
-    handler = MagicMock()
-    opc_manager_subscribed.nodes = tags
-    opc_manager_subscribed.non_receive_count = 4
+    opc_manager.check_cycles()
 
-    opc_manager_subscribed.check_opc_listenning(handler)
-
-    handler.assert_called_once_with(
-        f'5 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}',
-        'OPC_LISTENNING_STOPPED__TestConnector',
-        NotificationLevel.ERROR
+    # After increment, cycle_count = 5.5, should trigger notification
+    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(
+        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
     )
 
 
-def test_check_opc_listenning_no_cycles(opc_manager_subscribed):
-    handler = MagicMock()
-    opc_manager_subscribed.non_receive_count = 0
+def test_check_opc_listenning_no_notification(opc_manager):
+    opc_manager.non_receive_count = 3
+    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
-    opc_manager_subscribed.check_opc_listenning(handler)
+    result = opc_manager.check_opc_listenning()
 
-    handler.assert_not_called()
+    assert opc_manager.non_receive_count == 4
+    opc_manager.notification_handler.build_and_send_notification.assert_not_called()
+    assert result is False
 
 
-def test_check_opc_listenning_no_handler(opc_manager_subscribed):
-    opc_manager_subscribed.non_receive_count = 5
-    opc_manager_subscribed.check_opc_listenning(None)
+def test_check_opc_listenning_warning_notification(opc_manager):
+    opc_manager.non_receive_count = 4
+    opc_manager.notification_handler.build_and_send_notification = MagicMock()
 
-    assert opc_manager_subscribed.non_receive_count == 6
+    result = opc_manager.check_opc_listenning()
 
-
-def test_check_opc_listenning_15_cycles(opc_manager_subscribed):
-    handler = MagicMock()
-    opc_manager_subscribed.init_collector = MagicMock()
-    opc_manager_subscribed.non_receive_count = 14
-    opc_manager_subscribed.collect_period = 1000
-    opc_manager_subscribed.period = 500
-    opc_manager_subscribed.nodes = tags
-
-    opc_manager_subscribed.check_opc_listenning(handler)
-    handler.assert_any_call(
-        f'15 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}',
-        'OPC_LISTENNING_STOPPED__TestConnector',
-        NotificationLevel.ERROR
+    assert opc_manager.non_receive_count == 5
+    opc_manager.notification_handler.build_and_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
     )
-    handler.assert_any_call(
-        'Retrying to connect to server TestConnector',
-        'OPC_CONNECTION_RETRY__TestConnector',
-        NotificationLevel.ERROR
+    assert result is False
+
+
+def test_check_opc_listenning_error_notification_and_retry(opc_manager):
+    opc_manager.non_receive_count = 14
+    opc_manager.notification_handler.build_and_send_notification = MagicMock()
+
+    result = opc_manager.check_opc_listenning()
+
+    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
+    calls = opc_manager.notification_handler.build_and_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
     )
-    opc_manager_subscribed.init_collector.assert_called_once_with(
-        tags, 1000, 500)
+    # Second call: 15 cycles retry
+    assert calls[1].kwargs == dict(
+        notification_id=f'OPC_CONNECTION_RETRY__{opc_manager.name}',
+        message=f'Retrying to connect to server {opc_manager.name}',
+        block="opc_manager",
+        level=NotificationLevel.ERROR
+    )
+    assert result is True

tests/unit/test_ingestor.py

@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 from pytest import fixture
 from ingestor.ingestor import Ingestor
@@ -6,21 +6,27 @@ from ingestor.ingestor import Ingestor
 
 @patch("ingestor.ingestor.getenv")
 @patch("ingestor.ingestor.Ingestor.init_logger")
-def test___init__(init_logger, getenv):
+@patch("ingestor.ingestor.NotificationHandler")
+def test___init__(notification_handler, init_logger, getenv):
     getenv.side_effect = [
         "localhost:9092,localhost:35",  # KAFKA_SERVERS
         "localhost1",       # REDIS_HOST
         '63790',             # REDIS_PORT
+        "user",             # REDIS_USERNAME
+        "password",         # REDIS_PASSWORD
         '100',               # LEASE_TTL
         '200',               # HEARTBEAT_TTL
         "localhost1",      # HOSTNAME
         '50'                 # POLL_INTERVAL
     ]
+
     ingestor = Ingestor()
 
     getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092")
     getenv.assert_any_call("REDIS_HOST", "localhost")
     getenv.assert_any_call("REDIS_PORT", 6379)
+    getenv.assert_any_call("REDIS_USERNAME", None)
+    getenv.assert_any_call("REDIS_PASSWORD", None)
     getenv.assert_any_call("LEASE_TTL", 10)
     getenv.assert_any_call("HEARTBEAT_TTL", 20)
     getenv.assert_any_call("HOSTNAME", "localhost")
@@ -29,18 +35,30 @@ def test___init__(init_logger, getenv):
     assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"]
     assert ingestor.redis_host == "localhost1"
     assert ingestor.redis_port == 63790
+    assert ingestor.redis_username == "user"
+    assert ingestor.redis_password == "password"
     assert ingestor.lease_ttl == 100
     assert ingestor.heartbeat_ttl == 200
     assert ingestor.pod_id == "localhost1"
     assert ingestor.poll_interval == 50
 
     init_logger.assert_called_once()
+    notification_handler.assert_called_once_with(
+        servers=["localhost:9092", "localhost:35"],
+        logger=ingestor.logger,
+        project_name="OPC_INGESTOR",
+        pipeline_name="-",
+        trigger_name="-",
+        model_name="-",
+        model="-"
+    )
 
 
 @fixture
 @patch("ingestor.ingestor.getenv")
 @patch("ingestor.ingestor.Ingestor.init_logger")
-def ingestor(init_logger, getenv):
+@patch("ingestor.ingestor.NotificationHandler")
+def ingestor(notification_handler, init_logger, getenv):
     ing = Ingestor()
     ing.logger = MagicMock()
 
@@ -104,7 +122,10 @@ def test_prepare_ingestor(ingestor_manager_mock, ingestor):
         ingestor.heartbeat_ttl,
         ingestor.pod_id,
         ingestor.poll_interval,
-        ingestor.logger
+        ingestor.logger,
+        ingestor.redis_username,
+        ingestor.redis_password,
+        ingestor.notification_handler
     )
     ingestor_manager.declare_active.assert_called_once()
     ingestor_manager.get_slot_leases.assert_called_once()
@@ -206,6 +227,8 @@ def test_loop(ingestor_manager_started):
         return_value=["ingestor1", "ingestor2"])
     ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock(
         return_value=5)
+    ingestor_manager_started.ingestor_manager.get_number_of_leases = MagicMock(
+        return_value=1)
 
     ingestor_manager_started.loop()
 
@@ -217,7 +240,7 @@ def test_loop(ingestor_manager_started):
         ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value)
     # Explanation: 5 - 2 = 3, 3 - 1 = 2
     ingestor_manager_started.manage_leases.assert_called_once_with(
-        3, 2)
+        4, 3, 2)
     ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once()
 
 
@@ -240,7 +263,7 @@ def test_loop_no_managed(ingestor_manager_started):
         ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value)
     # Explanation: 5 - 2 = 3, 3 - 1 = 2
     ingestor_manager_started.manage_leases.assert_called_once_with(
-        3, -1)
+        ANY, 3, -1)
     ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once()
     ingestor_manager_started.logger.info.assert_any_call(
         "No slots acquired in this loop")