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

SIENTIAPDE-1163

Browse Source 
SIENTIAPDE-1163 Enhance OPC tests to improve error handling and notifications. Added tests for MongoDB configuration and refined connection error handling in OPC repository.
This commit is contained in:
Vitor Pimentel dos Santos
2025-07-17 15:21:10 -03:00
parent 488674c7e1
commit b440c7d547
3 changed files with 139 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,11 +16,28 @@ metadata = {
@patch("laborious.activities.opc.OpcRepository")
def test___init__(mock_opc_repository):
@patch("laborious.activities.opc.OPC.send_notification")
def test___init__(mock_send_notification, mock_opc_repository):
mock_logger = MagicMock()
server1 = MagicMock()
server2 = MagicMock()
mock_opc_repository.side_effect = [server1, server2]
server1 = MagicMock(
connect=MagicMock(return_value=(True, {})),
write_data=MagicMock(return_value=(True, {}))
)
server2 = MagicMock(
connect=MagicMock(return_value=(True, {})),
write_data=MagicMock(return_value=(True, {}))
)
server3 = MagicMock(
connect=MagicMock(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, {}))
)
mock_opc_repository.side_effect = [server1, server2, server3]
mock_notification_handler = MagicMock()
servers = {
'server1': {
@@ -40,6 +57,15 @@ def test___init__(mock_opc_repository):
'private_key_path': '',
'server_cert_path': '',
'reconnection_interval': 60,
},
'server3': {
'id': 'server3',
'url': 'http://localhost:8080',
'server_uri': 'opc.tcp://localhost:4840',
'cert_path': '',
'private_key_path': '',
'server_cert_path': '',
'reconnection_interval': 60,
}
}
opc = OPC(
@@ -84,6 +110,22 @@ def test___init__(mock_opc_repository):
server1.connect.assert_called_once()
server2.connect.assert_called_once()
mock_send_notification.assert_has_calls([
call(
metadata={
'model_id': '-',
'model_name': '-',
'workflow_name': '-',
'schedule_name': 'INITIALIZATION'
},
notification_id="OPC_CONNECTION_ERROR_server3",
message="Failed to connect to OPC server: Test error",
block="opc_repository",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
])
@fixture
@patch("laborious.activities.opc.OpcRepository")
@@ -99,8 +141,12 @@ def opc(mock_opc_repository):
'reconnection_interval': 60,
}
}
mock_opc_repository.write_data = MagicMock(
return_value=True
mock_opc_repository.return_value.write_data = MagicMock(
return_value=(True, {})
)
mock_opc_repository.return_value.connect = MagicMock(
return_value=(True, {})
)
opc = OPC(
opc_servers=servers,
@@ -128,6 +174,28 @@ def test_write_data_success(opc, tag, data_type, data):
tag, data, data_type, opc.logger, metadata)
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',
'block': 'opc_repository',
'level': NotificationLevel.ERROR,
'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
opc.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="OPC_WRITE_DATA_ERROR_server1",
message="Failed to write data to OPC server: Test error",
block="opc_repository",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
def test_write_data_exception(opc):
opc.opc_repository['server1'].write_data.side_effect = Exception(
"Test error")