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 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..019de79 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -26,7 +26,12 @@ 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): + + self.logger.info("Initializing OPC servers...") + for id, server in self.opc_servers.items(): self.opc_repository[id] = OpcRepository( id=server['id'], url=server['url'], @@ -39,7 +44,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={ @@ -55,9 +60,12 @@ class OPC(BaseActivity): attachment_content=error_data.get( 'attachment_content', None) ) + else: + self.logger.info( + f"OPC server {id} connected successfully.") - 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 +81,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 +121,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 +144,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,12 +193,12 @@ 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 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) @@ -221,6 +229,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..3439a19 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: @@ -157,19 +158,10 @@ class OpcRepository(): f"Failed to disconnect from OPC server: {e}", self.metadata) self.client = None - def __del__(self): + async def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ - Disconnects from the OPC server when the object is destroyed. - """ - try: - self.disconnect() - except Exception as e: - self.logger.custom_error( - f"Error in destructor: {e}", self.metadata) + Validates the connection to the OPC server using protocol state checking. - 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. If the connection is established but the client is not connected, it attempts to reconnect. @@ -179,13 +171,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,33 +185,42 @@ 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() + return await 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"): - - 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() - self.logger.custom_info( - f"Trying to reconnect to OPC server {self.id}...", self.metadata) - return self.connect() + # Check if client is connected using asyncua's connection state + try: + 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, {} - - 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 +232,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 +240,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 +279,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_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/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 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 diff --git a/values.yaml b/values.yaml index 0a59346..76ca717 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: @@ -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"