sientia-dataops-laborious-legacy/tests/laborious/activities/test_opc.py

from unittest.mock import patch, MagicMock, ANY, call
from pytest import fixture, mark
from laborious.activities.opc import NotificationLevel

from laborious.activities.opc import OPC


@patch("laborious.activities.opc.OpcRepository")
def test___init__(mock_opc_repository):
    mock_logger = MagicMock()
    server1 = MagicMock()
    server2 = MagicMock()
    mock_opc_repository.side_effect = [server1, server2]
    mock_notification_handler = MagicMock()
    servers = {
        'server1': {
            'url': 'http://localhost:8080',
            'server_uri': 'opc.tcp://localhost:4840',
            'cert_path': '',
            'private_key_path': '',
            'server_cert_path': '',
            'reconnection_interval': 60,
        },
        'server2': {
            '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
    )

    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(
            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,
        ),
    ])
    mock_opc_repository.assert_has_calls([
        call(
            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,
        )
    ])

    server1.connect.assert_called_once()
    server2.connect.assert_called_once()


@fixture
@patch("laborious.activities.opc.OpcRepository")
def opc(_mock_opc_repository):
    servers = {
        'server1': {
            'url': 'http://localhost:8080',
            'server_uri': 'opc.tcp://localhost:4840',
            'cert_path': '',
            'private_key_path': '',
            'server_cert_path': '',
            'reconnection_interval': 60,
        }
    }
    return OPC(
        opc_servers=servers,
        logger=MagicMock(),
        notification_handler=MagicMock()
    )


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.write_data(server='server1', tag=tag, data=data,
                   data_type=data_type, tag_type='prediction')
    opc.opc_repository['server1'].write_data.assert_called_once_with(
        tag, data, data_type)


def test_write_data_exception(opc):
    opc.opc_repository['server1'].write_data.side_effect = Exception(
        "Test error")

    try:
        opc.write_data(server='server1', tag='tag1', data=50,
                       data_type='int', tag_type='prediction')

    except Exception:
        opc.notification_handler.build_and_send_notification.assert_called_once_with(
            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:
        assert False, "Expected an exception to be raised"


@mark.asyncio
async def test_write_opc_data_success(opc):
    # Arrange
    input_data = {
        '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_data = MagicMock()
    await opc.write_opc_data(input_data)

    # Assert
    opc.write_data.assert_has_calls([
        call(
            server='server1',
            tag='tag1',
            data=0.75,
            data_type='float',
            tag_type='prediction'
        )])
    opc.write_data.assert_has_calls([
        call(
            server='server1',
            tag='tag2',
            data=0.95,
            data_type='float',
            tag_type='confidence'
        )
    ])
    assert opc.write_data.call_count == 2


@mark.asyncio
async def test_write_opc_data_empty_config(opc):
    # Arrange
    input_data = {
        'data': {
            'prediction': [0.75],
            'prediction_confidence': [0.95]
        },
        'opc_servers': ['server1'],
        'opc_output_config': {
            'prediction_tags': {},
            'confidence_tags': {}
        }
    }

    # Act
    await opc.write_opc_data(input_data)

    # Assert
    opc.opc_repository['server1'].write_data.assert_not_called()


def test_shutdown(opc):
    opc.shutdown()
    opc.opc_repository['server1'].disconnect.assert_called_once()