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

SIENTIAPDE-1205

Browse Source 
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.
This commit is contained in:
vitor-aignosi
2025-08-28 10:21:20 -03:00
parent fc799de42e
commit 7f3ecc9add
5 changed files with 260 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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