SIENTIAPDE-1314

Enhance OPC Testing for Output Management

- Refactored the test_manage_output_tags_failed function to use parameterization for improved test coverage of different side effects.
- Added a new test_manage_output_tags_do_nothing to verify behavior when an invalid key is provided in the configuration, ensuring that no write operations are performed in this case.
This commit is contained in:
vitor-aignosi
2025-10-27 08:17:57 -03:00
parent 1b93eb19cb
commit 56122efefa

View File

@@ -299,8 +299,9 @@ async def test_manage_output_tags_success(opc):
@mark.asyncio
async def test_manage_output_tags_failed(opc):
opc.write_data = AsyncMock(side_effect=[0.1, None])
@mark.parametrize('side_effect', [[0.1, None], [None, 0.2]])
async def test_manage_output_tags_failed(opc, side_effect):
opc.write_data = AsyncMock(side_effect=side_effect)
data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
config = {
'prediction_tags': {'tag1': {'data_type': 'float'}},
@@ -313,7 +314,7 @@ async def test_manage_output_tags_failed(opc):
metadata=metadata['metadata'],
)
assert output_data is False
assert opc_metrics == {'tag1': 0.1, 'tag2': None}
assert opc_metrics == {'tag1': side_effect[0], 'tag2': side_effect[1]}
opc.write_data.assert_has_calls(
[
call(
@@ -335,6 +336,19 @@ async def test_manage_output_tags_failed(opc):
]
)
@mark.asyncio
async def test_manage_output_tags_do_nothing(opc):
opc.write_data = AsyncMock(return_value=0.1)
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_data.assert_not_called()
@mark.asyncio
@patch('laborious.activities.opc.DataFrame')