gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #18 from Aignosi/SIENTIAPDE-1205-alterar-opc-para-assincrono

Browse Source 
SIENTIAPDE-1205: Refactor OPC Repository for Asynchronous Handling and Configuration Updates
This commit is contained in:
Bruno Domingues
2025-08-29 14:00:56 +00:00
committed by GitHub
parent a8f7d3562e f937eca151
commit 5a456dcc7a
12 changed files with 358 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.env
View File

@@ -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

4
.gitignore vendored
View File

@@ -41,4 +41,6 @@ htmlcov/
# git keys
git_key*
git_log
git_log
.env

4
laborious/activities/activities.py
View File

@@ -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)

32
laborious/activities/opc.py
View File

@@ -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()

95
laborious/utils/repository/opc_repository.py
View File

@@ -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,

7
laborious/worker/worker.py
View File

@@ -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)

11
run_coverage.sh Executable file
View File

@@ -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

18
run_local.sh Executable file
View File

@@ -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

7
tests/laborious/activities/test_activities.py
View File

@@ -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()

115
tests/laborious/activities/test_opc.py
View File

@@ -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()

248
tests/laborious/utils/repository/test_opc_repository.py
View File

@@ -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

4
values.yaml
View File

@@ -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"