057c1fd1ce
The main loop snapshots managed_tags, then two steps remove a slot from
managed_tags before update_ingestor_manager compares the snapshot with the
current state and calls unsubscribe_slot for every slot that disappeared:
- manage_leases drops an extra lease ("Extra slots available"): it already
unsubscribed the slot, so the second call is redundant;
- update_slot_config finds the slot gone from Redis (deleted by the
orchestrator): nothing unsubscribed it yet.
unsubscribe_slot looked the slot up in managed_tags[slot] to find its servers,
so both cases raised KeyError, which stopped the main loop and restarted the
container (seen on dev on 2026-10-06 18:30 during a rollout, ~40 s without
reads). In the second case the subscription also stayed open on the OPC server.
The same code is on main, so production has it too.
unsubscribe_slot now asks the readers themselves: every reader (ServerReader or
the legacy OpcManager, both keep `subscriptions` by slot) holding a
subscription for the slot unsubscribes it. A slot no reader holds is a no-op.
This also covers a resubscription whose new config dropped a server: the old
server's subscription for that slot is closed too.
Tests: unit tests for the new lookup and the double call, plus two regressions
at the Ingestor level with the real unsubscribe_slot (dropped extra lease;
slot removed from Redis). Without the fix they fail with the same KeyError at
ingestor_manager.py:558. 280 unit tests.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
961 lines
35 KiB
Python
961 lines
35 KiB
Python
import gc
|
|
from unittest.mock import AsyncMock, MagicMock, call, patch
|
|
|
|
from pytest import fixture, mark
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
|
|
from ingestor.config import GatewayConfig
|
|
from ingestor.connection.pool import OpcConnectionPool
|
|
from ingestor.io.slots import parse_slot
|
|
from ingestor.managers.ingestor_manager import IngestorManager
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': 'test_model',
|
|
'model_name': 'test_model',
|
|
'workflow_name': 'test_workflow',
|
|
'schema_name': 'test_schedule',
|
|
},
|
|
}
|
|
|
|
|
|
def raw_slot_v1(tag_name='Counter', address='ns=2;i=2'):
|
|
"""A slot as the current (v1) Orchestrator writes it to Redis."""
|
|
return {
|
|
'server1': {
|
|
'name': 'server1',
|
|
'server_id': '1',
|
|
'url': 'opc.tcp://localhost:4840',
|
|
'server_uri': 'urn:sientia:opc-ingestor',
|
|
'cert_path': '/app/certs/opc-ingestor-cert.pem',
|
|
'subscription_period_ms': 500.0,
|
|
'tags': {address: {'tag_name': tag_name, 'frequency': '1000', 'topics': ['raw_a']}},
|
|
}
|
|
}
|
|
|
|
|
|
@fixture(autouse=True)
|
|
def _collect_garbage_after_test():
|
|
# ponytail: IngestorManager.__del__ calls asyncio.run(self.shutdown()); if a
|
|
# cyclic-referenced instance from a previous test is swept mid-test by the
|
|
# generational GC, it can double-call whatever asyncio mock happens to be
|
|
# patched at that moment (e.g. test___del__). Force a deterministic sweep
|
|
# after each test so finalizers run in their own test's scope. Upgrade path:
|
|
# only needed if IngestorManager stops relying on __del__ for cleanup.
|
|
yield
|
|
gc.collect()
|
|
|
|
|
|
@fixture
|
|
@patch('ingestor.managers.ingestor_manager.DataManager')
|
|
@patch('ingestor.managers.ingestor_manager.ResourceManager')
|
|
def ingestor_manager(data_manager_mock, resource_manager_mock):
|
|
ingestor = IngestorManager(
|
|
redis_data={'host': 'localhost', 'port': 6379},
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
poll_interval=5,
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metadata=metadata['metadata'],
|
|
metrics_controller=MagicMock(),
|
|
)
|
|
|
|
ingestor.send_notification = MagicMock()
|
|
ingestor.send_notification_async = AsyncMock()
|
|
ingestor.emit_metric = AsyncMock()
|
|
|
|
return ingestor
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.ServerReader')
|
|
@patch('ingestor.managers.ingestor_manager.DataManager')
|
|
@patch('ingestor.managers.ingestor_manager.ResourceManager')
|
|
@patch('ingestor.managers.ingestor_manager.NotificationHandler')
|
|
def test___init__(
|
|
notification_handler_mock, resource_manager_mock, data_manager_mock, server_reader_mock
|
|
):
|
|
ingestor = IngestorManager(
|
|
redis_data={'host': 'localhost', 'port': 6379},
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
poll_interval=5,
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metadata=metadata['metadata'],
|
|
metrics_controller=MagicMock(),
|
|
)
|
|
|
|
server_reader_mock.assert_not_called()
|
|
data_manager_mock.assert_called_once_with(
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
metadata=metadata['metadata'],
|
|
logger=ingestor.logger,
|
|
notification_handler=ingestor.notification_handler,
|
|
metrics_controller=ingestor.metrics_controller,
|
|
)
|
|
resource_manager_mock.assert_called_once_with(
|
|
host='localhost',
|
|
port=6379,
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
metadata=metadata['metadata'],
|
|
logger=ingestor.logger,
|
|
notification_handler=ingestor.notification_handler,
|
|
username=None,
|
|
password=None,
|
|
metrics_controller=ingestor.metrics_controller,
|
|
)
|
|
assert ingestor.poll_interval == 5
|
|
assert ingestor.managed_tags == {}
|
|
assert ingestor.opc_servers == {}
|
|
assert ingestor.opc_managers == {}
|
|
assert ingestor.data_manager == data_manager_mock.return_value
|
|
assert ingestor.resource_manager == resource_manager_mock.return_value
|
|
|
|
# Read path (SIENTIAPDE-1974): pool + read sink built here, nothing connected yet.
|
|
assert isinstance(ingestor.pool, OpcConnectionPool)
|
|
assert ingestor.pool.cert_provider is ingestor.cert_provider
|
|
assert ingestor.cert_provider.mongo is data_manager_mock.return_value.mongo_repository
|
|
assert ingestor.sink.data_manager is data_manager_mock.return_value
|
|
assert ingestor.sink.postgres is None
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.DataManager')
|
|
@patch('ingestor.managers.ingestor_manager.ResourceManager')
|
|
def test___init___cert_provider_uses_gateway_config(resource_manager_mock, data_manager_mock):
|
|
ingestor = IngestorManager(
|
|
redis_data={'host': 'localhost', 'port': 6379},
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
poll_interval=5,
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metadata=metadata['metadata'],
|
|
metrics_controller=MagicMock(),
|
|
gateway_config=GatewayConfig(cert_bucket='my-bucket', cert_cache_dir='certs-cache'),
|
|
)
|
|
|
|
assert ingestor.cert_provider.bucket == 'my-bucket'
|
|
assert ingestor.cert_provider.cache_dir == 'certs-cache'
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('ingestor.managers.ingestor_manager.ServerReader')
|
|
async def test_initialize_opc_from_config(server_reader, ingestor_manager):
|
|
server_config = {
|
|
'name': 'server1',
|
|
'server_id': '1',
|
|
'url': 'opc.tcp://localhost:4840',
|
|
'server_uri': 'urn:sientia:opc-ingestor',
|
|
'use_security': True,
|
|
'subscription_period_ms': 1000,
|
|
}
|
|
|
|
server_reader.return_value = MagicMock(connect=AsyncMock())
|
|
result = await ingestor_manager.initialize_opc_from_config(server_config)
|
|
|
|
server_reader.assert_called_once_with(
|
|
name=server_config['name'],
|
|
server_config=server_config,
|
|
pool=ingestor_manager.pool,
|
|
sink=ingestor_manager.sink,
|
|
logger=ingestor_manager.logger,
|
|
notification_handler=ingestor_manager.notification_handler,
|
|
metrics_controller=ingestor_manager.metrics_controller,
|
|
metadata=metadata['metadata'],
|
|
)
|
|
|
|
assert result == server_reader.return_value
|
|
assert result.config == server_config
|
|
result.connect.assert_awaited_once()
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('ingestor.managers.ingestor_manager.ServerReader')
|
|
@patch('ingestor.managers.ingestor_manager.traceback')
|
|
async def test_initialize_opc_from_config_exception(
|
|
traceback_mock, server_reader, ingestor_manager
|
|
):
|
|
server_config = {
|
|
'name': 'server1',
|
|
'server_id': '1',
|
|
'url': 'opc.tcp://localhost:4840',
|
|
'server_uri': 'urn:sientia:opc-ingestor',
|
|
'use_security': True,
|
|
'subscription_period_ms': 1000,
|
|
}
|
|
|
|
ingestor_manager.logger.error = MagicMock()
|
|
# e.g. the server is unreachable or the certificate can't be fetched from MinIO
|
|
server_reader.return_value = MagicMock(
|
|
connect=AsyncMock(side_effect=Exception('Initialization error'))
|
|
)
|
|
|
|
result = await ingestor_manager.initialize_opc_from_config(server_config)
|
|
|
|
assert result is None
|
|
|
|
traceback_mock.format_exc.assert_called_once()
|
|
ingestor_manager.send_notification.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
|
|
message='Error initializing OPC manager: Initialization error',
|
|
block='opc_manager',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=traceback_mock.format_exc.return_value,
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_shutdown(ingestor_manager):
|
|
ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
|
|
|
|
ingestor_manager.data_manager.shutdown = MagicMock()
|
|
ingestor_manager.pool = MagicMock(close_all=AsyncMock())
|
|
|
|
await ingestor_manager.shutdown()
|
|
|
|
ingestor_manager.opc_managers['server1'].shutdown.assert_called_once()
|
|
ingestor_manager.opc_managers['server2'].shutdown.assert_called_once()
|
|
# The pool is owned by the read path: closed after the readers unsubscribe.
|
|
ingestor_manager.pool.close_all.assert_awaited_once()
|
|
|
|
ingestor_manager.data_manager.shutdown.assert_called_once()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_remove_server(ingestor_manager):
|
|
server1 = AsyncMock()
|
|
server2 = AsyncMock()
|
|
ingestor_manager.opc_managers = {'server1': server1, 'server2': server2}
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
|
|
}
|
|
await ingestor_manager.remove_server('server1')
|
|
server1.shutdown.assert_called_once()
|
|
server2.shutdown.assert_not_called()
|
|
assert 'server1' not in ingestor_manager.opc_managers
|
|
assert 'server2' in ingestor_manager.opc_managers
|
|
assert ingestor_manager.managed_tags == {'slot1': {'server2': {'config': 'config2'}}}
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_remove_server_not_found(ingestor_manager):
|
|
server1 = AsyncMock()
|
|
server2 = AsyncMock()
|
|
ingestor_manager.opc_managers = {'server1': server1, 'server2': server2}
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
|
|
}
|
|
await ingestor_manager.remove_server('server3')
|
|
server1.shutdown.assert_not_called()
|
|
server2.shutdown.assert_not_called()
|
|
assert 'server1' in ingestor_manager.opc_managers
|
|
assert 'server2' in ingestor_manager.opc_managers
|
|
assert ingestor_manager.managed_tags == {
|
|
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
|
|
}
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
async def test_update_opc_servers(metrics, ingestor_manager):
|
|
manager1 = MagicMock(config={'config': 'config1'})
|
|
manager2 = MagicMock(config={'config': 'config2'})
|
|
manager3 = MagicMock(config={'config': 'config3'})
|
|
|
|
async def mock_initialize_from_config(config):
|
|
if config == {'config': 'config1'}:
|
|
return manager1
|
|
elif config == {'config': 'config2'}:
|
|
return manager2
|
|
elif config == {'config': 'config3'}:
|
|
return manager3
|
|
else:
|
|
return None
|
|
|
|
ingestor_manager.initialize_opc_from_config = AsyncMock(side_effect=mock_initialize_from_config)
|
|
ingestor_manager.remove_server = AsyncMock()
|
|
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': {
|
|
'server1': {'config': 'config1'},
|
|
'server2': {'config': 'config2'},
|
|
'server5': {'config': 'config5'},
|
|
},
|
|
'slot2': {
|
|
'server3': {'config': 'config3'},
|
|
'server1': {'config': 'config1'},
|
|
},
|
|
}
|
|
|
|
mock = AsyncMock(config={'config': 'old_config2'})
|
|
ingestor_manager.opc_managers['server3'] = AsyncMock(config={'config': 'config3'})
|
|
ingestor_manager.opc_managers['server2'] = mock
|
|
ingestor_manager.opc_managers['server4'] = AsyncMock()
|
|
|
|
await ingestor_manager.update_opc_servers()
|
|
|
|
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config1'})
|
|
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config2'})
|
|
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config5'})
|
|
|
|
assert ingestor_manager.initialize_opc_from_config.call_count == 3
|
|
|
|
assert ingestor_manager.opc_managers['server1'].config == {'config': 'config1'}
|
|
assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'}
|
|
assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'}
|
|
|
|
# server4 left this pod's slots: unsubscribe only, the connection stays for writes
|
|
ingestor_manager.remove_server.assert_any_call('server4', evict=False)
|
|
# server5 failed to initialize: removed with the default (evict)
|
|
ingestor_manager.remove_server.assert_any_call('server5')
|
|
assert ingestor_manager.remove_server.call_count == 2
|
|
|
|
assert ingestor_manager.opc_managers['server2'] != mock
|
|
|
|
ingestor_manager.emit_metric.assert_called_once_with(
|
|
metric_object=metrics.OPC_MANAGERS_ACTIVE,
|
|
method='set',
|
|
value=len(ingestor_manager.opc_managers),
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
async def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager):
|
|
# Setup mock OPC managers
|
|
opc_manager1 = MagicMock()
|
|
opc_manager1.check_cycles = AsyncMock(return_value=None)
|
|
opc_manager1.check_opc_listenning = AsyncMock(return_value=False)
|
|
opc_manager1.config = {'config': 'config1'}
|
|
|
|
opc_manager2 = MagicMock()
|
|
opc_manager2.check_cycles = AsyncMock(return_value=None)
|
|
opc_manager2.check_opc_listenning = AsyncMock(return_value=False)
|
|
opc_manager2.config = {'config': 'config2'}
|
|
|
|
ingestor_manager.opc_managers = {'server1': opc_manager1, 'server2': opc_manager2}
|
|
|
|
# Mock the initialize_opc_from_config method
|
|
ingestor_manager.initialize_opc_from_config = MagicMock()
|
|
|
|
# Call the method
|
|
await ingestor_manager.check_opc_servers_integrity()
|
|
|
|
# Verify that check_cycles and check_opc_listenning were called for each server
|
|
opc_manager1.check_cycles.assert_called_once()
|
|
opc_manager1.check_opc_listenning.assert_called_once()
|
|
opc_manager2.check_cycles.assert_called_once()
|
|
opc_manager2.check_opc_listenning.assert_called_once()
|
|
|
|
# Verify that no reinitialization was needed
|
|
ingestor_manager.initialize_opc_from_config.assert_not_called()
|
|
|
|
ingestor_manager.emit_metric.assert_called_once_with(
|
|
metric_object=metrics.OPC_MANAGERS_ACTIVE,
|
|
method='set',
|
|
value=len(ingestor_manager.opc_managers),
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_check_opc_servers_integrity_server_lost(ingestor_manager):
|
|
# Setup mock OPC manager that will be lost
|
|
opc_manager = AsyncMock()
|
|
opc_manager.check_cycles.return_value = None
|
|
opc_manager.check_opc_listenning.return_value = True # Server is lost
|
|
opc_manager.config = {'config': 'config1'}
|
|
|
|
ingestor_manager.opc_managers = {'server1': opc_manager}
|
|
ingestor_manager.managed_tags = {'slot1': MagicMock()}
|
|
|
|
# Call the method
|
|
await ingestor_manager.check_opc_servers_integrity()
|
|
|
|
assert 'server1' not in ingestor_manager.opc_managers
|
|
ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server1', None)
|
|
|
|
|
|
def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager):
|
|
# Setup mock OPC manager that will be lost
|
|
opc_manager = MagicMock()
|
|
opc_manager.check_cycles.return_value = None
|
|
opc_manager.check_opc_listenning.return_value = True # Server is lost
|
|
opc_manager.config = {'config': 'config1'}
|
|
|
|
ingestor_manager.opc_managers = {'server1': opc_manager}
|
|
|
|
# Setup managed tags
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': {'server1': {'config': 'config1', 'tags': {'tag1': 'value1'}}}
|
|
}
|
|
|
|
# Mock the initialize_opc_from_config method to return a new manager
|
|
new_manager = MagicMock()
|
|
ingestor_manager.initialize_opc_from_config = MagicMock(return_value=new_manager)
|
|
|
|
# Call the method
|
|
ingestor_manager.check_opc_servers_integrity()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_declare_active(ingestor_manager):
|
|
ingestor_manager.resource_manager.ingestor_heartbeat = AsyncMock()
|
|
await ingestor_manager.declare_active()
|
|
ingestor_manager.resource_manager.ingestor_heartbeat.assert_called_once()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_active_ingestors(ingestor_manager):
|
|
ingestor_manager.resource_manager.get_all_ingestors = AsyncMock()
|
|
await ingestor_manager.get_active_ingestors()
|
|
ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_active_ingestors_empty(ingestor_manager):
|
|
ingestor_manager.resource_manager.get_all_ingestors = AsyncMock(return_value=None)
|
|
result = await ingestor_manager.get_active_ingestors()
|
|
assert result == []
|
|
ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
@mark.asyncio
|
|
async def test_get_number_of_leases_success(metrics, ingestor_manager):
|
|
ingestor_manager.resource_manager.get_all_leases = AsyncMock(return_value=['lease1', 'lease2'])
|
|
result = await ingestor_manager.get_number_of_leases()
|
|
assert result == 2
|
|
|
|
ingestor_manager.emit_metric.assert_called_once_with(
|
|
metric_object=metrics.LEASES_TOTAL,
|
|
method='set',
|
|
value=2,
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
)
|
|
ingestor_manager.resource_manager.get_all_leases.assert_called_once()
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
@mark.asyncio
|
|
async def test_get_number_of_slots_success(metrics, ingestor_manager):
|
|
ingestor_manager.resource_manager.get_all_slots = AsyncMock(return_value=['slot1', 'slot2'])
|
|
result = await ingestor_manager.get_number_of_slots()
|
|
assert result == 2
|
|
ingestor_manager.resource_manager.get_all_slots.assert_called_once()
|
|
ingestor_manager.emit_metric.assert_called_once_with(
|
|
metric_object=metrics.SLOTS_TOTAL,
|
|
method='set',
|
|
value=2,
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
)
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
@mark.asyncio
|
|
async def test_get_number_of_slots_empty(metrics, ingestor_manager):
|
|
ingestor_manager.resource_manager.get_all_slots = AsyncMock(return_value=None)
|
|
result = await ingestor_manager.get_number_of_slots()
|
|
assert result == 0
|
|
ingestor_manager.resource_manager.get_all_slots.assert_called_once()
|
|
ingestor_manager.emit_metric.assert_called_once_with(
|
|
metric_object=metrics.SLOTS_TOTAL,
|
|
method='set',
|
|
value=0,
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_slot_leases_1_success(ingestor_manager):
|
|
ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True)
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value=raw_slot_v1())
|
|
|
|
ingestor_manager.number_of_slots = 1
|
|
result = await ingestor_manager.get_slot_leases()
|
|
|
|
# Stored already normalized to slot v2 (v1 converted by parse_slot).
|
|
assert result == {'1': parse_slot(raw_slot_v1())}
|
|
assert ingestor_manager.managed_tags == result
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_slot_leases_2_success(ingestor_manager):
|
|
ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True])
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
|
|
side_effect=[raw_slot_v1('Counter'), raw_slot_v1('Ramp', 'ns=2;i=3')]
|
|
)
|
|
|
|
ingestor_manager.number_of_slots = 2
|
|
result = await ingestor_manager.get_slot_leases(max_slots=2)
|
|
|
|
assert result == {
|
|
'1': parse_slot(raw_slot_v1('Counter')),
|
|
'2': parse_slot(raw_slot_v1('Ramp', 'ns=2;i=3')),
|
|
}
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_slot_leases_skips_malformed_slot(ingestor_manager):
|
|
ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True)
|
|
# tag without names/topics/frequency
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
|
|
return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}}
|
|
)
|
|
|
|
ingestor_manager.number_of_slots = 1
|
|
result = await ingestor_manager.get_slot_leases()
|
|
|
|
assert result == {}
|
|
_, kwargs = ingestor_manager.send_notification.call_args
|
|
assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1'
|
|
assert kwargs['level'] == NotificationLevel.ERROR
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_slot_leases_2_1_none(ingestor_manager):
|
|
ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True])
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
|
|
side_effect=[None, {'tags': ['tag1']}]
|
|
)
|
|
|
|
ingestor_manager.number_of_slots = 1
|
|
result = await ingestor_manager.get_slot_leases(max_slots=1)
|
|
|
|
assert result == {}
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_get_slot_leases_1_failure(ingestor_manager):
|
|
ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=False)
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value={'tags': ['tag1']})
|
|
|
|
result = await ingestor_manager.get_slot_leases()
|
|
ingestor_manager.resource_manager.get_tag_slot.assert_not_called()
|
|
|
|
assert result == {}
|
|
|
|
|
|
def _reader(*slots):
|
|
"""An OPC reader double whose unsubscribe drops the slot, like the real readers."""
|
|
reader = MagicMock()
|
|
reader.subscriptions = dict.fromkeys(slots, MagicMock())
|
|
reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot))
|
|
return reader
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_unsubscribe_slot(ingestor_manager):
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}},
|
|
'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}},
|
|
}
|
|
ingestor_manager.opc_managers = {
|
|
'server1': _reader('slot1', 'slot2'),
|
|
'server2': _reader('slot1'),
|
|
'server3': _reader('slot2'),
|
|
}
|
|
await ingestor_manager.unsubscribe_slot('slot1')
|
|
|
|
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1')
|
|
ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot1')
|
|
ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called()
|
|
assert list(ingestor_manager.opc_managers['server1'].subscriptions) == ['slot2']
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager):
|
|
"""The slot is already gone from managed_tags: the readers still holding it are unsubscribed"""
|
|
ingestor_manager.managed_tags = {'slot1': {'server1': {'tags': 'config1'}}}
|
|
ingestor_manager.opc_managers = {
|
|
'server1': _reader('slot1', 'slot3'),
|
|
'server2': _reader('slot3'),
|
|
}
|
|
|
|
await ingestor_manager.unsubscribe_slot('slot3')
|
|
|
|
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot3')
|
|
ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot3')
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager):
|
|
"""A second unsubscribe of the same slot (lease already dropped) does nothing and never raises"""
|
|
ingestor_manager.managed_tags = {}
|
|
ingestor_manager.opc_managers = {'server1': _reader('slot2')}
|
|
|
|
await ingestor_manager.unsubscribe_slot('slot2')
|
|
await ingestor_manager.unsubscribe_slot('slot2')
|
|
|
|
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot2')
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_update_slot_config(ingestor_manager):
|
|
ingestor_manager.managed_tags = {
|
|
'slot1': parse_slot(raw_slot_v1('Old')),
|
|
'slot2': parse_slot(raw_slot_v1('Same')),
|
|
'slot3': parse_slot(raw_slot_v1('Old')),
|
|
}
|
|
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
|
|
side_effect=[raw_slot_v1('Updated'), raw_slot_v1('Same'), None]
|
|
)
|
|
ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
|
|
|
|
await ingestor_manager.update_slot_config()
|
|
|
|
assert ingestor_manager.managed_tags['slot1'] == parse_slot(raw_slot_v1('Updated'))
|
|
assert ingestor_manager.managed_tags['slot2'] == parse_slot(raw_slot_v1('Same'))
|
|
assert 'slot3' not in ingestor_manager.managed_tags
|
|
|
|
ingestor_manager.resource_manager.renew_tag_lease.assert_any_call('slot1')
|
|
ingestor_manager.resource_manager.renew_tag_lease.assert_any_call('slot3')
|
|
ingestor_manager.resource_manager.renew_tag_lease.assert_any_call('slot2')
|
|
assert ingestor_manager.resource_manager.renew_tag_lease.call_count == 3
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.metrics')
|
|
@mark.asyncio
|
|
async def test_drop_slot_leases(metrics, ingestor_manager):
|
|
ingestor_manager.resource_manager.drop_tag_lease = AsyncMock()
|
|
await ingestor_manager.drop_slot_leases(['1', '2'])
|
|
|
|
ingestor_manager.resource_manager.drop_tag_lease.assert_any_call('1')
|
|
ingestor_manager.resource_manager.drop_tag_lease.assert_any_call('2')
|
|
|
|
ingestor_manager.emit_metric.assert_has_calls(
|
|
[
|
|
call(
|
|
metric_object=metrics.SLOTS_RELEASED,
|
|
method='inc',
|
|
value=1,
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
),
|
|
call(
|
|
metric_object=metrics.SLOTS_RELEASED,
|
|
method='inc',
|
|
value=1,
|
|
tags={'pod_id': ingestor_manager.pod_id},
|
|
),
|
|
]
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_server_no_server(ingestor_manager):
|
|
ingestor_manager.opc_managers = {'server1': MagicMock(), 'server2': MagicMock()}
|
|
server_config = {'tags': 'config1'}
|
|
|
|
result = await ingestor_manager.manage_server('slot1', 'server3', server_config, server_config)
|
|
|
|
assert result == 1
|
|
ingestor_manager.opc_managers['server1'].create_subscription.assert_not_called()
|
|
ingestor_manager.opc_managers['server1'].subscribe.assert_not_called()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_server_create_subscription_failure(ingestor_manager):
|
|
ingestor_manager.opc_managers = {'server1': MagicMock(), 'server2': MagicMock()}
|
|
ingestor_manager.subscriptions = {'server1': MagicMock()}
|
|
server_config = {'tags': 'config1'}
|
|
|
|
ingestor_manager.opc_managers['server1'].create_subscription.side_effect = Exception(
|
|
'Subscription error'
|
|
)
|
|
|
|
result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config)
|
|
|
|
assert result == 2
|
|
ingestor_manager.opc_managers['server1'].create_subscription.assert_called_once_with('slot1')
|
|
ingestor_manager.opc_managers['server1'].subscribe.assert_not_called()
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_manage_server(ingestor_manager):
|
|
ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
|
|
ingestor_manager.subscriptions = {'server1': AsyncMock()}
|
|
server_config = {'tags': 'config1'}
|
|
|
|
result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config)
|
|
|
|
assert result == 0
|
|
ingestor_manager.opc_managers['server1'].create_subscription.assert_called_once_with('slot1')
|
|
ingestor_manager.opc_managers['server1'].subscribe.assert_called_once_with(
|
|
'slot1', 'config1', ingestor_manager.poll_interval
|
|
)
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.traceback')
|
|
@mark.asyncio
|
|
async def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager):
|
|
ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
|
|
ingestor_manager.subscriptions = {'server1': AsyncMock()}
|
|
server_config = {'tags': 'config1'}
|
|
|
|
ingestor_manager.opc_managers['server1'].subscribe.side_effect = Exception('Subscription error')
|
|
|
|
result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config)
|
|
|
|
assert result == 2
|
|
ingestor_manager.opc_managers['server1'].create_subscription.assert_called_once_with('slot1')
|
|
ingestor_manager.opc_managers['server1'].subscribe.assert_called_once_with(
|
|
'slot1', 'config1', ingestor_manager.poll_interval
|
|
)
|
|
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1')
|
|
|
|
traceback_mock.format_exc.assert_called_once()
|
|
|
|
ingestor_manager.send_notification_async.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1',
|
|
message="Failed to subscribe to tags from slot1:server1\n{'tags': 'config1'}: Subscription error",
|
|
block='opc_manager',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=traceback_mock.format_exc.return_value,
|
|
)
|
|
|
|
ingestor_manager.logger.warning.assert_any_call(
|
|
'Removing subscription from server server1 for slot slot1'
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_subscribe_to_tags(ingestor_manager):
|
|
ingestor_manager.manage_server = AsyncMock(side_effect=[0, 1, 2])
|
|
ingestor_manager.managed_tags = {'slot1': MagicMock(), 'slot2': MagicMock()}
|
|
|
|
ingestor_manager.opc_managers = {
|
|
'server1': AsyncMock(),
|
|
'server2': AsyncMock(),
|
|
'server3': AsyncMock(),
|
|
}
|
|
ingestor_manager.subscriptions = {'server1': AsyncMock()}
|
|
tags = {
|
|
'slot1': {
|
|
'server1': {'tags': 'config1'},
|
|
'server2': {'tags': 'config2'},
|
|
'server3': {'tags': 'config3'},
|
|
}
|
|
}
|
|
|
|
await ingestor_manager.subscribe_to_tags(tags)
|
|
|
|
ingestor_manager.manage_server.assert_any_call('slot1', 'server1', {'tags': 'config1'}, tags)
|
|
ingestor_manager.manage_server.assert_any_call('slot1', 'server2', {'tags': 'config2'}, tags)
|
|
ingestor_manager.manage_server.assert_any_call('slot1', 'server3', {'tags': 'config3'}, tags)
|
|
|
|
assert ingestor_manager.manage_server.call_count == 3
|
|
|
|
ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server3', None)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_update_slot_config_keeps_last_good_config_when_malformed(ingestor_manager):
|
|
good = parse_slot(raw_slot_v1('Counter'))
|
|
ingestor_manager.managed_tags = {'slot1': good}
|
|
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
|
|
return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}}
|
|
)
|
|
ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
|
|
|
|
await ingestor_manager.update_slot_config()
|
|
|
|
# Still reading with the last good config; the bad one is notified, not applied.
|
|
assert ingestor_manager.managed_tags == {'slot1': good}
|
|
_, kwargs = ingestor_manager.send_notification.call_args
|
|
assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_slot1'
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_remove_server_forwards_evict(ingestor_manager):
|
|
reader = AsyncMock()
|
|
ingestor_manager.opc_managers = {'server1': reader}
|
|
ingestor_manager.managed_tags = {'slot1': {'server1': {}}}
|
|
|
|
await ingestor_manager.remove_server('server1', evict=False)
|
|
|
|
reader.shutdown.assert_awaited_once_with(evict=False)
|
|
assert 'server1' not in ingestor_manager.opc_managers
|
|
assert ingestor_manager.managed_tags == {'slot1': {}}
|
|
|
|
|
|
@patch('ingestor.managers.ingestor_manager.DataManager')
|
|
@patch('ingestor.managers.ingestor_manager.ResourceManager')
|
|
def test___init___ignores_malformed_write_env(
|
|
resource_manager_mock, data_manager_mock, monkeypatch
|
|
):
|
|
"""A malformed write/gRPC-only env var must not stop the read path (review #1)."""
|
|
monkeypatch.setenv('GRPC_PORT', '')
|
|
monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0,5')
|
|
monkeypatch.setenv('CERT_BUCKET', 'streamlit-connectors')
|
|
|
|
ingestor = IngestorManager(
|
|
redis_data={'host': 'localhost', 'port': 6379},
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
poll_interval=5,
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metadata=metadata['metadata'],
|
|
metrics_controller=MagicMock(),
|
|
)
|
|
|
|
assert ingestor.cert_provider.bucket == 'streamlit-connectors'
|
|
|
|
|
|
def test_parse_slot_config_warns_once_about_server_cert_path(ingestor_manager):
|
|
raw = raw_slot_v1()
|
|
raw['server1']['server_cert_path'] = '/app/certs/kepserver.der'
|
|
|
|
ingestor_manager.parse_slot_config('1', raw)
|
|
ingestor_manager.parse_slot_config('1', raw) # every loop refreshes the slot
|
|
|
|
warnings = [
|
|
c for c in ingestor_manager.logger.warning.call_args_list if 'server_cert_path' in str(c)
|
|
]
|
|
assert len(warnings) == 1
|
|
|
|
|
|
# --- READ_ENGINE=legacy: plan B with the untouched rc3 OpcManager -------------------
|
|
|
|
|
|
def make_manager(read_engine=None):
|
|
with (
|
|
patch('ingestor.managers.ingestor_manager.DataManager'),
|
|
patch('ingestor.managers.ingestor_manager.ResourceManager'),
|
|
):
|
|
manager = IngestorManager(
|
|
redis_data={'host': 'localhost', 'port': 6379},
|
|
lease_ttl=60,
|
|
heartbeat_ttl=60,
|
|
poll_interval=5,
|
|
mongo_connection_string='mongodb://localhost:27017',
|
|
mongo_database='sientia',
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metadata=metadata['metadata'],
|
|
metrics_controller=MagicMock(),
|
|
read_engine=read_engine,
|
|
)
|
|
manager.send_notification = MagicMock()
|
|
manager.emit_metric = AsyncMock()
|
|
return manager
|
|
|
|
|
|
def test_read_engine_defaults_to_v2(monkeypatch):
|
|
monkeypatch.delenv('READ_ENGINE', raising=False)
|
|
|
|
assert make_manager().read_engine == 'v2'
|
|
|
|
|
|
def test_read_engine_from_env(monkeypatch):
|
|
monkeypatch.setenv('READ_ENGINE', 'Legacy ')
|
|
|
|
assert make_manager().read_engine == 'legacy'
|
|
|
|
|
|
def test_invalid_read_engine_falls_back_to_v2(monkeypatch):
|
|
monkeypatch.setenv('READ_ENGINE', 'v3')
|
|
|
|
manager = make_manager()
|
|
|
|
assert manager.read_engine == 'v2'
|
|
assert any('Invalid READ_ENGINE' in str(c) for c in manager.logger.warning.call_args_list)
|
|
|
|
|
|
@mark.asyncio
|
|
@patch('ingestor.managers.ingestor_manager.ServerReader')
|
|
@patch('ingestor.managers.ingestor_manager.OpcManager')
|
|
async def test_legacy_initialize_builds_rc3_opc_manager(opc_manager, server_reader):
|
|
manager = make_manager('legacy')
|
|
server_config = {
|
|
'name': 'server1',
|
|
'server_id': '1',
|
|
'url': 'opc.tcp://localhost:4840',
|
|
'subscription_period_ms': 1000,
|
|
'server_uri': 'urn:sientia:opc-ingestor',
|
|
'cert_path': '/app/certs/opc-ingestor-cert.pem',
|
|
'private_key_path': '/app/certs/opc-ingestor-key.pem',
|
|
'server_cert_path': None,
|
|
}
|
|
opc_manager.return_value = MagicMock(connect=AsyncMock())
|
|
|
|
result = await manager.initialize_opc_from_config(server_config)
|
|
|
|
server_reader.assert_not_called()
|
|
opc_manager.assert_called_once_with(
|
|
name='server1',
|
|
url='opc.tcp://localhost:4840',
|
|
subscription_period_ms=1000,
|
|
data_manager=manager.data_manager,
|
|
logger=manager.logger,
|
|
server_uri='urn:sientia:opc-ingestor',
|
|
notification_handler=manager.notification_handler,
|
|
metadata=metadata['metadata'],
|
|
cert_path='/app/certs/opc-ingestor-cert.pem',
|
|
private_key_path='/app/certs/opc-ingestor-key.pem',
|
|
server_cert_path=None,
|
|
metrics_controller=manager.metrics_controller,
|
|
)
|
|
assert result is opc_manager.return_value
|
|
result.connect.assert_awaited_once()
|
|
|
|
|
|
def test_legacy_keeps_raw_slot_v1():
|
|
manager = make_manager('legacy')
|
|
raw = raw_slot_v1()
|
|
|
|
# OpcManager reads tag_name/topics and the file cert paths straight from the slot
|
|
assert manager.parse_slot_config('1', raw) is raw
|
|
|
|
|
|
def test_legacy_rejects_slot_v2():
|
|
manager = make_manager('legacy')
|
|
raw_v2 = {
|
|
'server1': {
|
|
'name': 'server1',
|
|
'server_id': '1',
|
|
'url': 'opc.tcp://localhost:4840',
|
|
'use_security': True,
|
|
'subscription_period_ms': 500,
|
|
'tags': {
|
|
'ns=2;i=2': {
|
|
'frequency': '1000',
|
|
'names': [{'collection': 'raw_a', 'tag_name': 'Counter', 'model_id': '1'}],
|
|
}
|
|
},
|
|
}
|
|
}
|
|
|
|
assert manager.parse_slot_config('1', raw_v2) is None
|
|
_, kwargs = manager.send_notification.call_args
|
|
assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1'
|
|
assert 'READ_ENGINE=legacy' in kwargs['message']
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_legacy_remove_server_calls_plain_shutdown():
|
|
manager = make_manager('legacy')
|
|
opc_manager = MagicMock(shutdown=AsyncMock())
|
|
manager.opc_managers = {'server1': opc_manager}
|
|
manager.managed_tags = {'slot1': {'server1': {}}}
|
|
|
|
await manager.remove_server('server1', evict=False)
|
|
|
|
opc_manager.shutdown.assert_awaited_once_with()
|
|
assert manager.opc_managers == {}
|