sientia-dataops-opc-ingestor/tests/unit/test_gateway_runtime.py

from unittest.mock import AsyncMock, MagicMock

import pytest

from ingestor import gateway_runtime as gr
from ingestor.config import GatewayConfig

_PATCHED_CLASSES = (
    'Postgres',
    'build_postgres_config',
    'DataSink',
    'WriteEngine',
    'NatsWriteConsumer',
    'ActiveReadEngine',
    'GrpcServer',
)


@pytest.fixture
def mocks(monkeypatch):
    patched = {}
    for name in _PATCHED_CLASSES:
        m = MagicMock()
        monkeypatch.setattr(gr, name, m)
        patched[name] = m

    patched['build_postgres_config'].return_value = {
        'host': 'localhost',
        'port': 5432,
        'user': 'u',
        'password': 'p',
        'dbname': 'd',
        'min_connections': 5,
        'max_connections': 20,
    }
    # The pool is owned by IngestorManager (read path) and only borrowed here.
    patched['pool'] = MagicMock()
    patched['pool'].close_all = AsyncMock()
    patched['pool'].release_idle = AsyncMock()
    patched['NatsWriteConsumer'].return_value.start = AsyncMock()
    patched['NatsWriteConsumer'].return_value.stop = AsyncMock()
    patched['NatsWriteConsumer'].return_value.closed = False
    patched['GrpcServer'].return_value.start = AsyncMock()
    patched['GrpcServer'].return_value.stop = AsyncMock()
    return patched


def make_runtime(pool=None, **config_overrides):
    if pool is None:
        pool = MagicMock(close_all=AsyncMock(), release_idle=AsyncMock())
    return gr.GatewayRuntime(
        config=GatewayConfig(**config_overrides),
        data_manager=MagicMock(),
        pool=pool,
        logger=MagicMock(),
        notification_handler=MagicMock(),
        metrics_controller=MagicMock(),
    )


def test_builds_object_graph_with_config_flags(mocks):
    runtime = make_runtime(
        pool=mocks['pool'],
        write_timestamp=False,
        write_quality=False,
        write_min_pause_s=0.25,
        grpc_enabled=False,
    )

    # Writes and active reads share the pool built by the read path.
    assert runtime.pool is mocks['pool']
    mocks['DataSink'].assert_called_once()

    mocks['WriteEngine'].assert_called_once()
    args, kwargs = mocks['WriteEngine'].call_args
    assert args[0] is mocks['pool']
    assert kwargs['write_timestamp'] is False
    assert kwargs['write_quality'] is False
    assert kwargs['min_pause_s'] == 0.25

    args, _ = mocks['ActiveReadEngine'].call_args
    assert args[0] is mocks['pool']

    mocks['GrpcServer'].assert_not_called()
    assert runtime.grpc_server is None


def test_builds_grpc_server_when_enabled(mocks):
    runtime = make_runtime(grpc_enabled=True, grpc_port=12345)

    mocks['GrpcServer'].assert_called_once()
    assert runtime.grpc_server is mocks['GrpcServer'].return_value


@pytest.mark.asyncio
async def test_start_starts_nats_consumer(mocks):
    runtime = make_runtime(grpc_enabled=False)

    await runtime.start()

    mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
    mocks['GrpcServer'].return_value.start.assert_not_called()
    assert runtime._nats_started is True


@pytest.mark.asyncio
async def test_start_also_starts_grpc_when_enabled(mocks):
    runtime = make_runtime(grpc_enabled=True)

    await runtime.start()

    mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
    mocks['GrpcServer'].return_value.start.assert_awaited_once()


@pytest.mark.asyncio
async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks):
    runtime = make_runtime()
    runtime.send_notification_async = AsyncMock()
    nats_mock = mocks['NatsWriteConsumer'].return_value
    nats_mock.start = AsyncMock(side_effect=[Exception('boom'), Exception('boom'), None])

    await runtime.start()  # must not raise

    assert runtime._nats_started is False
    runtime.logger.error.assert_called_once()
    runtime.send_notification_async.assert_awaited_once()

    await runtime.idle_sweep()  # second failure: retry connects, but no re-notify

    assert runtime._nats_started is False
    runtime.send_notification_async.assert_awaited_once()  # still just one notification

    await runtime.idle_sweep()

    assert runtime._nats_started is True
    assert nats_mock.start.await_count == 3
    # flag reset on success: a future outage notifies again
    assert runtime._nats_notified is False


@pytest.mark.asyncio
async def test_stop_stops_started_components_but_not_the_pool(mocks):
    runtime = make_runtime(pool=mocks['pool'], grpc_enabled=True)
    await runtime.start()

    await runtime.stop()

    mocks['GrpcServer'].return_value.stop.assert_awaited_once()
    mocks['NatsWriteConsumer'].return_value.stop.assert_awaited_once()
    # IngestorManager.shutdown() closes it, after its readers unsubscribe.
    mocks['pool'].close_all.assert_not_called()


@pytest.mark.asyncio
async def test_stop_skips_nats_stop_if_never_started(mocks):
    runtime = make_runtime(pool=mocks['pool'])
    runtime.send_notification_async = AsyncMock()
    nats_mock = mocks['NatsWriteConsumer'].return_value
    nats_mock.start = AsyncMock(side_effect=Exception('boom'))

    await runtime.start()  # nats fails to start
    await runtime.stop()

    nats_mock.stop.assert_not_called()
    mocks['pool'].close_all.assert_not_called()


@pytest.mark.asyncio
async def test_idle_sweep_releases_idle_connections(mocks):
    runtime = make_runtime(pool=mocks['pool'], idle_conn_max_s=42.0)
    runtime._nats_started = True  # isolate from the retry path

    await runtime.idle_sweep()

    mocks['pool'].release_idle.assert_awaited_once_with(42.0)


def test_write_engine_resolves_servers_from_the_pool_and_mongo(mocks):
    """SIENTIAPDE-1975 GW1: write commands carry only the server_id."""
    data_manager = MagicMock()
    runtime = gr.GatewayRuntime(
        config=GatewayConfig(),
        data_manager=data_manager,
        pool=mocks['pool'],
        logger=MagicMock(),
        notification_handler=MagicMock(),
        metrics_controller=MagicMock(),
    )

    _, kwargs = mocks['WriteEngine'].call_args
    assert kwargs['servers'] is runtime.servers
    assert runtime.servers.pool is mocks['pool']
    assert runtime.servers.mongo is data_manager.mongo_repository


@pytest.mark.asyncio
async def test_idle_sweep_restarts_a_closed_nats_connection(mocks):
    runtime = make_runtime(pool=mocks['pool'])
    nats_mock = mocks['NatsWriteConsumer'].return_value
    await runtime.start()
    assert nats_mock.start.await_count == 1

    nats_mock.closed = True  # closed for good (e.g. by the server)
    await runtime.idle_sweep()

    assert nats_mock.start.await_count == 2
    assert runtime._nats_started is True
    runtime.logger.warning.assert_called_once()


@pytest.mark.asyncio
async def test_idle_sweep_leaves_a_live_nats_connection_alone(mocks):
    runtime = make_runtime(pool=mocks['pool'])
    nats_mock = mocks['NatsWriteConsumer'].return_value
    await runtime.start()

    await runtime.idle_sweep()

    assert nats_mock.start.await_count == 1