SIENTIAPDE-988

Implement initial structure for data ingestion with Kafka and OPC UA integration

- Added DataManager for Kafka message handling
- Introduced IngestorManager to manage data ingestion processes
- Created OpcManager for OPC UA client interactions
- Developed ResourceManager for Redis-based resource management
- Established testing framework with unit tests for DataManager and OpcManager
- Configured project settings for Python testing with pytest

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "."
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

ingestor/managers/data_manager.py

@@ -0,0 +1,51 @@
+import json
+from logging import Logger
+from kafka import KafkaProducer
+
+
+class DataManager():
+    def __init__(self, kafka_servers: str, logger: Logger) -> None:
+        self.kafka_producer = KafkaProducer(
+            bootstrap_servers=kafka_servers,
+            value_serializer=lambda v: json.dumps(v).encode(
+                'utf-8'),  # Serialize JSON messages
+            key_serializer=lambda k: str(k).encode('utf-8') if k else None,
+        )
+        self.logger = logger
+
+    def __del__(self):
+        """Destructor to close the producer connection."""
+        self.logger.info("Closing Kafka producer...")
+        self.kafka_producer.flush()
+        self.kafka_producer.close()
+
+    def delivery_report(self, msg: str):
+        """Callback for delivery reports from Kafka."""
+        self.logger.info(
+            f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}")
+
+    def delivery_error(self, err: str):
+        """Callback for delivery reports from Kafka."""
+        self.logger.error(f"Delivery failed for record : {err}")
+
+    def publish(self, topic: str, data: dict) -> None:
+        """
+        Publishes a message to a specified Kafka topic.
+
+        Args:
+            topic (str): The name of the Kafka topic to which the message will be published.
+            data (dict): The message data to be sent to the Kafka topic.
+
+        Returns:
+            None
+
+        Raises:
+            Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback.
+        """
+
+        self.kafka_producer.send(
+            topic=topic, value=data).add_callback(
+                self.delivery_report).add_errback(
+                    self.delivery_error)
+
+        self.kafka_producer.flush()

ingestor/managers/ingestor_manager.py

@@ -0,0 +1,36 @@
+from logging import Logger
+from typing import Dict, List
+from ingestor.managers.data_manager import DataManager
+from ingestor.managers.opc_manager import OpcManager
+from ingestor.managers.resource_manager import ResourceManager
+
+
+class IngestorManager():
+    def __init__(self,
+                 kafka_servers: str, redis_host: str, redis_port: int,
+                 lease_ttl: int, heartbeat_ttl: int, poll_interval: int,
+                 logger: Logger):
+
+        self.data_manager = DataManager(kafka_servers, logger)
+        self.opc_managers = []
+        self.resource_manager = ResourceManager(
+            redis_host, redis_port, lease_ttl, heartbeat_ttl, logger
+        )
+
+    def init_ingestor(self):
+        pass
+
+    def declare_active(self):
+        pass
+
+    def get_active_ingestors(self):
+        pass
+
+    def get_slot_leases(self, max_slots: int = 1) -> List[Dict]:
+        pass
+
+    def drop_slot_leases(self, ids: List[str]) -> None:
+        pass
+
+    def subscribe_to_tags(self, tags: List[Dict]) -> None:
+        pass

ingestor/managers/opc_manager.py

@@ -0,0 +1,291 @@
+import json
+from logging import Logger
+from pathlib import Path
+from typing import Callable
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from asyncua.sync import Client
+from sientia_do.notifications.models import NotificationLevel
+
+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):
+        self.url = url
+        self.name = name
+        self.server_uri = server_uri
+        self.data_queue = {}
+        self.logger = logger
+        self.non_receive_count = 0
+        self.client = None
+        self.cert_path = cert_path
+        self.private_key_path = private_key_path
+        self.server_cert_path = server_cert_path
+        self.nodes = {}
+        self.subscription = None
+        self.data_manager = data_manager
+
+    def set_security(self):
+        """
+        Configures the security settings for the OPC UA client.
+        This method sets up the security policy, certificates, and timeouts
+        required for establishing a secure connection with the OPC UA server.
+        Raises:
+            ValueError: If either the certificate path or private key path is not provided.
+        Attributes:
+            cert_path (str): Path to the client's certificate file.
+            private_key_path (str): Path to the client's private key file.
+            server_cert_path (str, optional): Path to the server's certificate file.
+            server_uri (str): The URI of the server to be used as the application URI.
+            client (opcua.Client): The OPC UA client instance.
+            logger (logging.Logger): Logger instance for logging information.
+        Security Settings:
+            - Security Policy: Basic256
+            - Secure Channel Timeout: 10,000,000 ms
+            - Session Timeout: 10,000,000 ms
+        """
+
+        if not all([self.cert_path, self.private_key_path]):
+            raise ValueError(
+                "Certificate and private key paths must be provided for secure connection.")
+        cert = Path(self.cert_path)
+        private_key = Path(self.private_key_path)
+        server_cert = Path(
+            self.server_cert_path) if self.server_cert_path else None
+
+        self.client.application_uri = self.server_uri
+        self.logger.info('Setting security...')
+        self.client.set_security(
+            SecurityPolicyBasic256,
+            certificate=str(cert),
+            private_key=str(private_key),
+            server_certificate=str(server_cert)
+        )
+        self.client.secure_channel_timeout = 10000000
+        self.client.session_timeout = 10000000
+
+    def connect(self):
+        """
+        Establishes a connection to the OPC server.
+        This method initializes the OPC client using the provided URL and
+        sets up security if a certificate path is specified. It then
+        attempts to connect to the server and logs the connection status.
+        Raises:
+            Exception: If the connection to the OPC server fails.
+        """
+
+        self.client = Client(self.url)
+        if self.cert_path:
+            self.set_security()
+        self.logger.info('Starting connection...')
+        self.client.connect()
+
+    def create_subscription(self, period: int = 500):
+        """
+        Creates a subscription with the specified monitoring period.
+        This method establishes a subscription to monitor data changes or events
+        from the OPC UA server. If the client is not connected, an exception is raised.
+        Args:
+            period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
+        Raises:
+            ValueError: If the client is not connected.
+        Side Effects:
+            - Sets the `self.period` attribute to the specified or default period.
+            - Creates a subscription and assigns it to `self.subscription`.
+            - Logs the creation of the subscription.
+        """
+
+        if not self.client:
+            raise ValueError("Client not connected. Call connect first.")
+
+        p = period if period != None else 500
+        self.period = p
+        self.subscription = self.client.create_subscription(
+            p, self)
+        self.logger.info('Subscription created.')
+
+    def subscribe(self, nodes: dict, collect_period: int):
+        """
+        Subscribes to a set of OPC UA nodes for data change notifications.
+        This method adds the specified nodes to the subscription and configures
+        their data collection rules based on the provided collection period and
+        node-specific frequency.
+        Args:
+            nodes (dict): A dictionary where keys are node identifiers (e.g., node
+                IDs or paths) and values are configurations for each node. Each
+                configuration must include a 'frequency' key indicating the
+                frequency of data collection in Hz.
+            collect_period (int): The data collection period in seconds.
+        Raises:
+            ValueError: If the subscription has not been created by calling
+                `create_subscription` prior to this method.
+        """
+
+        if not self.subscription:
+            raise ValueError(
+                "Subscription not created. Call create_subscription first.")
+
+        self.addr_nodes = [self.client.get_node(
+            n) for n in nodes if n not in self.nodes]
+        self.nodes.update(nodes)
+        self.collect_period = collect_period
+
+        for node, config in self.nodes.items():
+            self.nodes[node]['cycle_rule'] = {
+                'cycle_increment': collect_period*1000/config['frequency'],
+                'cycle_count': 0
+            }
+
+        self.subscription.subscribe_data_change(self.addr_nodes)
+
+    def __del__(self):
+        self.disconnect()
+
+    def disconnect(self):
+        """
+        Disconnects from the OPC UA server.
+        This method handles the disconnection process by deleting the subscription
+        and disconnecting the client from the OPC UA server. It logs the disconnection
+        process and handles any exceptions that may occur during cleanup.
+        Raises:
+            Exception: If an error occurs while deleting the subscription or disconnecting
+                       from the OPC UA server, it logs the error details.
+        """
+
+        self.logger.warning('Disconnecting from OPC server')
+        try:
+            self.subscription.delete()
+            self.client.disconnect()
+            self.logger.warning("Disconnected from OPC UA server.")
+        except Exception as sub_error:
+            self.logger.error(f"Failed to clean up subscription: {sub_error}")
+
+    def init_collector(self, nodes: dict, collect_period: int, period: int):
+        """
+        Initializes the OPC data collector by connecting to the server, creating a subscription,
+        and subscribing to the specified nodes.
+        Args:
+            nodes (dict): A dictionary where keys are node identifiers and values are the node details
+                          to be subscribed to.
+            collect_period (int): The interval (in milliseconds) at which data should be collected
+                                  from the subscribed nodes.
+            period (int): The publishing interval (in milliseconds) for the subscription.
+        Returns:
+            None
+        """
+
+        self.connect()
+        self.create_subscription(period)
+        self.subscribe(nodes, collect_period)
+
+        self.logger.info(
+            f"Connected to OPC server {self.name} at {self.url}.")
+
+    def datachange_notification(self, node, _val, data):
+        """
+        Handles data change notifications for monitored OPC UA nodes.
+        This method is triggered when a monitored node's value changes. It processes
+        the notification, updates internal state, and publishes the data to the
+        appropriate topics.
+        Args:
+            node (NodeId): The OPC UA node that triggered the data change notification.
+            _val (Any): The new value of the node (unused in this implementation).
+            data (DataChangeNotification): The data change notification object containing
+                details about the change.
+        Behavior:
+            - Extracts the value and source timestamp from the monitored item.
+            - Resets the cycle count for the node's cycle rule.
+            - Resets the non-receive count.
+            - Constructs a data dictionary containing the tag, tag name, timestamp, and value.
+            - Publishes the data to all topics associated with the node.
+        """
+
+        # get data value
+        monitored_item = data.monitored_item
+        value = monitored_item.Value.Value.Value
+        # source_timestamp
+        source_timestamp = monitored_item.Value.SourceTimestamp
+        tag = str(node)
+
+        self.nodes[tag]['cycle_rule']['cycle_count'] = 0
+        self.non_receive_count = 0
+
+        data = {
+            'tag': tag,
+            'name': self.nodes[str(node)]['tag_name'],
+            'timestamp': source_timestamp,
+            'value': value
+        }
+
+        [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.
+        """
+
+        if node not in removed_data.keys():
+            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
+            )
+
+    def check_opc_listenning(self, handle_listen_events: Callable[[str, str, NotificationLevel], None]) -> None:
+        """
+        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.
+        """
+
+        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)

ingestor/managers/resource_manager.py

@@ -0,0 +1,84 @@
+import json
+from redis import Redis
+
+
+class ResourceManager:
+    def __init__(self, host: str, port: int,
+                 lease_ttl: int, heartbeat_ttl: int, pod_id: str) -> None:
+        self.redis = Redis(host=host, port=port, decode_responses=True)
+        self.lease_ttl = lease_ttl
+        self.heartbeat_ttl = heartbeat_ttl
+        self.pod_id = pod_id
+
+    def get(self, key: str) -> dict:
+        """
+        Retrieve a value from Redis by its key and return it as a dictionary.
+        Args:
+            key (str): The key to look up in Redis.
+        Returns:
+            dict: The value associated with the key, parsed as a dictionary, 
+                  or None if the key does not exist or the value is empty.
+        """
+
+        history = self.redis.get(key)
+        return json.loads(history) if history else None
+
+    def get_tag_slot(self, id: str) -> dict:
+        """
+        Retrieve the tag slot information for a given ID.
+        Args:
+            id (str): The unique identifier of the tag slot to retrieve.
+        Returns:
+            dict: A dictionary containing the tag slot information associated with the given ID.
+        """
+
+        return self.get(f"slot:opc_tags:{id}")
+
+    def ingestor_heartbeat(self) -> None:
+        """
+        Sends a heartbeat signal to Redis to indicate that the ingestor is active.
+        This method sets a key in Redis with a specific format that includes the
+        ingestor's pod ID. The key is set with a value of 1 and an expiration
+        time defined by `self.heartbeat_ttl`. This allows monitoring systems to
+        track the activity and health of the ingestor.
+        Returns:
+            None
+        """
+
+        self.redis.set(
+            f"heartbeat:ingestor:{self.pod_id}", 1, ex=self.heartbeat_ttl)
+
+    def lease_tag(self, tag_id: str) -> None:
+        """
+        Acquires a lease for a specific OPC tag by setting a key in Redis with a 
+        time-to-live (TTL). This ensures that the tag is associated with the 
+        current pod for a limited duration.
+        Args:
+            tag_id (str): The unique identifier of the OPC tag to lease.
+        Raises:
+            redis.exceptions.RedisError: If there is an issue communicating with 
+                the Redis server.
+        """
+
+        self.redis.set(
+            f"lease:opc_tags:{tag_id}", self.pod_id, nx=True, ex=self.lease_ttl)
+
+    def renew_tag_lease(self, tag_id: str) -> bool:
+        """
+        Renews the lease for a specific OPC tag if the current pod holds the lease.
+        This method checks if the current pod (identified by `self.pod_id`) holds the lease
+        for the given OPC tag. If so, it extends the lease by resetting its expiration time
+        in Redis to the configured lease TTL (`self.lease_ttl`).
+        Args:
+            tag_id (str): The identifier of the OPC tag whose lease is to be renewed.
+        Returns:
+            bool: True if the lease was successfully renewed, False otherwise.
+        """
+
+        current = self.redis.get(
+            f"lease:opc_tags:{tag_id}")
+        if current == self.pod_id:
+            self.redis.expire(f"lease:opc_tags:{tag_id}", self.lease_ttl)
+            return True
+
+        return False

requirements.txt

@@ -0,0 +1,3 @@
+asyncua==1.1.5
+redis
+git+https://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@github.com/Aignosi/sientia-dataops-library.git

tests/managers/test_data_manager.py

@@ -0,0 +1,68 @@
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+
+from ingestor.managers.data_manager import DataManager
+
+
+@fixture
+@patch("ingestor.managers.data_manager.KafkaProducer")
+def data_manager(kafka):
+    return DataManager(
+        kafka_servers="localhost:9092",
+        logger=MagicMock()
+    )
+
+
+def test___del__(data_manager):
+    flush_mock = MagicMock()
+    close_mock = MagicMock()
+
+    data_manager.kafka_producer.flush = flush_mock
+    data_manager.kafka_producer.close = close_mock
+
+    data_manager.__del__()
+    flush_mock.assert_called_once()
+    close_mock.assert_called_once()
+
+
+def test_delivery_report(data_manager):
+    msg = MagicMock()
+    msg.topic = "test_topic"
+    msg.partition = 0
+    msg.offset = 1
+
+    data_manager.delivery_report(msg)
+
+    data_manager.logger.info.assert_called_once_with(
+        f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}"
+    )
+
+
+def test_delivery_error(data_manager):
+    err = "Test error"
+    data_manager.delivery_error(err)
+
+    data_manager.logger.error.assert_called_once_with(
+        f"Delivery failed for record : {err}"
+    )
+
+
+def test_publish(data_manager):
+    topic = "test_topic"
+    data = {"key": "value"}
+
+    # Mock the send method of the Kafka producer
+    send_mock = MagicMock()
+    data_manager.kafka_producer.send = send_mock
+
+    # Call the publish method
+    data_manager.publish(topic, data)
+
+    # Check if the send method was called with the correct arguments
+    send_mock.assert_called_once_with(
+        topic=topic, value=data
+    )
+
+    send_mock.return_value.add_callback.assert_called_once()
+
+    data_manager.kafka_producer.flush.assert_called_once()

tests/managers/test_opc_manager.py

@@ -0,0 +1,316 @@
+import json
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from ingestor.managers.opc_manager import OpcManager
+from sientia_do.notifications.models import NotificationLevel
+
+tags = {
+    'ns=3;i=1001': {
+        'aggregation_function': 'LTS',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Counter'
+    },
+    'ns=3;i=1003': {
+        'aggregation_function': 'AVG',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Random'
+    },
+    'ns=3;i=1004': {
+        'aggregation_function': 'MDN',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Sawtooth'
+    }
+}
+
+
+@fixture
+def raw_opc_manager():
+    return OpcManager(
+        'TestConnector', 'opc.tcp://localhost:4840', MagicMock(),
+        MagicMock(), 'opc.tcp://localhost:4840'
+    )
+
+
+@fixture
+def opc_manager(raw_opc_manager):
+    raw_opc_manager.client = MagicMock()
+    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'
+
+    return raw_opc_manager
+
+
+@fixture
+def opc_manager_subscribed(opc_manager):
+    opc_manager.subscription = MagicMock()
+
+    return opc_manager
+
+
+def test_set_security_success(opc_manager):
+    opc_manager.set_security()
+
+    assert opc_manager.client.application_uri == opc_manager.server_uri
+
+    opc_manager.client.set_security.assert_called_once_with(
+        SecurityPolicyBasic256,
+        certificate=opc_manager.cert_path,
+        private_key=opc_manager.private_key_path,
+        server_certificate=opc_manager.server_cert_path
+    )
+
+    assert opc_manager.client.secure_channel_timeout == 10000000
+    assert opc_manager.client.session_timeout == 10000000
+
+
+def test_set_security_no_cert(opc_manager):
+    opc_manager.cert_path = None
+    opc_manager.private_key_path = None
+
+    try:
+        opc_manager.set_security()
+    except ValueError as e:
+        assert str(
+            e) == "Certificate and private key paths must be provided for secure connection."
+    else:
+        assert False, "ValueError not raised"
+
+    assert opc_manager.client.set_security.call_count == 0
+
+
+@patch('ingestor.managers.opc_manager.Client')
+def test_connect_no_security(client, raw_opc_manager):
+    raw_opc_manager.set_security = MagicMock()
+
+    raw_opc_manager.connect()
+
+    client.assert_called_once_with(raw_opc_manager.url)
+    raw_opc_manager.client.connect.assert_called_once()
+    raw_opc_manager.set_security.assert_not_called()
+
+
+@patch('ingestor.managers.opc_manager.Client')
+def test_connect_with_security(client, 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.set_security = MagicMock()
+
+    raw_opc_manager.connect()
+
+    client.assert_called_once_with(raw_opc_manager.url)
+    raw_opc_manager.client.connect.assert_called_once()
+    raw_opc_manager.set_security.assert_called_once()
+
+
+def test_create_subscription_no_client(raw_opc_manager):
+    try:
+        raw_opc_manager.create_subscription()
+    except ValueError as e:
+        assert str(e) == "Client not connected. Call connect first."
+    else:
+        assert False, "ValueError not raised"
+
+
+def test_create_subscription_success_has_period(opc_manager):
+    opc_manager.create_subscription(1000)
+
+    assert opc_manager.period == 1000
+    opc_manager.client.create_subscription.assert_called_once_with(
+        1000, opc_manager)
+    assert opc_manager.subscription is not None
+
+
+def test_create_subscription_success_no_period(opc_manager):
+    opc_manager.create_subscription(None)
+
+    assert opc_manager.period == 500
+    opc_manager.client.create_subscription.assert_called_once_with(
+        500, opc_manager)
+    assert opc_manager.subscription is not None
+
+
+def test_subscribe_no_subscription(opc_manager):
+    try:
+        opc_manager.subscribe(tags, 1000)
+    except ValueError as e:
+        assert str(
+            e) == "Subscription not created. Call create_subscription first."
+    else:
+        assert False, "ValueError not raised"
+
+
+def test_subscribe_success(opc_manager_subscribed):
+    opc_manager_subscribed.nodes = {
+        'ns=3;i=1001': 'data'
+    }
+    opc_manager_subscribed.subscribe(tags, 1000)
+
+    assert opc_manager_subscribed.nodes == tags
+    assert opc_manager_subscribed.addr_nodes == [
+        opc_manager_subscribed.client.get_node(n) for n in tags if n != 'ns=3;i=1001']
+
+
+def test_disconnect_success(opc_manager_subscribed):
+    opc_manager_subscribed.client = MagicMock()
+    opc_manager_subscribed.subscription = MagicMock()
+
+    opc_manager_subscribed.disconnect()
+
+    opc_manager_subscribed.subscription.delete.assert_called_once()
+    opc_manager_subscribed.client.disconnect.assert_called_once()
+
+
+def test_disconnect_error(opc_manager_subscribed):
+    opc_manager_subscribed.client = MagicMock()
+    opc_manager_subscribed.subscription = MagicMock(
+        delete=MagicMock(side_effect=Exception("Test error"))
+    )
+
+    opc_manager_subscribed.disconnect()
+
+    opc_manager_subscribed.subscription.delete.assert_called_once()
+    opc_manager_subscribed.client.disconnect.assert_not_called()
+    opc_manager_subscribed.logger.error.assert_called_once_with(
+        "Failed to clean up subscription: Test error")
+
+
+def test_init_collector(opc_manager_subscribed):
+    opc_manager_subscribed.connect = MagicMock()
+    opc_manager_subscribed.create_subscription = MagicMock()
+    opc_manager_subscribed.subscribe = MagicMock()
+
+    opc_manager_subscribed.init_collector(tags, 1000, 500)
+
+    opc_manager_subscribed.connect.assert_called_once()
+    opc_manager_subscribed.create_subscription.assert_called_once_with(500)
+    opc_manager_subscribed.subscribe.assert_called_once_with(tags, 1000)
+
+
+def test_datachange_notification(opc_manager_subscribed):
+    data = MagicMock(
+        monitored_item=MagicMock(
+            Value=MagicMock(
+                Value=MagicMock(Value=42),
+                SourceTimestamp='2021-01-01T00:00:00'
+
+            )))
+    opc_manager_subscribed.nodes = {
+        'ns=3;i=1001': {
+            'tag_name': 'Counter',
+            'cycle_rule': {
+                'cycle_increment': 1.0,
+                'cycle_count': 2
+            },
+            'topics': ['topic1', 'topic2']
+        }
+    }
+
+    opc_manager_subscribed.datachange_notification(
+        'ns=3;i=1001', None, data)
+
+    opc_manager_subscribed.data_manager.publish.assert_any_call(
+        'topic1', {
+            'tag': 'ns=3;i=1001',
+            'name': 'Counter',
+            'timestamp': '2021-01-01T00:00:00',
+            'value': 42
+        })
+    opc_manager_subscribed.data_manager.publish.assert_any_call(
+        'topic2', {
+            'tag': 'ns=3;i=1001',
+            'name': 'Counter',
+            'timestamp': '2021-01-01T00:00:00',
+            'value': 42
+        })
+    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
+    }
+    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'] == 1
+
+    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)
+
+
+def test_check_opc_listenning_5_cycles(opc_manager_subscribed):
+
+    handler = MagicMock()
+    opc_manager_subscribed.nodes = tags
+    opc_manager_subscribed.non_receive_count = 4
+
+    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
+    )
+
+
+def test_check_opc_listenning_no_cycles(opc_manager_subscribed):
+    handler = MagicMock()
+    opc_manager_subscribed.non_receive_count = 0
+
+    opc_manager_subscribed.check_opc_listenning(handler)
+
+    handler.assert_not_called()
+
+
+def test_check_opc_listenning_no_handler(opc_manager_subscribed):
+    opc_manager_subscribed.non_receive_count = 5
+    opc_manager_subscribed.check_opc_listenning(None)
+
+    assert opc_manager_subscribed.non_receive_count == 6
+
+
+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
+    )
+    handler.assert_any_call(
+        'Retrying to connect to server TestConnector',
+        'OPC_CONNECTION_RETRY__TestConnector',
+        NotificationLevel.ERROR
+    )
+    opc_manager_subscribed.init_collector.assert_called_once_with(
+        tags, 1000, 500)

tests/managers/test_resource_manager.py

@@ -0,0 +1,73 @@
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+from ingestor.managers.resource_manager import ResourceManager
+
+
+@fixture
+@patch('ingestor.managers.resource_manager.Redis')
+def resource_manager(redis):
+
+    return ResourceManager(
+        'localhost', 6379, 10, 10, 'pod_id'
+    )
+
+
+def test_get_success(resource_manager):
+    resource_manager.redis.get.return_value = '{"key": "value"}'
+    result = resource_manager.get('key')
+    assert result == {"key": "value"}
+    resource_manager.redis.get.assert_called_once_with('key')
+
+
+def test_get_failure(resource_manager):
+    resource_manager.redis.get.return_value = None
+    result = resource_manager.get('key')
+    assert result is None
+    resource_manager.redis.get.assert_called_once_with('key')
+
+
+def test_get_tag_slot(resource_manager):
+    resource_manager.get = MagicMock(return_value={"tag": "slot"})
+    result = resource_manager.get_tag_slot('id')
+    assert result == {"tag": "slot"}
+    resource_manager.get.assert_called_once_with('slot:opc_tags:id')
+
+
+def test_ingestor_heartbeat(resource_manager):
+    resource_manager.redis.set.return_value = True
+    resource_manager.ingestor_heartbeat()
+    resource_manager.redis.set.assert_called_once_with(
+        'heartbeat:ingestor:pod_id', 1, ex=10
+    )
+
+
+def test_lease_tag(resource_manager):
+    resource_manager.redis.set.return_value = True
+    resource_manager.lease_tag('tag_id')
+    resource_manager.redis.set.assert_called_once_with(
+        'lease:opc_tags:tag_id', 'pod_id', nx=True, ex=10
+    )
+
+
+def test_renew_tag_lease_success(resource_manager):
+    resource_manager.redis.get.return_value = 'pod_id'
+    resource_manager.redis.expire.return_value = True
+    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
+    )
+
+
+def test_renew_tag_lease_failure(resource_manager):
+    resource_manager.redis.get.return_value = 'other_pod_id'
+    resource_manager.redis.expire.return_value = False
+    result = resource_manager.renew_tag_lease('tag_id')
+    assert result is False
+    resource_manager.redis.get.assert_called_once_with(
+        'lease:opc_tags:tag_id'
+    )
+    resource_manager.redis.expire.assert_not_called()