d00078d323
Enhance ingestor and manager classes with notification handling - Integrated NotificationHandler into Ingestor, DataManager, IngestorManager, and OpcManager for improved error reporting and monitoring. - Updated methods to send notifications on critical events such as Kafka publishing errors, OPC connection issues, and cycle count warnings. - Refactored related tests to ensure coverage of new notification functionalities and validate integration with existing components. - Improved logging and error handling across the system to enhance traceability and operational insights.
343 lines
11 KiB
Python
343 lines
11 KiB
Python
import json
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, patch
|
|
from pytest import fixture
|
|
|
|
from asyncua.crypto.security_policies import SecurityPolicyBasic256
|
|
import pytest
|
|
from ingestor.managers.opc_manager import OpcManager
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
|
|
tags = {
|
|
'ns=3;i=1001': {
|
|
'aggregation_function': 'LTS',
|
|
'frequency': 1000,
|
|
'max_value': 100,
|
|
'min_value': 0,
|
|
'tag_name': 'Counter'
|
|
},
|
|
'ns=3;i=1003': {
|
|
'aggregation_function': 'AVG',
|
|
'frequency': 1000,
|
|
'max_value': 100,
|
|
'min_value': 0,
|
|
'tag_name': 'Random'
|
|
},
|
|
'ns=3;i=1004': {
|
|
'aggregation_function': 'MDN',
|
|
'frequency': 1000,
|
|
'max_value': 100,
|
|
'min_value': 0,
|
|
'tag_name': 'Sawtooth'
|
|
}
|
|
}
|
|
|
|
|
|
@fixture
|
|
def raw_opc_manager():
|
|
return OpcManager(
|
|
'TestConnector', 'opc.tcp://localhost:4840', MagicMock(),
|
|
MagicMock(), 'opc.tcp://localhost:4840', MagicMock()
|
|
)
|
|
|
|
|
|
@fixture
|
|
def opc_manager(raw_opc_manager):
|
|
raw_opc_manager.client = MagicMock()
|
|
raw_opc_manager.cert_path = 'cert.pem'
|
|
raw_opc_manager.private_key_path = 'private_key.pem'
|
|
raw_opc_manager.server_cert_path = 'server_cert.pem'
|
|
|
|
return raw_opc_manager
|
|
|
|
|
|
@fixture
|
|
def opc_manager_subscribed(opc_manager):
|
|
opc_manager.subscriptions['sub1'] = MagicMock()
|
|
|
|
return opc_manager
|
|
|
|
|
|
def test___str__(opc_manager):
|
|
assert str(opc_manager) == 'OpcManager(name=TestConnector, url=opc.tcp://localhost:4840, server_uri=opc.tcp://localhost:4840)\nnodes={}, subscriptions={}'
|
|
|
|
|
|
def test_set_security_success(opc_manager):
|
|
opc_manager.set_security()
|
|
|
|
assert opc_manager.client.application_uri == opc_manager.server_uri
|
|
|
|
opc_manager.client.set_security.assert_called_once_with(
|
|
SecurityPolicyBasic256,
|
|
certificate=opc_manager.cert_path,
|
|
private_key=opc_manager.private_key_path,
|
|
server_certificate=opc_manager.server_cert_path
|
|
)
|
|
|
|
assert opc_manager.client.secure_channel_timeout == 10000000
|
|
assert opc_manager.client.session_timeout == 10000000
|
|
|
|
|
|
def test_set_security_no_cert(opc_manager):
|
|
opc_manager.cert_path = None
|
|
opc_manager.private_key_path = None
|
|
|
|
try:
|
|
opc_manager.set_security()
|
|
except ValueError as e:
|
|
assert str(
|
|
e) == "Certificate and private key paths must be provided for secure connection."
|
|
else:
|
|
assert False, "ValueError not raised"
|
|
|
|
assert opc_manager.client.set_security.call_count == 0
|
|
|
|
|
|
@patch('ingestor.managers.opc_manager.Client')
|
|
def test_connect_no_security(client, raw_opc_manager):
|
|
raw_opc_manager.set_security = MagicMock()
|
|
|
|
raw_opc_manager.connect()
|
|
|
|
client.assert_called_once_with(raw_opc_manager.url)
|
|
raw_opc_manager.client.connect.assert_called_once()
|
|
raw_opc_manager.set_security.assert_not_called()
|
|
|
|
|
|
@patch('ingestor.managers.opc_manager.Client')
|
|
def test_connect_with_security(client, raw_opc_manager):
|
|
raw_opc_manager.cert_path = 'cert.pem'
|
|
raw_opc_manager.private_key_path = 'private_key.pem'
|
|
raw_opc_manager.server_cert_path = 'server_cert.pem'
|
|
raw_opc_manager.set_security = MagicMock()
|
|
|
|
raw_opc_manager.connect()
|
|
|
|
client.assert_called_once_with(raw_opc_manager.url)
|
|
raw_opc_manager.client.connect.assert_called_once()
|
|
raw_opc_manager.set_security.assert_called_once()
|
|
|
|
|
|
def test_create_subscription_no_client(raw_opc_manager):
|
|
try:
|
|
raw_opc_manager.create_subscription('sub1')
|
|
except ValueError as e:
|
|
assert str(e) == "Client not connected. Call connect first."
|
|
else:
|
|
assert False, "ValueError not raised"
|
|
|
|
|
|
def test_create_subscription_success_has_period(opc_manager):
|
|
opc_manager.create_subscription('sub1', 1000)
|
|
|
|
opc_manager.client.create_subscription.assert_called_once_with(
|
|
1000, opc_manager)
|
|
assert opc_manager.subscriptions['sub1'] is not None
|
|
|
|
|
|
def test_create_subscription_success_no_period(opc_manager):
|
|
opc_manager.create_subscription('sub1', None)
|
|
|
|
opc_manager.client.create_subscription.assert_called_once_with(
|
|
500, opc_manager)
|
|
assert opc_manager.subscriptions['sub1'] is not None
|
|
|
|
|
|
def test_subscribe_no_subscription(opc_manager):
|
|
try:
|
|
opc_manager.subscribe('sub1', tags, 1000)
|
|
except ValueError as e:
|
|
assert str(
|
|
e) == "Subscription not created. Call create_subscription first."
|
|
else:
|
|
assert False, "ValueError not raised"
|
|
|
|
|
|
def test_subscribe_success(opc_manager_subscribed):
|
|
opc_manager_subscribed.nodes = {
|
|
'ns=3;i=1001': 'data'
|
|
}
|
|
|
|
opc_manager_subscribed.subscribe('sub1', tags, 1000)
|
|
|
|
assert opc_manager_subscribed.nodes == tags
|
|
assert opc_manager_subscribed.addr_nodes == [
|
|
opc_manager_subscribed.client.get_node(n) for n in tags if n != 'ns=3;i=1001']
|
|
|
|
|
|
def test_unsubscribe_no_subscription(opc_manager):
|
|
opc_manager.unsubscribe('sub1')
|
|
|
|
opc_manager.logger.warning.assert_called_once_with(
|
|
"Subscription 'sub1' not found. Cannot unsubscribe.")
|
|
assert opc_manager.subscriptions.get('sub1') is None
|
|
|
|
|
|
def test_unsubscribe_success(opc_manager_subscribed):
|
|
opc_manager_subscribed.unsubscribe('sub1')
|
|
|
|
opc_manager_subscribed.subscriptions.get('sub1') is None
|
|
|
|
|
|
def test_disconnect_success(opc_manager_subscribed):
|
|
opc_manager_subscribed.client = MagicMock()
|
|
|
|
opc_manager_subscribed.disconnect()
|
|
|
|
opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once()
|
|
assert opc_manager_subscribed.client is None
|
|
|
|
|
|
def test_disconnect_error(opc_manager_subscribed):
|
|
opc_manager_subscribed.client = MagicMock()
|
|
opc_manager_subscribed.subscriptions['sub1'] = MagicMock(
|
|
delete=MagicMock(side_effect=Exception("Test error"))
|
|
)
|
|
|
|
opc_manager_subscribed.disconnect()
|
|
|
|
opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once()
|
|
opc_manager_subscribed.client.disconnect.assert_not_called()
|
|
opc_manager_subscribed.logger.error.assert_called_once_with(
|
|
"Failed to clean up subscription: Test error")
|
|
|
|
|
|
def test_datachange_notification(opc_manager_subscribed):
|
|
data = MagicMock(
|
|
monitored_item=MagicMock(
|
|
Value=MagicMock(
|
|
Value=MagicMock(Value=42),
|
|
SourceTimestamp=datetime.strptime(
|
|
'2021-01-01T00:00:00', '%Y-%m-%dT%H:%M:%S')
|
|
)))
|
|
opc_manager_subscribed.nodes = {
|
|
'ns=3;i=1001': {
|
|
'tag_name': 'Counter',
|
|
'cycle_rule': {
|
|
'cycle_increment': 1.0,
|
|
'cycle_count': 2
|
|
},
|
|
'topics': ['topic1', 'topic2']
|
|
}
|
|
}
|
|
|
|
opc_manager_subscribed.datachange_notification(
|
|
'ns=3;i=1001', None, data)
|
|
|
|
opc_manager_subscribed.data_manager.publish.assert_any_call(
|
|
'topic1', {
|
|
'tag': 'ns=3;i=1001',
|
|
'name': 'Counter',
|
|
'timestamp': '2021-01-01 00:00:00',
|
|
'value': 42
|
|
})
|
|
opc_manager_subscribed.data_manager.publish.assert_any_call(
|
|
'topic2', {
|
|
'tag': 'ns=3;i=1001',
|
|
'name': 'Counter',
|
|
'timestamp': '2021-01-01 00:00:00',
|
|
'value': 42
|
|
})
|
|
assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0
|
|
|
|
|
|
def test_check_cycles_no_notification(opc_manager):
|
|
# Setup: node with cycle_count just below threshold
|
|
opc_manager.nodes = {
|
|
'ns=3;i=1001': {
|
|
'tag_name': 'Counter',
|
|
'cycle_rule': {
|
|
'cycle_increment': 1.0,
|
|
'cycle_count': 3.0
|
|
}
|
|
}
|
|
}
|
|
opc_manager.notification_handler.build_and_send_notification = MagicMock()
|
|
|
|
opc_manager.check_cycles()
|
|
|
|
# After one increment, cycle_count = 4.0, still below threshold
|
|
assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx(
|
|
4.0)
|
|
opc_manager.notification_handler.build_and_send_notification.assert_not_called()
|
|
|
|
|
|
def test_check_cycles_triggers_notification(opc_manager):
|
|
# Setup: node with cycle_count just below threshold, increment will cross threshold
|
|
opc_manager.nodes = {
|
|
'ns=3;i=1001': {
|
|
'tag_name': 'Counter',
|
|
'cycle_rule': {
|
|
'cycle_increment': 2.5,
|
|
'cycle_count': 3.0
|
|
}
|
|
}
|
|
}
|
|
opc_manager.notification_handler.build_and_send_notification = MagicMock()
|
|
|
|
opc_manager.check_cycles()
|
|
|
|
# After increment, cycle_count = 5.5, should trigger notification
|
|
assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx(
|
|
5.5)
|
|
opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
|
|
notification_id='TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED',
|
|
message='5.5 cycles without receive from ns=3;i=1001:Counter',
|
|
block="opc_manager",
|
|
level=NotificationLevel.WARNING
|
|
)
|
|
|
|
|
|
def test_check_opc_listenning_no_notification(opc_manager):
|
|
opc_manager.non_receive_count = 3
|
|
opc_manager.notification_handler.build_and_send_notification = MagicMock()
|
|
|
|
result = opc_manager.check_opc_listenning()
|
|
|
|
assert opc_manager.non_receive_count == 4
|
|
opc_manager.notification_handler.build_and_send_notification.assert_not_called()
|
|
assert result is False
|
|
|
|
|
|
def test_check_opc_listenning_warning_notification(opc_manager):
|
|
opc_manager.non_receive_count = 4
|
|
opc_manager.notification_handler.build_and_send_notification = MagicMock()
|
|
|
|
result = opc_manager.check_opc_listenning()
|
|
|
|
assert opc_manager.non_receive_count == 5
|
|
opc_manager.notification_handler.build_and_send_notification.assert_called_once_with(
|
|
notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}',
|
|
message=f'5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.ERROR
|
|
)
|
|
assert result is False
|
|
|
|
|
|
def test_check_opc_listenning_error_notification_and_retry(opc_manager):
|
|
opc_manager.non_receive_count = 14
|
|
opc_manager.notification_handler.build_and_send_notification = MagicMock()
|
|
|
|
result = opc_manager.check_opc_listenning()
|
|
|
|
assert opc_manager.non_receive_count == 15
|
|
# Should be called twice: once for 5, once for 15
|
|
assert opc_manager.notification_handler.build_and_send_notification.call_count == 2
|
|
calls = opc_manager.notification_handler.build_and_send_notification.call_args_list
|
|
# First call: 5 cycles warning
|
|
assert calls[0].kwargs == dict(
|
|
notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}',
|
|
message=f'15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.ERROR
|
|
)
|
|
# Second call: 15 cycles retry
|
|
assert calls[1].kwargs == dict(
|
|
notification_id=f'OPC_CONNECTION_RETRY__{opc_manager.name}',
|
|
message=f'Retrying to connect to server {opc_manager.name}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.ERROR
|
|
)
|
|
assert result is True
|