gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 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 df2c64a1fa
commit f937eca151
5 changed files with 260 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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