diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index cb9e97e..fe0666d 100644 --- a/ingestor/ingestor.py +++ b/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() diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 159b4a4..3f9ca3c 100644 --- a/ingestor/managers/data_manager.py +++ b/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) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index edd62ea..b86eb99 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/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}" diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index f376615..e7005bd 100644 --- a/ingestor/managers/opc_manager.py +++ b/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 diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 24df1b1..34c387a 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/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 ) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 991b375..003cf6d 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/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"} + ) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 005d038..8886709 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/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 diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 7e539a3..5a5b603 100644 --- a/tests/unit/test_ingestor.py +++ b/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")