diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..3e99ede --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,7 @@ +{ + "python.testing.pytestArgs": [ + "." + ], + "python.testing.unittestEnabled": false, + "python.testing.pytestEnabled": true +} \ No newline at end of file diff --git a/app.py b/app.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/__init__.py b/ingestor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/managers/__init__.py b/ingestor/managers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py new file mode 100644 index 0000000..742bca2 --- /dev/null +++ b/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() diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py new file mode 100644 index 0000000..953d1e6 --- /dev/null +++ b/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 diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py new file mode 100644 index 0000000..cb4563f --- /dev/null +++ b/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) diff --git a/ingestor/managers/resource_manager.py b/ingestor/managers/resource_manager.py new file mode 100644 index 0000000..2e45ed5 --- /dev/null +++ b/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 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ff96ac6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +asyncua==1.1.5 +redis +git+https://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@github.com/Aignosi/sientia-dataops-library.git \ No newline at end of file diff --git a/tests/managers/test_data_manager.py b/tests/managers/test_data_manager.py new file mode 100644 index 0000000..b2f11a8 --- /dev/null +++ b/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() diff --git a/tests/managers/test_opc_manager.py b/tests/managers/test_opc_manager.py new file mode 100644 index 0000000..855a281 --- /dev/null +++ b/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) diff --git a/tests/managers/test_resource_manager.py b/tests/managers/test_resource_manager.py new file mode 100644 index 0000000..11c6a43 --- /dev/null +++ b/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()