sientia-dataops-opc-ingestor/tests/unit/managers/test_ingestor_manager.py

import gc
from unittest.mock import ANY, 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
from ingestor.managers.opc_manager import OpcManager
from ingestor.managers.server_reader import ServerReader

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 == {}


@mark.asyncio
async def test_unsubscribe_slot(ingestor_manager, reader_double):
    ingestor_manager.managed_tags = {
        'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}},
        'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}},
    }
    ingestor_manager.opc_managers = {
        'server1': reader_double('slot1', 'slot2'),
        'server2': reader_double('slot1'),
        'server3': reader_double('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, reader_double):
    """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_double('slot1', 'slot3'),
        'server2': reader_double('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, reader_double):
    """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_double('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_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, reader_double):
    """A reader that fails to unsubscribe is notified and forgotten; the others still go"""
    broken = reader_double('slot1')
    broken.unsubscribe = AsyncMock(side_effect=ConnectionError('session closed'))
    healthy = reader_double('slot1')
    ingestor_manager.opc_managers = {'server1': broken, 'server2': healthy}

    await ingestor_manager.unsubscribe_slot('slot1')

    healthy.unsubscribe.assert_called_once_with('slot1')
    assert 'slot1' not in broken.subscriptions
    ingestor_manager.send_notification_async.assert_called_once_with(
        metadata=ingestor_manager.metadata,
        notification_id='OPC_UNSUBSCRIBE_ERROR_slot1:server1',
        message='Failed to unsubscribe slot slot1 from server1: session closed',
        block='opc_manager',
        level=NotificationLevel.ERROR,
        attachment_content=ANY,
    )


@mark.asyncio
async def test_unsubscribe_slot_with_real_readers(ingestor_manager):
    """Contract: both real readers expose `subscriptions` keyed by slot, and
    unsubscribe_slot closes the slot's subscription on each of them"""
    pool = MagicMock(get=AsyncMock(), evict=AsyncMock())
    server_reader = ServerReader(
        name='kepserver',
        server_config={'server_id': '1', 'url': 'opc.tcp://kepserver:49320'},
        pool=pool,
        sink=AsyncMock(),
        logger=MagicMock(),
        notification_handler=MagicMock(),
        metrics_controller=AsyncMock(),
        metadata={},
    )
    server_reader.engine.emit_metric = AsyncMock()
    v2_subscription = AsyncMock()
    server_reader.engine.subscriptions['slot1'] = v2_subscription
    server_reader.engine._servers['slot1'] = '1'

    with patch('ingestor.managers.opc_manager.metrics'):
        legacy_reader = OpcManager(
            name='legacy',
            url='opc.tcp://legacy:4840',
            data_manager=AsyncMock(),
            subscription_period_ms=1000,
            logger=MagicMock(),
            server_uri='opc.tcp://legacy:4840',
            notification_handler=MagicMock(),
            metadata={},
            metrics_controller=AsyncMock(),
        )
    legacy_subscription = AsyncMock()
    legacy_reader.subscriptions['slot1'] = legacy_subscription

    ingestor_manager.opc_managers = {'kepserver': server_reader, 'legacy': legacy_reader}

    await ingestor_manager.unsubscribe_slot('slot1')

    v2_subscription.delete.assert_awaited_once()
    legacy_subscription.delete.assert_awaited_once()
    assert 'slot1' not in server_reader.subscriptions
    assert 'slot1' not in legacy_reader.subscriptions
    pool.unpin.assert_called_once_with('1')


@mark.asyncio
async def test_update_slot_config_unsubscribes_removed_slot(ingestor_manager, reader_double):
    """A slot deleted from Redis is unsubscribed right away, not left to the main loop"""
    ingestor_manager.managed_tags = {
        'slot1': parse_slot(raw_slot_v1('Same')),
        'slot2': parse_slot(raw_slot_v1('Same')),
    }
    reader = reader_double('slot1', 'slot2')
    ingestor_manager.opc_managers = {'test_server': reader}
    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
        side_effect=lambda slot: raw_slot_v1('Same') if slot == 'slot1' else None
    )
    ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()

    await ingestor_manager.update_slot_config()

    reader.unsubscribe.assert_called_once_with('slot2')
    assert list(ingestor_manager.managed_tags) == ['slot1']
    assert list(reader.subscriptions) == ['slot1']


@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 == {}