vitor-aignosi
00b6382a88
Simplify bandit nosec comments and mock metrics_controller with real spe - drift.py/simple_metrics.py: move nosec rationale to a code comment above the query instead of an inline suppression comment, keeping bandit happy with a plain `# nosec` - tests: replace bare `MagicMock()` with `MagicMock(spec=MetricsController)` for metrics_controller across activities/repository tests so mocks fail fast on attribute typos instead of silently accepting any call
754 lines
23 KiB
Python
754 lines
23 KiB
Python
from unittest.mock import ANY, MagicMock, call, patch
|
|
|
|
import pytest
|
|
from pandas import DataFrame
|
|
from pytest import mark
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
|
|
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=MagicMock(spec=MetricsController),
|
|
)
|
|
|
|
assert opc.opc_servers == servers
|
|
assert opc.opc_repository == {}
|
|
|
|
|
|
@patch('laborious.activities.opc.OpcRepository')
|
|
@patch('laborious.activities.opc.OPC.send_notification')
|
|
def test_init_opc(mock_send_notification, mock_opc_repository):
|
|
mock_logger = MagicMock()
|
|
mock_metrics_controller = MagicMock(spec=MetricsController)
|
|
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': {
|
|
'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,
|
|
)
|
|
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.fixture
|
|
@patch('laborious.activities.opc.OpcRepository')
|
|
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 = MagicMock(return_value=(True, {}))
|
|
mock_opc_repository.return_value.connect = MagicMock(return_value=(True, {}))
|
|
opc = OPC(
|
|
opc_servers=servers,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
opc.init_opc()
|
|
opc.send_notification = MagicMock()
|
|
opc.emit_metric_sync = MagicMock()
|
|
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)
|
|
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 = 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)
|
|
|
|
|
|
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 = 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.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')
|
|
|
|
try:
|
|
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(
|
|
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
|
|
|
|
|
|
def test_write_tags_from_config_prediction_success(opc):
|
|
opc.write_data = MagicMock(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 = 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'],
|
|
)
|
|
|
|
|
|
def test_write_tags_from_config_confidence_success(opc):
|
|
opc.write_data = MagicMock(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 = 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'],
|
|
)
|
|
|
|
|
|
def test_write_tags_from_config_write_failure(opc):
|
|
opc.write_data = MagicMock(return_value=(None, {}))
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
|
|
response_times, session_bad, opc_status, reconnect = 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
|
|
|
|
|
|
def test_write_tags_from_config_session_bad(opc):
|
|
opc.write_data = MagicMock(
|
|
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 = 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
|
|
|
|
|
|
def test_write_tags_from_config_reconnect_in_progress(opc):
|
|
opc.write_data = MagicMock(
|
|
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 = 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
|
|
|
|
|
|
def test_manage_output_tags_success(opc):
|
|
opc._write_tags_from_config = MagicMock(
|
|
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 = 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.call_count == 2
|
|
|
|
|
|
def test_manage_output_tags_failed(opc):
|
|
opc._write_tags_from_config = MagicMock(
|
|
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, _, _, _ = 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}
|
|
|
|
|
|
def test_manage_output_tags_do_nothing(opc):
|
|
opc._write_tags_from_config = MagicMock()
|
|
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
|
|
config = {'_invalid_key': {'tag1': {'data_type': 'float'}}}
|
|
|
|
output_data, opc_metrics, _, _, _ = 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()
|
|
|
|
|
|
@patch('laborious.activities.opc.DataFrame')
|
|
def test_write_opc_data_propagates_session_bad_and_reconnect_in_progress(mock_dataframe, opc):
|
|
"""A single server reporting session_bad/reconnect_in_progress must flip the aggregated
|
|
flags passed to process_confidence, even though nothing raised."""
|
|
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'}},
|
|
}
|
|
},
|
|
}
|
|
|
|
opc.manage_output_tags = MagicMock(
|
|
return_value=(False, {'tag1': 0.1, 'tag2': 0.2}, True, 'BadSessionClosed', True)
|
|
)
|
|
opc.process_confidence = MagicMock(return_value={'data': 'data'})
|
|
|
|
opc.write_opc_data(input_data)
|
|
|
|
opc.process_confidence.assert_called_once_with(
|
|
mock_dataframe.return_value,
|
|
False,
|
|
metadata['metadata'],
|
|
session_bad=True,
|
|
opc_status='BadSessionClosed',
|
|
reconnect_in_progress=True,
|
|
)
|
|
|
|
|
|
@patch('laborious.activities.opc.DataFrame')
|
|
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 = MagicMock(
|
|
return_value=(True, {'tag1': 0.1, 'tag2': 0.2}, False, None, False)
|
|
)
|
|
|
|
opc.process_confidence = MagicMock(return_value={'data': 'data'})
|
|
output_data, opc_metrics = 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,
|
|
)
|
|
|
|
|
|
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
|
|
opc.write_opc_data(input_data)
|
|
|
|
# Assert
|
|
opc.opc_repository['server1'].write_data.assert_not_called()
|
|
|
|
|
|
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'}},
|
|
}
|
|
},
|
|
}
|
|
|
|
# Act
|
|
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
|
|
|
|
|
|
def test_manage_output_tags_merges_error_flags(opc):
|
|
opc._write_tags_from_config = MagicMock(
|
|
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,
|
|
) = 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]
|
|
)
|
|
|
|
|
|
def test_validate_server(opc):
|
|
assert opc.validate_server('server1', metadata) is True
|
|
assert opc.validate_server('server2', metadata) is False
|
|
|
|
|
|
def test_close(opc):
|
|
repo = opc.opc_repository['server1']
|
|
repo.disconnect = MagicMock(return_value=True)
|
|
opc.close()
|
|
repo.disconnect.assert_called_once()
|