From 1bd532b7d893567dabb0ce2c6a2b04d27f28bb73 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 26 Aug 2025 16:49:19 -0300 Subject: [PATCH 1/5] Update environment configuration and refactor OPC activities for async handling - Expanded the .env file with configurations for MongoDB, Postgres, MlFlow, and Temporal. - Refactored OPC class methods to be asynchronous, including init_opc, write_data, manage_output_tags, and shutdown. - Updated the worker to initialize OPC asynchronously and adjusted shutdown handling for activities. --- .gitignore | 4 +- laborious/activities/activities.py | 4 +- laborious/activities/opc.py | 25 +++++---- laborious/utils/repository/opc_repository.py | 54 ++++++++++---------- laborious/worker/worker.py | 7 ++- run_local.sh | 18 +++++++ 6 files changed, 70 insertions(+), 42 deletions(-) create mode 100755 run_local.sh diff --git a/.gitignore b/.gitignore index 9035481..b14c9ac 100644 --- a/.gitignore +++ b/.gitignore @@ -41,4 +41,6 @@ htmlcov/ # git keys git_key* -git_log \ No newline at end of file +git_log + +.env \ No newline at end of file diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index 3970a28..b6d7bfc 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -44,6 +44,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): logger=logger, notification_handler=notification_handler) - def shutdown(self): + async def shutdown(self): Postgres.close(self) - OPC.shutdown(self) + await OPC.shutdown(self) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 5eaecf2..56610d1 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -26,7 +26,10 @@ class OPC(BaseActivity): self, logger, notification_handler, set_error_counter=True) self.opc_repository: dict[str, OpcRepository] = {} - for id, server in opc_servers.items(): + self.opc_servers = opc_servers + + async def init_opc(self): + for id, server in self.opc_servers.items(): self.opc_repository[id] = OpcRepository( id=server['id'], url=server['url'], @@ -39,7 +42,7 @@ class OPC(BaseActivity): reconnection_interval=server['reconnection_interval'], pod_id=self.pod_id ) - is_connected, error_data = self.opc_repository[id].connect() + is_connected, error_data = await self.opc_repository[id].connect() if not is_connected: self.send_notification( metadata={ @@ -56,8 +59,8 @@ class OPC(BaseActivity): 'attachment_content', None) ) - def write_data(self, server_id: str, tag: str, data: Any, - data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: + async def write_data(self, server_id: str, tag: str, data: Any, + data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: """ Write data to OPC server. @@ -73,7 +76,7 @@ class OPC(BaseActivity): """ try: - is_success, error_data = self.opc_repository[server_id].write_data( + is_success, error_data = await self.opc_repository[server_id].write_data( tag, data, data_type, self.logger, metadata) if not is_success: self.send_notification( @@ -113,14 +116,14 @@ class OPC(BaseActivity): return False return True - def manage_output_tags( + async def manage_output_tags( self, server_id: str, config: dict[str, Any], data: DataFrame, metadata: dict[str, Any], success: bool) -> tuple[bool, int]: count = 0 if 'prediction_tags' in config: for tag, tag_config in config['prediction_tags'].items(): - local_success = self.write_data( + local_success = await self.write_data( server_id=server_id, tag=tag, data=data.head(1)['prediction'].values[0], @@ -136,7 +139,7 @@ class OPC(BaseActivity): if 'confidence_tags' in config: for tag, tag_config in config['confidence_tags'].items(): - local_success = self.write_data( + local_success = await self.write_data( server_id=server_id, tag=tag, data=data.head(1)['prediction_confidence'].values[0], @@ -185,7 +188,7 @@ class OPC(BaseActivity): success = False continue - local_success, local_count = self.manage_output_tags( + local_success, local_count = await self.manage_output_tags( server_id, config, data, metadata, success) success = success and local_success @@ -221,6 +224,6 @@ class OPC(BaseActivity): return data.to_dict() - def shutdown(self): + async def shutdown(self): for opc in self.opc_repository.values(): - opc.disconnect() + await opc.disconnect() diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 4f97ef9..6742194 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,9 +1,10 @@ +import asyncio import traceback import time from datetime import datetime from pathlib import Path from typing import Any -from asyncua.sync import Client +from asyncua import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.ua import DataValue, Variant, VariantType, DateTime from regex import F @@ -62,7 +63,7 @@ class OpcRepository(): 'schedule_name': '-' } - def set_security(self): + async def set_security(self): """ Configures the security settings for the OPC UA client. This method sets up the security policy, certificates, and timeouts @@ -92,7 +93,7 @@ class OpcRepository(): self.client.application_uri = self.server_uri self.logger.custom_info('Setting security...', self.metadata) - self.client.set_security( + await self.client.set_security( SecurityPolicyBasic256, certificate=str(cert), private_key=str(private_key), @@ -101,7 +102,7 @@ class OpcRepository(): self.client.secure_channel_timeout = 10000000 self.client.session_timeout = 10000000 - def connect(self) -> tuple[bool, dict[str, Any]]: + async def connect(self) -> tuple[bool, dict[str, Any]]: """ Establishes a connection to the OPC server. This method initializes the OPC client using the provided URL and @@ -113,12 +114,12 @@ class OpcRepository(): self.client = Client(self.url) if self.cert_path: - self.set_security() + await self.set_security() self.logger.custom_info( f'Starting connection to OPC server {self.id}...', self.metadata) - return self.try_connect() + return await self.try_connect() - def try_connect(self) -> tuple[bool, dict[str, Any]]: + async def try_connect(self) -> tuple[bool, dict[str, Any]]: """ Tries to connect to the OPC server. @@ -128,7 +129,7 @@ class OpcRepository(): try: self.last_reconnection_time = datetime.now() - self.client.connect() + await self.client.connect() return True, {} except Exception as e: trace = traceback.format_exc() @@ -142,14 +143,14 @@ class OpcRepository(): "attachment_content": trace } - def disconnect(self): + async def disconnect(self): """ Disconnects from the OPC server. """ if self.client is None: return try: - self.client.disconnect() + await self.client.disconnect() self.logger.custom_info( 'Disconnected from OPC server', self.metadata) except Exception as e: @@ -162,12 +163,12 @@ class OpcRepository(): Disconnects from the OPC server when the object is destroyed. """ try: - self.disconnect() + asyncio.run(self.disconnect()) except Exception as e: self.logger.custom_error( f"Error in destructor: {e}", self.metadata) - def validate_connection(self) -> tuple[bool, dict[str, Any]]: + async def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ Validates the connection to the OPC server. If the connection is not established, it attempts to reconnect. @@ -179,13 +180,13 @@ class OpcRepository(): If the client is connected, it returns True. """ if self.client is None: - return self.connect() + return await self.connect() if self.error_count > 5: self.logger.custom_warning( f"OPC server {self.id} will be disconnected due to multiple errors", self.metadata) try: - self.disconnect() + await self.disconnect() except Exception as e: trace = traceback.format_exc() self.logger.custom_error( @@ -193,21 +194,22 @@ class OpcRepository(): self.logger.custom_error(trace, self.metadata) self.logger.custom_info( f"Attempting to reconnect to OPC server {self.id}...", self.metadata) - return self.connect() - - if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \ - (hasattr(self.client.aio_obj.uaclient, 'protocol') and - self.client.aio_obj.uaclient.protocol.state == "closed"): + return await self.connect() + # Check if client is connected using asyncua's connection state + try: + # Try to get a simple node to test connection + await self.client.get_node("ns=0;i=2253") # Server node + except Exception: self.logger.custom_error( f"OPC server {self.id} is not connected", self.metadata) if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( ) > self.reconnection_interval: - self.disconnect() + await self.disconnect() self.logger.custom_info( f"Trying to reconnect to OPC server {self.id}...", self.metadata) - return self.connect() + return await self.connect() return False, { "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}", @@ -218,8 +220,8 @@ class OpcRepository(): return True, {} - def write_data(self, node: str, value: Any, data_type: str, - logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]: + async def write_data(self, node: str, value: Any, data_type: str, + logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """ Writes data to the OPC server. If the connection is not established, it attempts to reconnect. @@ -231,7 +233,7 @@ class OpcRepository(): If the client is connected, it returns True. """ - is_connected, error = self.validate_connection() + is_connected, error = await self.validate_connection() if not is_connected: return False, error @@ -239,7 +241,7 @@ class OpcRepository(): start_time = time.time() try: - node = self.client.get_node(node) + node_obj = self.client.get_node(node) except Exception as e: trace = traceback.format_exc() logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) @@ -278,7 +280,7 @@ class OpcRepository(): ) try: - node.write_value(ua_data) + await node_obj.write_value(ua_data) metrics.PREDICTION_OPC_WRITING_COUNT.labels( pod_id=self.pod_id, diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 7f668ed..2901859 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -64,6 +64,9 @@ async def main(): notification_handler=notification_handler ) + logger.custom_info('Initializing OPC...', metadata) + await activities.init_opc() + logger.custom_info( f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) @@ -74,7 +77,7 @@ async def main(): ) ) - logger.custom_info('Starting Temporal Client...', metadata) + logger.custom_info(f'Starting Temporal Client at {host}...', metadata) temporal_client = await client.Client.connect( target_host=host, @@ -151,7 +154,7 @@ async def main(): if notification_handler: notification_handler.shutdown() if activities: - activities.shutdown() + await activities.shutdown() # Exit with a non-zero status code to indicate failure to Kubernetes metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN sys.exit(1) diff --git a/run_local.sh b/run_local.sh new file mode 100755 index 0000000..2bbd5c2 --- /dev/null +++ b/run_local.sh @@ -0,0 +1,18 @@ +#!/bin/bash + +# Exit on any error +set -e + +echo "Activating virtual environment..." +source ./venv/bin/activate + +echo "Loading environment variables from .env..." +if [ -f .env ]; then + export $(cat .env | grep -v '^#' | xargs) + echo "Environment variables loaded from .env" +else + echo "Warning: .env file not found. Continuing without environment variables." +fi + +echo "Starting ingestor application..." +python -m laborious.worker.worker From 09d2eb1436dbda60398b1ca4f518ce9fcb0c43c1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 26 Aug 2025 16:49:44 -0300 Subject: [PATCH 2/5] Remove .env file containing simulator configuration --- .env | 4 ---- 1 file changed, 4 deletions(-) delete mode 100644 .env diff --git a/.env b/.env deleted file mode 100644 index af9d68d..0000000 --- a/.env +++ /dev/null @@ -1,4 +0,0 @@ -# === Simulator Git Repo === -# Use SSH format because the Dockerfile uses SSH to clone -SIMULATOR_GIT_REPO=git@github.com:Aignosi/sientia-dataops-opc_simulator.git -SIMULATOR_GIT_BRANCH=main From 390e439a18d2ce6b2a3e74fac756c2356ab7107b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 10:48:56 -0300 Subject: [PATCH 3/5] SIENTIAPDE-1205 Update values.yaml and enhance logging in OPC activity - Change replicaCount from 5 to 1 for reduced resource allocation. - Update image tag in values.yaml to 0.4.5 for the latest version. - Modify GITHUB_BRANCH environment variable for improved async handling. - Add logging statements in OPC class to track server initialization and connection status. --- laborious/activities/opc.py | 7 ++++++- values.yaml | 6 +++--- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 56610d1..019de79 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -29,6 +29,8 @@ class OPC(BaseActivity): self.opc_servers = opc_servers async def init_opc(self): + + self.logger.info("Initializing OPC servers...") for id, server in self.opc_servers.items(): self.opc_repository[id] = OpcRepository( id=server['id'], @@ -58,6 +60,9 @@ class OPC(BaseActivity): attachment_content=error_data.get( 'attachment_content', None) ) + else: + self.logger.info( + f"OPC server {id} connected successfully.") async def write_data(self, server_id: str, tag: str, data: Any, data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: @@ -193,7 +198,7 @@ class OPC(BaseActivity): success = success and local_success self.info( - f"Data written to OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata) + f"Process completed for OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata) return self.process_confidence(data, success, metadata) diff --git a/values.yaml b/values.yaml index 0a59346..0b5632e 100644 --- a/values.yaml +++ b/values.yaml @@ -3,7 +3,7 @@ # Declare variables to be passed into your templates. # This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/ -replicaCount: 5 +replicaCount: 1 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ image: @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.4" + tag: "0.4.5" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal" + value: "SIENTIAPDE-1205-alterar-opc-para-assincrono" - name: PYTHON_APP value: "laborious.worker.worker" From df2c64a1fa78f003ab0adac53a7675a4f7c58e0b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 14:28:52 -0300 Subject: [PATCH 4/5] SIENTIAPDE-1205 Revert image tag in values.yaml to 0.4.4 and rename shutdown method in OpcRepository class - Changed image tag in values.yaml from 0.4.5 back to 0.4.4. - Renamed __del__ method to shutdown in OpcRepository for clarity. - Enhanced connection validation logic in OpcRepository to improve error handling and logging. --- laborious/utils/repository/opc_repository.py | 47 ++++++++++++-------- values.yaml | 2 +- 2 files changed, 29 insertions(+), 20 deletions(-) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 6742194..1315c44 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -158,7 +158,7 @@ class OpcRepository(): f"Failed to disconnect from OPC server: {e}", self.metadata) self.client = None - def __del__(self): + def shutdown(self): """ Disconnects from the OPC server when the object is destroyed. """ @@ -170,7 +170,8 @@ class OpcRepository(): async def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ - Validates the connection to the OPC server. + Validates the connection to the OPC server using protocol state checking. + If the connection is not established, it attempts to reconnect. If the connection is established but the client is not connected, it attempts to reconnect. @@ -198,28 +199,36 @@ class OpcRepository(): # Check if client is connected using asyncua's connection state try: - # Try to get a simple node to test connection - await self.client.get_node("ns=0;i=2253") # Server node - except Exception: - self.logger.custom_error( - f"OPC server {self.id} is not connected", self.metadata) - - if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( - ) > self.reconnection_interval: - await self.disconnect() - self.logger.custom_info( - f"Trying to reconnect to OPC server {self.id}...", self.metadata) - return await self.connect() + if self.client.uaclient.protocol is None or self.client.uaclient.protocol.state == "closed": + # OPC server is not connected + self.logger.custom_error( + f"OPC server {self.id} is not connected", self.metadata) + if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( + ) > self.reconnection_interval: + await self.disconnect() + self.logger.custom_info( + f"Trying to reconnect to OPC server {self.id}...", self.metadata) + return await self.connect() + return False, { + "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}", + "message": f"OPC server {self.id} is not connected, waiting for next reconnection window...", + "block": "opc_repository", + "level": NotificationLevel.WARNING + } + return True, {} + except Exception as e: + trace = traceback.format_exc() + message = f"Failed to validate connection to OPC server: {e}" + self.logger.custom_error(message, self.metadata) return False, { - "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}", - "message": f"OPC server {self.id} is not connected, waiting for next reconnection window...", + "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{self.id}", + "message": message, "block": "opc_repository", - "level": NotificationLevel.WARNING + "level": NotificationLevel.ERROR, + "attachment_content": trace } - return True, {} - async def write_data(self, node: str, value: Any, data_type: str, logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """ diff --git a/values.yaml b/values.yaml index 0b5632e..76ca717 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.5" + tag: "0.4.4" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: From f937eca1518ce43e2e9594cad73b17299606cbde Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 28 Aug 2025 10:21:20 -0300 Subject: [PATCH 5/5] SIENTIAPDE-1205 Refactor OpcRepository and update tests for async handling - Removed the shutdown method from OpcRepository and adjusted the disconnect logic. - Updated tests in test_activities.py and test_opc.py to support async shutdown functionality. - Enhanced test cases in test_opc_repository.py to ensure proper async behavior and error handling in OpcRepository methods. --- laborious/utils/repository/opc_repository.py | 10 - run_coverage.sh | 11 + tests/laborious/activities/test_activities.py | 7 +- tests/laborious/activities/test_opc.py | 115 ++++++-- .../utils/repository/test_opc_repository.py | 248 +++++++++++------- 5 files changed, 260 insertions(+), 131 deletions(-) create mode 100755 run_coverage.sh diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 1315c44..3439a19 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -158,16 +158,6 @@ class OpcRepository(): f"Failed to disconnect from OPC server: {e}", self.metadata) self.client = None - def shutdown(self): - """ - Disconnects from the OPC server when the object is destroyed. - """ - try: - asyncio.run(self.disconnect()) - except Exception as e: - self.logger.custom_error( - f"Error in destructor: {e}", self.metadata) - async def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ Validates the connection to the OPC server using protocol state checking. diff --git a/run_coverage.sh b/run_coverage.sh new file mode 100755 index 0000000..f9af4cb --- /dev/null +++ b/run_coverage.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +# Exit on any error +set -e + +echo "Activating virtual environment..." +source ./venv/bin/activate + +pytest --cov=laborious --cov-report=html + +xdg-open htmlcov/index.html \ No newline at end of file diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index e482ca8..de418be 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -90,11 +90,12 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre ) +@mark.asyncio @patch('laborious.activities.activities.Postgres', return_value=MagicMock()) @patch('laborious.activities.activities.MLFlow', return_value=MagicMock()) @patch('laborious.activities.activities.OPC', return_value=MagicMock()) -def test_shutdown(mock_opc_init, - _mock_mlflow_init, mock_postgres_init): +async def test_shutdown(mock_opc_init, + _mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', 'port': 5432, @@ -129,6 +130,6 @@ def test_shutdown(mock_opc_init, notification_handler=notification_handler ) - activities.shutdown() + await activities.shutdown() mock_opc_init.shutdown.assert_called_once() mock_postgres_init.close.assert_called_once() diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index 37ae391..07f00b8 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -1,7 +1,8 @@ -from unittest.mock import patch, MagicMock, ANY, call +from unittest.mock import patch, MagicMock, ANY, call, AsyncMock from pandas import DataFrame from pytest import fixture, mark -from laborious.activities.opc import NotificationLevel +import pytest_asyncio +from sientia_do.notifications.models import NotificationLevel from laborious.activities.opc import OPC @@ -15,27 +16,42 @@ metadata = { } +def test__init__(): + servers = { + 'server1': 'config' + } + opc = OPC( + opc_servers=servers, + logger=MagicMock(), + notification_handler=MagicMock() + ) + + assert opc.opc_servers == servers + assert opc.opc_repository == {} + + +@mark.asyncio @patch("laborious.activities.opc.OpcRepository") @patch("laborious.activities.opc.OPC.send_notification") -def test___init__(mock_send_notification, mock_opc_repository): +async def test_init_opc(mock_send_notification, mock_opc_repository): mock_logger = MagicMock() server1 = MagicMock( - connect=MagicMock(return_value=(True, {})), - write_data=MagicMock(return_value=(True, {})) + connect=AsyncMock(return_value=(True, {})), + write_data=AsyncMock(return_value=(True, {})) ) server2 = MagicMock( - connect=MagicMock(return_value=(True, {})), - write_data=MagicMock(return_value=(True, {})) + connect=AsyncMock(return_value=(True, {})), + write_data=AsyncMock(return_value=(True, {})) ) server3 = MagicMock( - connect=MagicMock(return_value=(False, { + connect=AsyncMock(return_value=(False, { 'notification_id': 'OPC_CONNECTION_ERROR_server3', 'message': 'Failed to connect to OPC server: Test error', 'block': 'opc_repository', 'level': NotificationLevel.ERROR, 'attachment_content': 'Test error' })), - write_data=MagicMock(return_value=(True, {})) + write_data=AsyncMock(return_value=(True, {})) ) mock_opc_repository.side_effect = [server1, server2, server3] mock_notification_handler = MagicMock() @@ -73,6 +89,7 @@ def test___init__(mock_send_notification, mock_opc_repository): logger=mock_logger, notification_handler=mock_notification_handler ) + await opc.init_opc() assert opc.opc_servers == servers assert opc.logger == mock_logger @@ -129,9 +146,9 @@ def test___init__(mock_send_notification, mock_opc_repository): ]) -@fixture +@pytest_asyncio.fixture @patch("laborious.activities.opc.OpcRepository") -def opc(mock_opc_repository): +async def opc(mock_opc_repository): servers = { 'server1': { 'id': 'server1', @@ -144,10 +161,10 @@ def opc(mock_opc_repository): } } - mock_opc_repository.return_value.write_data = MagicMock( + mock_opc_repository.return_value.write_data = AsyncMock( return_value=(True, {}) ) - mock_opc_repository.return_value.connect = MagicMock( + mock_opc_repository.return_value.connect = AsyncMock( return_value=(True, {}) ) opc = OPC( @@ -155,7 +172,7 @@ def opc(mock_opc_repository): logger=MagicMock(), notification_handler=MagicMock() ) - + await opc.init_opc() opc.send_notification = MagicMock() return opc @@ -169,14 +186,17 @@ WRITE_DATA_CASES = [ @mark.parametrize('tag,data_type,data', WRITE_DATA_CASES) -def test_write_data_success(opc, tag, data_type, data): - assert opc.write_data(server_id='server1', tag=tag, data=data, - data_type=data_type, tag_type='prediction', metadata=metadata) +@mark.asyncio +async def test_write_data_success(opc, tag, data_type, data): + result = await opc.write_data(server_id='server1', tag=tag, data=data, + data_type=data_type, tag_type='prediction', metadata=metadata) + assert result is True opc.opc_repository['server1'].write_data.assert_called_once_with( tag, data, data_type, opc.logger, metadata) -def test_write_data_failed(opc): +@mark.asyncio +async def test_write_data_failed(opc): opc.opc_repository['server1'].write_data.return_value = (False, { 'notification_id': 'OPC_WRITE_DATA_ERROR_server1', 'message': 'Failed to write data to OPC server: Test error', @@ -185,8 +205,9 @@ def test_write_data_failed(opc): 'attachment_content': 'Test error' }) - assert opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) is False + result = await opc.write_data(server_id='server1', tag='tag1', data=50, + data_type='int', tag_type='prediction', metadata=metadata) + assert result is False opc.send_notification.assert_called_once_with( metadata=metadata, @@ -198,13 +219,14 @@ def test_write_data_failed(opc): ) -def test_write_data_exception(opc): +@mark.asyncio +async def test_write_data_exception(opc): opc.opc_repository['server1'].write_data.side_effect = Exception( "Test error") try: - opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) + await opc.write_data(server_id='server1', tag='tag1', data=50, + data_type='int', tag_type='prediction', metadata=metadata) except Exception: opc.send_notification.assert_called_once_with( @@ -242,7 +264,7 @@ async def test_write_opc_data_success(opc): } # Act - opc.write_data = MagicMock() + opc.write_data = AsyncMock(return_value=True) opc.process_confidence = MagicMock(return_value={'data': 'data'}) output = await opc.write_opc_data(input_data) @@ -281,8 +303,38 @@ async def test_write_opc_data_empty_config(opc): }, 'opc_servers': ['server1'], 'opc_output_config': { - 'prediction_tags': {}, - 'confidence_tags': {} + 'server1': { + 'prediction_tags': {}, + 'confidence_tags': {} + } + } + } + + # Act + await opc.write_opc_data(input_data) + + # Assert + opc.opc_repository['server1'].write_data.assert_not_called() + + +@mark.asyncio +async def test_write_opc_data_no_validate_server(opc): + opc.validate_server = MagicMock(return_value=False) + input_data = { + **metadata, + 'data': { + 'prediction': [0.75], + 'prediction_confidence': [0.95] + }, + 'opc_output_config': { + 'server1': { + 'prediction_tags': { + 'tag1': {'data_type': 'float'} + }, + 'confidence_tags': { + 'tag2': {'data_type': 'float'} + } + } } } @@ -305,6 +357,13 @@ def test_process_confidence(opc, data, success, expected): assert result['prediction_confidence'][0] == expected -def test_shutdown(opc): - opc.shutdown() +def test_validate_server(opc): + assert opc.validate_server('server1', metadata) is True + assert opc.validate_server('server2', metadata) is False + + +@mark.asyncio +async def test_shutdown(opc): + opc.opc_repository['server1'].disconnect = AsyncMock(return_value=True) + await opc.shutdown() opc.opc_repository['server1'].disconnect.assert_called_once() diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index cf9b493..bc84db8 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -1,17 +1,17 @@ -from unittest.mock import Mock, patch, MagicMock, ANY, call +import pytest +from unittest.mock import AsyncMock, Mock, patch, MagicMock, ANY, call from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from pytest import fixture from laborious.utils.repository.opc_repository import OpcRepository from sientia_do.notifications.models import NotificationLevel from datetime import datetime -@fixture +@pytest.fixture def mock_logger(): return Mock() -@fixture +@pytest.fixture def opc_repository(mock_logger): return OpcRepository( id="test_repo", @@ -26,10 +26,10 @@ def opc_repository(mock_logger): ) -@fixture +@pytest.fixture def mock_client(): with patch('laborious.utils.repository.opc_repository.Client') as mock: - client_instance = MagicMock() + client_instance = AsyncMock() mock.return_value = client_instance yield client_instance @@ -57,9 +57,10 @@ def test_init(opc_repository): assert opc_repository.error_count == 0 -def test_set_security(opc_repository, mock_client): +@pytest.mark.asyncio +async def test_set_security(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.set_security() + await opc_repository.set_security() mock_client.application_uri = "urn:test:server" mock_client.set_security.assert_called_once_with( @@ -72,50 +73,59 @@ def test_set_security(opc_repository, mock_client): assert mock_client.session_timeout == 10000000 -def test_set_security_missing_certificates(opc_repository): +@pytest.mark.asyncio +async def test_set_security_missing_certificates(opc_repository): opc_repository.cert_path = None opc_repository.private_key_path = None try: - opc_repository.set_security() + await opc_repository.set_security() except ValueError as e: assert str( e) == "Certificate and private key paths must be provided for secure connection." -def test_connect_with_security(opc_repository, mock_client): - opc_repository.try_connect = MagicMock() - opc_repository.connect() +@pytest.mark.asyncio +async def test_connect_with_security(opc_repository, mock_client): + opc_repository.try_connect = AsyncMock(return_value=(True, {})) + result = await opc_repository.connect() opc_repository.try_connect.assert_called_once() assert opc_repository.client == mock_client + assert result == (True, {}) -def test_connect_without_security(opc_repository, mock_client): +@pytest.mark.asyncio +async def test_connect_without_security(opc_repository, mock_client): opc_repository.cert_path = None - opc_repository.try_connect = MagicMock() - opc_repository.set_security = MagicMock() - opc_repository.connect() + opc_repository.try_connect = AsyncMock(return_value=(True, {})) + opc_repository.set_security = AsyncMock() + result = await opc_repository.connect() opc_repository.try_connect.assert_called_once() opc_repository.set_security.assert_not_called() assert opc_repository.client == mock_client + assert result == (True, {}) -def test_try_connect_sucess(opc_repository): +@pytest.mark.asyncio +async def test_try_connect_success(opc_repository): opc_repository.last_reconnection_time = None - opc_repository.client = MagicMock() - opc_repository.try_connect() + opc_repository.client = AsyncMock() + result = await opc_repository.try_connect() + opc_repository.client.connect.assert_called_once() assert opc_repository.last_reconnection_time is not None + assert result == (True, {}) -def test_try_connect_fail(opc_repository): +@pytest.mark.asyncio +async def test_try_connect_fail(opc_repository): opc_repository.last_reconnection_time = None opc_repository.client = MagicMock() opc_repository.client.connect.side_effect = Exception("Test error") - is_connected, error_data = opc_repository.try_connect() + is_connected, error_data = await opc_repository.try_connect() opc_repository.client.connect.assert_called_once() assert is_connected is False @@ -126,18 +136,26 @@ def test_try_connect_fail(opc_repository): assert error_data['attachment_content'] is not None -def test_disconnect(opc_repository, mock_client): +@pytest.mark.asyncio +async def test_disconnect(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.disconnect() + await opc_repository.disconnect() mock_client.disconnect.assert_called_once() assert opc_repository.client is None -def test_disconnect_error(opc_repository, mock_client): +@pytest.mark.asyncio +async def test_disconnect_no_client(opc_repository): + opc_repository.client = None + assert await opc_repository.disconnect() is None + + +@pytest.mark.asyncio +async def test_disconnect_error(opc_repository, mock_client): opc_repository.client = mock_client mock_client.disconnect.side_effect = Exception("Test error") - opc_repository.disconnect() + await opc_repository.disconnect() opc_repository.logger.custom_error.assert_called_once_with( "Failed to disconnect from OPC server: Test error", @@ -146,21 +164,25 @@ def test_disconnect_error(opc_repository, mock_client): assert opc_repository.client is None -def test_validate_connection_none_client(opc_repository): +@pytest.mark.asyncio +async def test_validate_connection_none_client(opc_repository): opc_repository.client = None - opc_repository.connect = MagicMock() - response = opc_repository.validate_connection() - assert response + opc_repository.connect = AsyncMock(return_value=(True, {})) + response = await opc_repository.validate_connection() + assert response == (True, {}) opc_repository.connect.assert_called_once() -def test_validate_connection_error_count_disconnect_error(opc_repository): +@pytest.mark.asyncio +async def test_validate_connection_error_count_disconnect_error(opc_repository): opc_repository.error_count = 6 - opc_repository.client = MagicMock() - opc_repository.disconnect = MagicMock(side_effect=Exception("Test error")) - opc_repository.connect = MagicMock() + opc_repository.client = AsyncMock() + opc_repository.disconnect = AsyncMock( + side_effect=Exception("Test error") + ) + opc_repository.connect = AsyncMock(return_value=(True, {})) - response = opc_repository.validate_connection() + response = await opc_repository.validate_connection() assert response == opc_repository.connect.return_value opc_repository.disconnect.assert_called_once() opc_repository.connect.assert_called_once() @@ -171,18 +193,37 @@ def test_validate_connection_error_count_disconnect_error(opc_repository): ) -@patch('laborious.utils.repository.opc_repository.hasattr', return_value=True) -@patch('laborious.utils.repository.opc_repository.datetime', - MagicMock(now=MagicMock(return_value=datetime(2025, 1, 1, 0, 0, 0)))) -def test_validate_connection_lost_not_time_to_reconect(_mock_datetime, opc_repository): +@pytest.mark.asyncio +async def test_validate_connection_error_validate_connection_error(opc_repository): + opc_repository.client = MagicMock( + uaclient=Exception("Test error") + ) + opc_repository.error_count = 0 + + response = await opc_repository.validate_connection() + + assert response == (False, { + "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}", + "message": "Failed to validate connection to OPC server: 'Exception' object has no attribute 'protocol'", + "block": "opc_repository", + "level": NotificationLevel.ERROR, + "attachment_content": ANY + }) + + +@pytest.mark.asyncio +@patch('laborious.utils.repository.opc_repository.datetime') +async def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository): + _mock_datetime.now = MagicMock( + return_value=datetime(2025, 1, 1, 0, 0, 0)) opc_repository.error_count = 0 opc_repository.client = MagicMock() - opc_repository.client.aio_obj.uaclient.protocol = None + opc_repository.client.uaclient.protocol = None opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.try_connect = MagicMock() + opc_repository.connect = MagicMock(return_value=(True, {})) - response = opc_repository.validate_connection() - opc_repository.try_connect.assert_not_called() + response = await opc_repository.validate_connection() + opc_repository.connect.assert_not_called() assert response == (False, { "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}", "message": f"OPC server {opc_repository.id} is not connected, waiting for next reconnection window...", @@ -191,96 +232,120 @@ def test_validate_connection_lost_not_time_to_reconect(_mock_datetime, opc_repos }) -@patch('laborious.utils.repository.opc_repository.hasattr', return_value=True) -@patch('laborious.utils.repository.opc_repository.datetime', - MagicMock(now=MagicMock(return_value=datetime(2025, 1, 1, 1, 0, 0)))) -def test_validate_connection_lost_time_to_reconect(_mock_datetime, opc_repository): +@pytest.mark.asyncio +@patch('laborious.utils.repository.opc_repository.datetime') +async def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository): + mock_datetime.now = MagicMock( + return_value=datetime(2025, 1, 1, 1, 0, 0)) opc_repository.error_count = 0 - opc_repository.client = MagicMock() - opc_repository.client.aio_obj.uaclient.protocol = None + opc_repository.client = AsyncMock() + opc_repository.client.uaclient.protocol = None opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock() + opc_repository.connect = AsyncMock(return_value=(True, {})) - response = opc_repository.validate_connection() + response = await opc_repository.validate_connection() opc_repository.connect.assert_called_once() assert response == opc_repository.connect.return_value -def test_validate_connection_failed(opc_repository): +@pytest.mark.asyncio +async def test_validate_connection_success(opc_repository): opc_repository.client = MagicMock() opc_repository.error_count = 0 + opc_repository.client.uaclient.protocol = MagicMock() + opc_repository.client.uaclient.protocol.state = "open" - output = opc_repository.validate_connection() + output = await opc_repository.validate_connection() assert output == (True, {}) -def test_write_data_validate_connection_do_nothing(opc_repository): - opc_repository.validate_connection = MagicMock(return_value=(True, {})) - opc_repository.client = MagicMock() - opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata) +@pytest.mark.asyncio +async def test_write_data_validate_connection_do_nothing(opc_repository): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) + opc_repository.client = AsyncMock( + get_node=MagicMock() + ) + mock_node = AsyncMock() + opc_repository.client.get_node.return_value = mock_node + + result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata['metadata']) + opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") + assert result == (True, {}) -def test_write_data_validate_connection_failed(opc_repository): - opc_repository.validate_connection = MagicMock(return_value=(False, {})) - opc_repository.client = MagicMock() +@pytest.mark.asyncio +async def test_write_data_validate_connection_failed(opc_repository): + opc_repository.validate_connection = AsyncMock(return_value=(False, {})) + opc_repository.client = AsyncMock() opc_repository.error_count = 0 - opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata) + + result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata['metadata']) + opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_not_called() + assert result == (False, {}) -def test_write_data_get_node_failed(opc_repository): - opc_repository.validate_connection = MagicMock(return_value=(True, {})) - opc_repository.client = MagicMock() +@pytest.mark.asyncio +async def test_write_data_get_node_failed(opc_repository): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) + opc_repository.client = AsyncMock() opc_repository.error_count = 0 - opc_repository.client.get_node.side_effect = Exception("Test error") - is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata) + opc_repository.client.get_node = MagicMock( + side_effect=Exception("Test error")) + + is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata['metadata']) + opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") assert is_success is False assert error_data['notification_id'] == f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to get node from OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}" + assert error_data['message'] == "Failed to get node from OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" assert error_data['block'] == "opc_repository" assert error_data['level'] == NotificationLevel.ERROR assert error_data['attachment_content'] is not None -def test_write_data_invalid_data_type(opc_repository, mock_client): - opc_repository.validate_connection = MagicMock(return_value=(True, {})) +@pytest.mark.asyncio +async def test_write_data_invalid_data_type(opc_repository, mock_client): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = mock_client - mock_node = MagicMock() - mock_client.get_node.return_value = mock_node + mock_node = AsyncMock() + mock_client.get_node = MagicMock(return_value=mock_node) + + is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "invalid_type", opc_repository.logger, metadata['metadata']) - is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, - "invalid_type", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") assert is_success is False assert error_data['notification_id'] == f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Unsupported data type: invalid_type | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}" + assert error_data['message'] == "Unsupported data type: invalid_type | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" assert error_data['block'] == "opc_repository" assert error_data['level'] == NotificationLevel.ERROR assert error_data.get('attachment_content') is None +@pytest.mark.asyncio @patch('laborious.utils.repository.opc_repository.metrics') -def test_write_data(mock_metrics, opc_repository, mock_client): - opc_repository.validate_connection = MagicMock(return_value=(True, {})) +async def test_write_data(mock_metrics, opc_repository, mock_client): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = mock_client - mock_node = MagicMock() - mock_client.get_node.return_value = mock_node + mock_node = AsyncMock() + mock_client.get_node = MagicMock(return_value=mock_node) - opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) + result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata['metadata']) mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_node.write_value.assert_called_once() + assert result == (True, {}) mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.assert_called_once_with( pod_id=opc_repository.pod_id, @@ -300,21 +365,24 @@ def test_write_data(mock_metrics, opc_repository, mock_client): ANY) -def test_write_data_write_value_failed(opc_repository, mock_client): - opc_repository.validate_connection = MagicMock(return_value=(True, {})) +@pytest.mark.asyncio +async def test_write_data_write_value_failed(opc_repository, mock_client): + opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = mock_client - mock_node = MagicMock() + mock_node = AsyncMock() opc_repository.error_count = 0 - mock_client.get_node.return_value = mock_node + mock_client.get_node = MagicMock(return_value=mock_node) mock_node.write_value.side_effect = Exception("Test error") - is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata) + + is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata['metadata']) + opc_repository.validate_connection.assert_called_once() mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_node.write_value.assert_called_once() assert is_success is False assert error_data['notification_id'] == f"OPC_WRITE_DATA_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to write data to OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}" + assert error_data['message'] == "Failed to write data to OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" assert error_data['block'] == "opc_repository" assert error_data['level'] == NotificationLevel.ERROR assert error_data['attachment_content'] is not None