vitor-aignosi
2a6def4056
SIENTIAPDE-1811 Implement OPC write error handling and refactor tag writing logic - Introduced a new function `_apply_opc_write_error` to manage session and reconnect flags based on OPC write error responses. - Refactored the `_write_tags_from_config` method to streamline the writing of OPC tags for both prediction and confidence data. - Enhanced unit tests to cover various scenarios for OPC write errors, including session bad and reconnect in progress cases. - Updated existing tests to validate the new logic and ensure robust error handling.
739 lines
22 KiB
Python
739 lines
22 KiB
Python
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
|
|
|
|
import pytest_asyncio
|
|
from pandas import DataFrame
|
|
from pytest import mark
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
|
|
from laborious.activities.opc import (
|
|
OPC,
|
|
OPC_COMMENT_SEPARATOR,
|
|
OPC_RECONNECT_IN_PROGRESS_COMMENT,
|
|
OPC_SESSION_BAD_COMMENT_PREFIX,
|
|
OPC_SESSION_BAD_CONFIDENCE,
|
|
OPC_WRITTING_ERROR_CONFIDENCE,
|
|
OPC_WRITTING_ERROR_MESSAGE,
|
|
_apply_opc_write_error,
|
|
)
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': 'test_model',
|
|
'model_name': 'test_model',
|
|
'workflow_name': 'test_workflow',
|
|
'schema_name': 'test_schedule',
|
|
},
|
|
}
|
|
|
|
|
|
def test__init__():
|
|
servers = {'server1': {'id': 'server1'}}
|
|
opc = OPC(
|
|
opc_servers=servers,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=AsyncMock(),
|
|
)
|
|
|
|
assert opc.opc_servers == servers
|
|
assert opc.opc_repository == {}
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('laborious.activities.opc.OpcRepository')
|
|
@patch('laborious.activities.opc.OPC.send_notification_async')
|
|
async def test_init_opc(mock_send_notification, mock_opc_repository):
|
|
mock_logger = MagicMock()
|
|
mock_metrics_controller = AsyncMock()
|
|
server1 = MagicMock(
|
|
connect=AsyncMock(return_value=(True, {})), write_data=AsyncMock(return_value=(True, {}))
|
|
)
|
|
server2 = MagicMock(
|
|
connect=AsyncMock(return_value=(True, {})), write_data=AsyncMock(return_value=(True, {}))
|
|
)
|
|
server3 = MagicMock(
|
|
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=AsyncMock(return_value=(True, {})),
|
|
)
|
|
mock_opc_repository.side_effect = [server1, server2, server3]
|
|
mock_notification_handler = MagicMock()
|
|
servers = {
|
|
'server1': {
|
|
'server_name': 'server1',
|
|
'id': 'server1',
|
|
'url': 'http://localhost:8080',
|
|
'server_uri': 'opc.tcp://localhost:4840',
|
|
'cert_path': '',
|
|
'private_key_path': '',
|
|
'server_cert_path': '',
|
|
'reconnection_interval': 60,
|
|
},
|
|
'server2': {
|
|
'server_name': 'server2',
|
|
'id': 'server2',
|
|
'url': 'http://localhost:8080',
|
|
'server_uri': 'opc.tcp://localhost:4840',
|
|
'cert_path': '',
|
|
'private_key_path': '',
|
|
'server_cert_path': '',
|
|
'reconnection_interval': 60,
|
|
},
|
|
'server3': {
|
|
'server_name': '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(
|
|
opc_servers=servers,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
metrics_controller=mock_metrics_controller,
|
|
)
|
|
await opc.init_opc()
|
|
|
|
assert opc.opc_servers == servers
|
|
assert opc.logger == mock_logger
|
|
assert opc.notification_handler == mock_notification_handler
|
|
assert opc.opc_repository['server1'] == server1
|
|
assert opc.opc_repository['server2'] == server2
|
|
|
|
mock_opc_repository.assert_has_calls(
|
|
[
|
|
call(
|
|
opc_id='server1',
|
|
server_name='server1',
|
|
url='http://localhost:8080',
|
|
logger=mock_logger,
|
|
server_uri='opc.tcp://localhost:4840',
|
|
cert_path='',
|
|
private_key_path='',
|
|
server_cert_path='',
|
|
notification_handler=mock_notification_handler,
|
|
reconnection_interval=60,
|
|
metrics_controller=mock_metrics_controller,
|
|
),
|
|
]
|
|
)
|
|
mock_opc_repository.assert_has_calls(
|
|
[
|
|
call(
|
|
opc_id='server2',
|
|
server_name='server2',
|
|
url='http://localhost:8080',
|
|
logger=mock_logger,
|
|
server_uri='opc.tcp://localhost:4840',
|
|
cert_path='',
|
|
private_key_path='',
|
|
server_cert_path='',
|
|
notification_handler=mock_notification_handler,
|
|
reconnection_interval=60,
|
|
metrics_controller=mock_metrics_controller,
|
|
)
|
|
]
|
|
)
|
|
|
|
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,
|
|
)
|
|
]
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
@patch('laborious.activities.opc.OpcRepository')
|
|
async def opc(mock_opc_repository):
|
|
servers = {
|
|
'server1': {
|
|
'id': 'server1',
|
|
'server_name': 'server1',
|
|
'url': 'http://localhost:8080',
|
|
'server_uri': 'opc.tcp://localhost:4840',
|
|
'cert_path': '',
|
|
'private_key_path': '',
|
|
'server_cert_path': '',
|
|
'reconnection_interval': 60,
|
|
}
|
|
}
|
|
|
|
mock_opc_repository.return_value.write_data = AsyncMock(return_value=(True, {}))
|
|
mock_opc_repository.return_value.connect = AsyncMock(return_value=(True, {}))
|
|
opc = OPC(
|
|
opc_servers=servers,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=AsyncMock(),
|
|
)
|
|
await opc.init_opc()
|
|
opc.send_notification = MagicMock()
|
|
opc.send_notification_async = AsyncMock()
|
|
opc.emit_metric = AsyncMock()
|
|
return opc
|
|
|
|
|
|
WRITE_DATA_CASES = [
|
|
('tag1', 'int', 50),
|
|
('tag2', 'float', 50.5),
|
|
('tag3', 'bool', True),
|
|
('tag4', 'string', 'test'),
|
|
]
|
|
|
|
|
|
@mark.parametrize('tag,data_type,data', WRITE_DATA_CASES)
|
|
@mark.asyncio
|
|
async def test_write_data_success(opc, tag, data_type, data):
|
|
opc.opc_repository['server1'].write_data.return_value = (True, {'response_time': 0.1})
|
|
|
|
response_time, error_info = await opc.write_data(
|
|
server_id='server1',
|
|
tag=tag,
|
|
data=data,
|
|
data_type=data_type,
|
|
tag_type='prediction',
|
|
metadata=metadata,
|
|
)
|
|
assert response_time == 0.1
|
|
assert error_info is None
|
|
opc.opc_repository['server1'].write_data.assert_called_once_with(tag, data, data_type, metadata)
|
|
|
|
|
|
@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',
|
|
'block': 'opc_repository',
|
|
'level': NotificationLevel.ERROR,
|
|
'attachment_content': 'Test error',
|
|
},
|
|
)
|
|
|
|
response_time, error_info = await opc.write_data(
|
|
server_id='server1',
|
|
tag='tag1',
|
|
data=50,
|
|
data_type='int',
|
|
tag_type='prediction',
|
|
metadata=metadata,
|
|
)
|
|
assert response_time is None
|
|
assert error_info is not None
|
|
|
|
opc.send_notification_async.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,
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_data_exception(opc):
|
|
opc.opc_repository['server1'].write_data.side_effect = Exception('Test error')
|
|
|
|
try:
|
|
await opc.write_data(
|
|
server_id='server1',
|
|
tag='tag1',
|
|
data=50,
|
|
data_type='int',
|
|
tag_type='prediction',
|
|
metadata=metadata,
|
|
)
|
|
|
|
except Exception:
|
|
opc.send_notification_async.assert_called_once_with(
|
|
metadata=metadata,
|
|
notification_id='WRITE_OPC_PREDICTION_ERROR',
|
|
message='Error writing data to OPC server: Test error',
|
|
block='write_opc_data',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=ANY,
|
|
)
|
|
|
|
else:
|
|
raise AssertionError('Expected an exception to be raised')
|
|
|
|
|
|
@mark.parametrize(
|
|
'error_info,initial_seen,initial_status,initial_reconnect,expected',
|
|
[
|
|
(None, False, None, False, (False, None, False)),
|
|
({}, False, None, False, (False, None, False)),
|
|
(
|
|
{'opc_error_kind': 'session_bad', 'opc_status': 'BadSessionIdInvalid'},
|
|
False,
|
|
None,
|
|
False,
|
|
(True, 'BadSessionIdInvalid', False),
|
|
),
|
|
(
|
|
{'opc_error_kind': 'session_bad', 'opc_status': 'NewStatus'},
|
|
True,
|
|
'OldStatus',
|
|
False,
|
|
(True, 'NewStatus', False),
|
|
),
|
|
(
|
|
{'opc_error_kind': 'session_bad'},
|
|
True,
|
|
'KeptStatus',
|
|
False,
|
|
(True, 'KeptStatus', False),
|
|
),
|
|
(
|
|
{'opc_error_kind': 'reconnect_in_progress'},
|
|
False,
|
|
None,
|
|
False,
|
|
(False, None, True),
|
|
),
|
|
(
|
|
{'opc_error_kind': 'other'},
|
|
True,
|
|
'Status',
|
|
True,
|
|
(True, 'Status', True),
|
|
),
|
|
],
|
|
)
|
|
def test_apply_opc_write_error(
|
|
error_info, initial_seen, initial_status, initial_reconnect, expected
|
|
):
|
|
result = _apply_opc_write_error(
|
|
error_info,
|
|
initial_seen,
|
|
initial_status,
|
|
initial_reconnect,
|
|
)
|
|
assert result == expected
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_tags_from_config_prediction_success(opc):
|
|
opc.write_data = AsyncMock(return_value=(0.1, None))
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
tags_config = {'tag1': {'data_type': 'float'}}
|
|
|
|
response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
|
|
server_id='server1',
|
|
tags_config=tags_config,
|
|
data=data,
|
|
data_column='prediction',
|
|
tag_type='prediction',
|
|
log_label='Prediction data',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert response_times == {'tag1': 0.1}
|
|
assert session_bad is False
|
|
assert opc_status is None
|
|
assert reconnect is False
|
|
opc.write_data.assert_called_once_with(
|
|
server_id='server1',
|
|
tag='tag1',
|
|
data=0.75,
|
|
data_type='float',
|
|
tag_type='prediction',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_tags_from_config_confidence_success(opc):
|
|
opc.write_data = AsyncMock(return_value=(0.2, None))
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
tags_config = {'tag2': {'data_type': 'float'}}
|
|
|
|
response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
|
|
server_id='server1',
|
|
tags_config=tags_config,
|
|
data=data,
|
|
data_column='prediction_confidence',
|
|
tag_type='confidence',
|
|
log_label='Confidence data',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert response_times == {'tag2': 0.2}
|
|
assert session_bad is False
|
|
assert opc_status is None
|
|
assert reconnect is False
|
|
opc.write_data.assert_called_once_with(
|
|
server_id='server1',
|
|
tag='tag2',
|
|
data=0.95,
|
|
data_type='float',
|
|
tag_type='confidence',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_tags_from_config_write_failure(opc):
|
|
opc.write_data = AsyncMock(return_value=(None, {}))
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
|
|
response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
|
|
server_id='server1',
|
|
tags_config={'tag1': {'data_type': 'float'}},
|
|
data=data,
|
|
data_column='prediction',
|
|
tag_type='prediction',
|
|
log_label='Prediction data',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert response_times == {'tag1': None}
|
|
assert session_bad is False
|
|
assert opc_status is None
|
|
assert reconnect is False
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_tags_from_config_session_bad(opc):
|
|
opc.write_data = AsyncMock(
|
|
return_value=(
|
|
None,
|
|
{
|
|
'opc_error_kind': 'session_bad',
|
|
'opc_status': 'BadSessionIdInvalid',
|
|
},
|
|
)
|
|
)
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
|
|
response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
|
|
server_id='server1',
|
|
tags_config={'tag1': {'data_type': 'float'}},
|
|
data=data,
|
|
data_column='prediction',
|
|
tag_type='prediction',
|
|
log_label='Prediction data',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert response_times == {'tag1': None}
|
|
assert session_bad is True
|
|
assert opc_status == 'BadSessionIdInvalid'
|
|
assert reconnect is False
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_tags_from_config_reconnect_in_progress(opc):
|
|
opc.write_data = AsyncMock(
|
|
return_value=(
|
|
None,
|
|
{'opc_error_kind': 'reconnect_in_progress'},
|
|
)
|
|
)
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
|
|
response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
|
|
server_id='server1',
|
|
tags_config={'tag1': {'data_type': 'float'}},
|
|
data=data,
|
|
data_column='prediction',
|
|
tag_type='prediction',
|
|
log_label='Prediction data',
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert response_times == {'tag1': None}
|
|
assert session_bad is False
|
|
assert opc_status is None
|
|
assert reconnect is True
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_output_tags_success(opc):
|
|
opc._write_tags_from_config = AsyncMock(
|
|
side_effect=[
|
|
({'tag1': 0.1}, False, None, False),
|
|
({'tag2': 0.1}, False, None, False),
|
|
]
|
|
)
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
config = {
|
|
'prediction_tags': {'tag1': {'data_type': 'float'}},
|
|
'confidence_tags': {'tag2': {'data_type': 'float'}},
|
|
}
|
|
|
|
output_data, opc_metrics, session_bad, opc_status, reconnect = await opc.manage_output_tags(
|
|
server_id='server1',
|
|
config=config,
|
|
data=data,
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert output_data is True
|
|
assert opc_metrics == {'tag1': 0.1, 'tag2': 0.1}
|
|
assert session_bad is False
|
|
assert opc_status is None
|
|
assert reconnect is False
|
|
assert opc._write_tags_from_config.await_count == 2
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_output_tags_failed(opc):
|
|
opc._write_tags_from_config = AsyncMock(
|
|
side_effect=[
|
|
({'tag1': 0.1}, False, None, False),
|
|
({'tag2': None}, False, None, False),
|
|
]
|
|
)
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
config = {
|
|
'prediction_tags': {'tag1': {'data_type': 'float'}},
|
|
'confidence_tags': {'tag2': {'data_type': 'float'}},
|
|
}
|
|
|
|
output_data, opc_metrics, _, _, _ = await opc.manage_output_tags(
|
|
server_id='server1',
|
|
config=config,
|
|
data=data,
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert output_data is False
|
|
assert opc_metrics == {'tag1': 0.1, 'tag2': None}
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_output_tags_do_nothing(opc):
|
|
opc._write_tags_from_config = AsyncMock()
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
config = {'_invalid_key': {'tag1': {'data_type': 'float'}}}
|
|
|
|
output_data, opc_metrics, _, _, _ = await opc.manage_output_tags(
|
|
server_id='server1',
|
|
config=config,
|
|
data=data,
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert output_data is True
|
|
assert opc_metrics == {}
|
|
opc._write_tags_from_config.assert_not_called()
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('laborious.activities.opc.DataFrame')
|
|
async def test_write_opc_data_success(mock_dataframe, opc):
|
|
# Arrange
|
|
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'}},
|
|
}
|
|
},
|
|
}
|
|
|
|
# Act
|
|
opc.manage_output_tags = AsyncMock(
|
|
return_value=(True, {'tag1': 0.1, 'tag2': 0.2}, False, None, False)
|
|
)
|
|
|
|
opc.process_confidence = MagicMock(return_value={'data': 'data'})
|
|
output_data, opc_metrics = await opc.write_opc_data(input_data)
|
|
|
|
# Assert
|
|
assert output_data == {'data': 'data'}
|
|
assert opc_metrics == {'server1': {'tag1': 0.1, 'tag2': 0.2}}
|
|
opc.manage_output_tags.assert_called_once_with(
|
|
'server1',
|
|
input_data['opc_output_config']['server1'],
|
|
mock_dataframe.return_value,
|
|
metadata['metadata'],
|
|
)
|
|
opc.process_confidence.assert_called_once_with(
|
|
mock_dataframe.return_value,
|
|
True,
|
|
metadata['metadata'],
|
|
session_bad=False,
|
|
opc_status=None,
|
|
reconnect_in_progress=False,
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_write_opc_data_empty_config(opc):
|
|
# Arrange
|
|
input_data = {
|
|
**metadata,
|
|
'data': {'prediction': [0.75], 'prediction_confidence': [0.95]},
|
|
'opc_servers': ['server1'],
|
|
'opc_output_config': {'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 = AsyncMock(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'}},
|
|
}
|
|
},
|
|
}
|
|
|
|
# Act
|
|
await opc.write_opc_data(input_data)
|
|
|
|
# Assert
|
|
opc.opc_repository['server1'].write_data.assert_not_called()
|
|
|
|
|
|
@mark.parametrize(
|
|
'data,success,expected',
|
|
[
|
|
(DataFrame({'prediction_confidence': [0]}), True, 0),
|
|
(DataFrame({'prediction_confidence': [0]}), False, 12),
|
|
],
|
|
)
|
|
def test_process_confidence(opc, data, success, expected):
|
|
result = opc.process_confidence(data, success, metadata['metadata'])
|
|
assert result['prediction_confidence'][0] == expected
|
|
|
|
|
|
def test_process_confidence_session_bad(opc):
|
|
data = DataFrame({'prediction_confidence': [0.9]})
|
|
result = opc.process_confidence(
|
|
data,
|
|
False,
|
|
metadata['metadata'],
|
|
session_bad=True,
|
|
opc_status='BadSessionIdInvalid',
|
|
)
|
|
assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
|
|
assert result['comments'][0].startswith(OPC_SESSION_BAD_COMMENT_PREFIX)
|
|
assert 'BadSessionIdInvalid' in result['comments'][0]
|
|
|
|
|
|
def test_process_confidence_generic_failure(opc):
|
|
data = DataFrame({'prediction_confidence': [0.9]})
|
|
result = opc.process_confidence(data, False, metadata['metadata'])
|
|
assert result['prediction_confidence'][0] == OPC_WRITTING_ERROR_CONFIDENCE
|
|
assert result['comments'][0] == OPC_WRITTING_ERROR_MESSAGE
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_output_tags_merges_error_flags(opc):
|
|
opc._write_tags_from_config = AsyncMock(
|
|
side_effect=[
|
|
({'tag1': None}, True, 'BadSessionIdInvalid', False),
|
|
({'tag2': 0.2}, False, None, True),
|
|
]
|
|
)
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
config = {
|
|
'prediction_tags': {'tag1': {'data_type': 'float'}},
|
|
'confidence_tags': {'tag2': {'data_type': 'float'}},
|
|
}
|
|
|
|
(
|
|
success,
|
|
metrics,
|
|
session_bad_seen,
|
|
opc_status,
|
|
reconnect_in_progress,
|
|
) = await opc.manage_output_tags('server1', config, data, metadata['metadata'])
|
|
|
|
assert success is False
|
|
assert session_bad_seen is True
|
|
assert reconnect_in_progress is True
|
|
assert opc_status == 'BadSessionIdInvalid'
|
|
assert metrics == {'tag1': None, 'tag2': 0.2}
|
|
|
|
|
|
def test_process_confidence_reconnect_in_progress(opc):
|
|
data = DataFrame({'prediction_confidence': [0.9]})
|
|
result = opc.process_confidence(
|
|
data,
|
|
False,
|
|
metadata['metadata'],
|
|
reconnect_in_progress=True,
|
|
)
|
|
assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
|
|
assert result['comments'][0] == OPC_RECONNECT_IN_PROGRESS_COMMENT
|
|
|
|
|
|
def test_process_confidence_concatenates_multiple_comments(opc):
|
|
data = DataFrame({'prediction_confidence': [0.9]})
|
|
session_comment = f'{OPC_SESSION_BAD_COMMENT_PREFIX} BadSessionIdInvalid'
|
|
|
|
result = opc.process_confidence(
|
|
data,
|
|
False,
|
|
metadata['metadata'],
|
|
session_bad=True,
|
|
opc_status='BadSessionIdInvalid',
|
|
reconnect_in_progress=True,
|
|
)
|
|
|
|
assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
|
|
assert result['comments'][0] == OPC_COMMENT_SEPARATOR.join(
|
|
[session_comment, OPC_RECONNECT_IN_PROGRESS_COMMENT]
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_validate_server(opc):
|
|
assert await opc.validate_server('server1', metadata) is True
|
|
assert await opc.validate_server('server2', metadata) is False
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_close(opc):
|
|
opc.opc_repository['server1'].disconnect = AsyncMock(return_value=True)
|
|
await opc.close()
|
|
opc.opc_repository['server1'].disconnect.assert_called_once()
|