From 3d8ee1c2ed67ab13d57fc397f70f5a16550f65df Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 6 Oct 2026 13:54:04 -0300 Subject: [PATCH] SIENTIAPDE-1974: fix(read): code-review fixes for the slot v2 read path - config: GatewayConfig.certs_from_env() reads only CERT_BUCKET/CERT_CACHE_DIR; IngestorManager uses it, so a malformed write/gRPC env var (GRPC_PORT, WRITE_MIN_PAUSE_S, IDLE_CONN_MAX_S) can no longer stop the reads. - certs: CertProvider raises CertificateNotConfiguredError with an actionable message when no platform_certificates document exists; downloads once under an asyncio.Lock, caches the paths, writes atomically (tmp + os.replace); invalidate() forces a fresh download. - pool: server_id normalized to str (1 and '1' share a session); warns when a get() asks for other url/server_uri/use_security than the pooled client was opened with (client reused, the slot config wins); invalidates the cert cache when a secured connect fails. - SubscriptionEngine: tracks nodes per subscription and forgets them on unsubscribe (no more cycle alarms / tag count for removed tags), keeping addresses still monitored by another slot. - ServerReader.shutdown(evict) / IngestorManager.remove_server(evict): a server that only left this pod's slots keeps the shared connection for NATS writes/gRPC; lost servers and config changes still evict. - IngestorManager warns once per server when a slot still sets server_cert_path (server certificate pinning is no longer applied). - slots: CONNECTION_FIELDS exported and reused by ServerReader. Accepted, not changed: read metrics label server_name with the server_id (same as the write path, which only knows the id); a malformed slot is notified each lease attempt (alerts are deduplicated by notification_id). Tests: 236 unit tests pass (+18); ruff clean; local e2e against an asyncua server incl. slot moved (connection kept and reused by the write path) and lost-server eviction/re-initialization. Co-Authored-By: Claude Opus 5.5 --- ingestor/config.py | 10 +++ ingestor/connection/certs.py | 49 +++++++++++-- ingestor/connection/pool.py | 32 ++++++++- ingestor/engines/subscription.py | 21 ++++++ ingestor/io/slots.py | 2 + ingestor/managers/ingestor_manager.py | 34 +++++++-- ingestor/managers/server_reader.py | 38 +++++----- tests/unit/connection/test_certs.py | 76 ++++++++++++++++++++ tests/unit/connection/test_pool.py | 60 ++++++++++++++++ tests/unit/engines/test_subscription.py | 43 +++++++++++ tests/unit/managers/test_ingestor_manager.py | 56 ++++++++++++++- tests/unit/managers/test_server_reader.py | 15 ++++ tests/unit/test_config.py | 16 +++++ 13 files changed, 421 insertions(+), 31 deletions(-) diff --git a/ingestor/config.py b/ingestor/config.py index 87492c0..93d5a9a 100644 --- a/ingestor/config.py +++ b/ingestor/config.py @@ -31,6 +31,16 @@ class GatewayConfig: write_min_pause_s: float = 0.0 idle_conn_max_s: float = 300.0 + @classmethod + def certs_from_env(cls) -> 'GatewayConfig': + """Only the certificate settings โ€” what the read path needs. Plain strings, + no number parsing, so a malformed write/gRPC env var (GRPC_PORT, + WRITE_MIN_PAUSE_S, ...) can only disable the runtime, never the reads.""" + return cls( + cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket), + cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir), + ) + @classmethod def from_env(cls) -> 'GatewayConfig': return cls( diff --git a/ingestor/connection/certs.py b/ingestor/connection/certs.py index 8c57b5b..2c8d4a8 100644 --- a/ingestor/connection/certs.py +++ b/ingestor/connection/certs.py @@ -1,3 +1,5 @@ +import asyncio +import os from os import getenv from pathlib import Path @@ -8,10 +10,20 @@ from ingestor.config import _env_bool COLLECTION_NAME = 'platform_certificates' +class CertificateNotConfiguredError(RuntimeError): + """No usable platform OPC client certificate is registered (Mongo + MinIO).""" + + class CertProvider: """Downloads the platform's client cert + private key (Mongo metadata, MinIO bytes) into a writable cache dir โ€” pod root FS is read-only (ยง12), - so `cache_dir` must be an emptyDir/tmpfs mount.""" + so `cache_dir` must be an emptyDir/tmpfs mount. + + The certificate is single and shared by every connection (reads, writes, + gRPC), so it is downloaded once and cached; `invalidate()` forces a fresh + download (the pool calls it when a secured connection fails, e.g. after the + certificate was replaced in MinIO). The lock serializes downloads so two + connections never rewrite the cache files at the same time.""" def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None): self.mongo = mongo @@ -19,19 +31,44 @@ class CertProvider: self.cache_dir = cache_dir self.bucket = bucket self.metadata = metadata or {} + self._lock = asyncio.Lock() + self._paths: tuple[str, str] | None = None async def fetch(self) -> tuple[str, str]: + async with self._lock: + if self._paths is None: + self._paths = await self._download_all() + return self._paths + + def invalidate(self) -> None: + self._paths = None + + async def _download_all(self) -> tuple[str, str]: doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata) + try: + cert_object = doc['client_cert']['minio_object_name'] + key_object = doc['private_key']['minio_object_name'] + except (KeyError, TypeError) as e: + raise CertificateNotConfiguredError( + f'No OPC client certificate configured: the Mongo collection ' + f"'{COLLECTION_NAME}' needs a document with client_cert.minio_object_name " + f'and private_key.minio_object_name (objects in the MinIO bucket ' + f"'{self.bucket}'). Register the platform certificate before using " + f'secured OPC servers.' + ) from e Path(self.cache_dir).mkdir(parents=True, exist_ok=True) - cert_path = await self._download(doc['client_cert']['minio_object_name']) - key_path = await self._download(doc['private_key']['minio_object_name']) + cert_path = await self._download(cert_object) + key_path = await self._download(key_object) return cert_path, key_path async def _download(self, object_name: str) -> str: data = await self.minio.download_file(object_name, self.bucket, self.metadata) - local_path = str(Path(self.cache_dir) / Path(object_name).name) - Path(local_path).write_bytes(data) - return local_path + local_path = Path(self.cache_dir) / Path(object_name).name + # Atomic replace: a connection loading the previous file never sees it half-written. + tmp_path = local_path.with_name(local_path.name + '.tmp') + tmp_path.write_bytes(data) + os.replace(tmp_path, local_path) + return str(local_path) class LazyMinio: diff --git a/ingestor/connection/pool.py b/ingestor/connection/pool.py index 1f66a5c..ea0bde6 100644 --- a/ingestor/connection/pool.py +++ b/ingestor/connection/pool.py @@ -30,6 +30,8 @@ class OpcConnectionPool(SientiaMonitoring): # callbacks and never calls get(), so without a pin release_idle() would # evict it as idle and silently stop the reads (SIENTIAPDE-1974 G4). self._pins: dict[str, int] = {} + # Connection settings each pooled client was opened with (see _check_cfg). + self._configs: dict[str, dict] = {} # ponytail: one Lock per server_id, never cleaned up โ€” server set is # small and bounded. Add eviction of stale locks only if it grows. self._locks: dict[str, asyncio.Lock] = {} @@ -45,16 +47,35 @@ class OpcConnectionPool(SientiaMonitoring): """Return the cached client for `server_id`, connecting one if needed. Double-checked under a per-server lock so concurrent callers for the - same new server share one connection (no leaked, untracked sessions).""" + same new server share one connection (no leaked, untracked sessions). + `server_id` is normalized to str, so 1 and '1' share the same session.""" + server_id = str(server_id) client = self._cached(server_id) if client is not None: + self._check_cfg(server_id, server_cfg) return client async with self._locks.setdefault(server_id, asyncio.Lock()): client = self._cached(server_id) # re-check: a racer may have connected if client is not None: + self._check_cfg(server_id, server_cfg) return client return await self._open(server_id, server_cfg) + def _check_cfg(self, server_id: str, server_cfg: dict) -> None: + """Warn when a caller asks for a server with other connection settings than + the pooled client was opened with. The client is reused as is: reconnecting + on every mismatch would let one malformed write command drop the reads. + Config changes from the slot evict the client first (ServerReader.shutdown).""" + opened = self._configs.get(server_id, {}) + diff = [ + k for k in ('url', 'server_uri', 'use_security') if opened.get(k) != server_cfg.get(k) + ] + if diff: + self.logger.warning( + f'OPC server {server_id}: reusing the pooled connection, but {diff} differ ' + f'from the settings it was opened with' + ) + def _cached(self, server_id: str) -> Client | None: entry = self._connections.get(server_id) if entry is None: @@ -67,6 +88,10 @@ class OpcConnectionPool(SientiaMonitoring): try: client = await self._connect(server_cfg) except Exception: + if server_cfg.get('use_security'): + # The cached certificate may be the cause (replaced in MinIO, + # rejected by the server): download it again on the next attempt. + self.cert_provider.invalidate() await self.emit_metric( metric_object=metrics.OPC_CONNECTIONS_FAILED, tags={'pod_id': self.pod_id, 'server_name': server_id}, @@ -74,6 +99,7 @@ class OpcConnectionPool(SientiaMonitoring): raise self._connections[server_id] = (client, self.clock.elapsed()) + self._configs[server_id] = dict(server_cfg) await self._record_connected(server_id, server_cfg['url']) return client @@ -102,10 +128,12 @@ class OpcConnectionPool(SientiaMonitoring): def pin(self, server_id: str) -> None: """Mark one more active subscription on `server_id` (never evicted as idle).""" + server_id = str(server_id) self._pins[server_id] = self._pins.get(server_id, 0) + 1 def unpin(self, server_id: str) -> None: """Drop one active subscription mark from `server_id`.""" + server_id = str(server_id) count = self._pins.get(server_id, 0) - 1 if count > 0: self._pins[server_id] = count @@ -127,6 +155,7 @@ class OpcConnectionPool(SientiaMonitoring): async def evict(self, server_id: str): """Disconnect `server_id` now (lost server or changed config); the next get() reconnects. No-op when the server has no pooled client.""" + server_id = str(server_id) if server_id in self._connections: await self._evict(server_id) @@ -139,6 +168,7 @@ class OpcConnectionPool(SientiaMonitoring): async def _evict(self, server_id: str): client, _ = self._connections.pop(server_id) + self._configs.pop(server_id, None) await self._disconnect_with_retries(client, server_id) await self.emit_metric( metric_object=metrics.OPC_DISCONNECTIONS_TOTAL, diff --git a/ingestor/engines/subscription.py b/ingestor/engines/subscription.py index 16ebabe..db5b8b7 100644 --- a/ingestor/engines/subscription.py +++ b/ingestor/engines/subscription.py @@ -35,6 +35,9 @@ class SubscriptionEngine(SientiaMonitoring): self.subscriptions: dict = {} self._clients: dict = {} self._servers: dict = {} + # Addresses each subscription monitors, so unsubscribe can forget them + # (otherwise check_cycles keeps alarming on tags nobody reads anymore). + self._sub_nodes: dict[str, set[str]] = {} self.non_receive_count = 0 SientiaMonitoring.__init__( @@ -75,6 +78,7 @@ class SubscriptionEngine(SientiaMonitoring): server_id = self._servers[subscription] addr_nodes = [client.get_node(n) for n in nodes] self.nodes.update(nodes) + self._sub_nodes.setdefault(subscription, set()).update(nodes) self._init_cycle_rules(nodes, collect_period, server_id) await self.subscriptions[subscription].subscribe_data_change(addr_nodes) @@ -102,6 +106,7 @@ class SubscriptionEngine(SientiaMonitoring): server_id = self._servers.pop(subscription, None) if server_id is not None: self.pool.unpin(server_id) + self._forget_nodes(subscription) try: await sub.delete() except Exception as e: @@ -109,6 +114,22 @@ class SubscriptionEngine(SientiaMonitoring): # dropped from the books either way so the slot can be re-subscribed. self.logger.warning(f'Failed to delete subscription {subscription}: {e}') self.logger.info(f'Unsubscribed from {subscription}.') + if server_id is not None: + await self.emit_metric( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=len(self.nodes), + tags={'pod_id': self.pod_id, 'server_name': server_id}, + ) + + def _forget_nodes(self, subscription: str) -> None: + """Drop the subscription's addresses, except those another subscription + of this engine still monitors.""" + nodes = self._sub_nodes.pop(subscription, set()) + still_monitored = set().union(*self._sub_nodes.values()) + for node in nodes: + if node not in still_monitored: + self.nodes.pop(node, None) async def datachange_notification(self, node, _val, data): """asyncua subscription callback: fan the new value out to every name diff --git a/ingestor/io/slots.py b/ingestor/io/slots.py index d9e0180..d8ff695 100644 --- a/ingestor/io/slots.py +++ b/ingestor/io/slots.py @@ -11,6 +11,8 @@ any order (SIENTIAPDE-1974): """ SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms') +# What a connection/subscription needs (the pool keys clients by server_id itself). +CONNECTION_FIELDS = tuple(f for f in SERVER_FIELDS if f != 'server_id') def _parse_names(address: str, tag: dict) -> list[dict]: diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index c7e0504..2061073 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -102,7 +102,7 @@ class IngestorManager(SientiaMonitoring): # Read path (SIENTIAPDE-1974): pooled connections with the certificate from # MinIO. Nothing here touches the network at construction (MinIO is lazy), so # a MinIO/Postgres/NATS outage can't stop the reads before they start. - config = gateway_config or GatewayConfig.from_env() + config = gateway_config or GatewayConfig.certs_from_env() self.cert_provider = build_cert_provider( mongo=self.data_manager.mongo_repository, config=config, @@ -115,6 +115,7 @@ class IngestorManager(SientiaMonitoring): self.cert_provider, logger, notification_handler, metrics_controller ) self.sink = DataSink(self.data_manager, None) + self._server_cert_warned: set[str] = set() self.resource_manager = ResourceManager( host=redis_host, port=redis_port, @@ -198,12 +199,18 @@ class IngestorManager(SientiaMonitoring): await self.pool.close_all() self.data_manager.shutdown() - async def remove_server(self, server: str): + async def remove_server(self, server: str, evict: bool = True): """ Removes an OPC server from the ingestor. + + Args: + server (str): The server name. + evict (bool): Also drop the pooled connection (lost server, failed + subscription). False when the server merely left this pod's slots: + the shared connection stays for writes/gRPC until the idle sweep. """ if server in self.opc_managers: - await self.opc_managers[server].shutdown() + await self.opc_managers[server].shutdown(evict=evict) del self.opc_managers[server] for slot, _config in self.managed_tags.items(): self.managed_tags[slot].pop(server, None) @@ -269,9 +276,10 @@ class IngestorManager(SientiaMonitoring): for server in servers: if server not in registered_servers: self.logger.warning( - f'Server {server} not found in managed tags. Desconnecting from server.' + f'Server {server} not found in managed tags. Unsubscribing from server.' ) - await self.remove_server(server) + # Slot moved away: keep the shared connection for writes/gRPC. + await self.remove_server(server, evict=False) await self.emit_metric( metric_object=metrics.OPC_MANAGERS_ACTIVE, @@ -559,7 +567,7 @@ class IngestorManager(SientiaMonitoring): (logged and notified). """ try: - return parse_slot(raw) + parsed = parse_slot(raw) except (AttributeError, KeyError, TypeError, ValueError) as e: self.send_notification( metadata=self.metadata, @@ -570,6 +578,20 @@ class IngestorManager(SientiaMonitoring): ) self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}') return None + self._warn_server_cert_path(raw) + return parsed + + def _warn_server_cert_path(self, raw: dict) -> None: + """Logs once per server when a slot still pins a server certificate + (`server_cert_path`): connections now come from the pool with the platform + certificate from MinIO, and that file-based pin is no longer applied.""" + for name, server in raw.items(): + if server.get('server_cert_path') and name not in self._server_cert_warned: + self._server_cert_warned.add(name) + self.logger.warning( + f'OPC server {name}: server_cert_path is set in the slot but is no ' + f'longer applied (server certificate validation is disabled)' + ) async def drop_slot_leases(self, ids: list[str]) -> None: """ diff --git a/ingestor/managers/server_reader.py b/ingestor/managers/server_reader.py index 7bb892c..21c76f7 100644 --- a/ingestor/managers/server_reader.py +++ b/ingestor/managers/server_reader.py @@ -20,8 +20,7 @@ import ingestor.metrics as metrics from ingestor.connection.pool import OpcConnectionPool from ingestor.engines.subscription import SubscriptionEngine from ingestor.io.data_sink import DataSink - -_CONNECTION_FIELDS = ('url', 'server_uri', 'use_security', 'subscription_period_ms') +from ingestor.io.slots import CONNECTION_FIELDS class ServerReader: @@ -42,7 +41,7 @@ class ServerReader: self.name = name self.server_id = str(server_config['server_id']) self.url = server_config['url'] - self.server_cfg = {k: server_config[k] for k in _CONNECTION_FIELDS if k in server_config} + self.server_cfg = {k: server_config[k] for k in CONNECTION_FIELDS if k in server_config} self.pool = pool self.logger = logger self.config: dict | None = None @@ -77,22 +76,27 @@ class ServerReader: async def check_opc_listenning(self) -> bool: return await self.engine.check_opc_listenning(self.server_id) - async def shutdown(self): - """Drop every subscription and the pooled connection, so the next - initialization reconnects (lost server or changed server config).""" + async def shutdown(self, evict: bool = True): + """Drop every subscription of this server. + + With `evict` (lost server or changed server config) the pooled connection is + also dropped, so the next initialization reconnects. Without it (the slot just + moved to another pod) the connection stays for writes/gRPC: the connection is + shared, and the idle sweep closes it if nobody uses it anymore.""" for slot in list(self.engine.subscriptions): await self.engine.unsubscribe(slot) - await self.pool.evict(self.server_id) - await self.engine.emit_metric( - metric_object=metrics.OPC_CONNECTION_STATUS, - method='set', - value=0, - tags={ - 'pod_id': self.engine.pod_id, - 'server_name': self.server_id, - 'server_url': self.url, - }, - ) + if evict: + await self.pool.evict(self.server_id) + await self.engine.emit_metric( + metric_object=metrics.OPC_CONNECTION_STATUS, + method='set', + value=0, + tags={ + 'pod_id': self.engine.pod_id, + 'server_name': self.server_id, + 'server_url': self.url, + }, + ) await self.engine.emit_metric( metric_object=metrics.OPC_TAGS_SUBSCRIBED, method='set', diff --git a/tests/unit/connection/test_certs.py b/tests/unit/connection/test_certs.py index db98ce7..dcfa97c 100644 --- a/tests/unit/connection/test_certs.py +++ b/tests/unit/connection/test_certs.py @@ -70,3 +70,79 @@ def test_build_cert_provider_is_lazy_and_uses_config(monkeypatch): assert provider.cache_dir == 'certs-cache' assert isinstance(provider.minio, certs.LazyMinio) minio_cls.assert_not_called() # MinIO is only reached when a cert is needed + + +def _provider(tmp_path, doc=None, data=b'pem'): + from unittest.mock import MagicMock + + mongo = MagicMock() + mongo.find_one = AsyncMock( + return_value=doc + if doc is not None + else { + 'client_cert': {'minio_object_name': 'platform_certificates/c.pem'}, + 'private_key': {'minio_object_name': 'platform_certificates/k.pem'}, + } + ) + minio = MagicMock() + minio.download_file = AsyncMock(return_value=data) + return CertProvider(mongo=mongo, minio=minio, cache_dir=str(tmp_path), bucket='b'), minio + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'doc', + [ + None, + {}, + {'client_cert': {'minio_object_name': 'c.pem'}}, + {'client_cert': None, 'private_key': {}}, + ], +) +async def test_fetch_without_platform_certificate_raises_clear_error(tmp_path, doc): + from ingestor.connection.certs import CertificateNotConfiguredError + + provider, minio = _provider(tmp_path) + provider.mongo.find_one.return_value = doc + + with pytest.raises(CertificateNotConfiguredError, match='platform_certificates'): + await provider.fetch() + minio.download_file.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_fetch_is_cached_until_invalidated(tmp_path): + provider, minio = _provider(tmp_path) + + first = await provider.fetch() + second = await provider.fetch() + assert first == second + assert minio.download_file.await_count == 2 # cert + key, once + + provider.invalidate() + await provider.fetch() + assert minio.download_file.await_count == 4 + + +@pytest.mark.asyncio +async def test_concurrent_fetches_download_once(tmp_path): + import asyncio + + provider, minio = _provider(tmp_path) + + results = await asyncio.gather(*(provider.fetch() for _ in range(5))) + + assert len(set(results)) == 1 + assert minio.download_file.await_count == 2 + + +@pytest.mark.asyncio +async def test_download_replaces_file_atomically(tmp_path): + provider, minio = _provider(tmp_path, data=b'new') + (tmp_path / 'c.pem').write_bytes(b'old') + + cert_path, _ = await provider.fetch() + + assert (tmp_path / 'c.pem').read_bytes() == b'new' + assert cert_path == str(tmp_path / 'c.pem') + assert not list(tmp_path.glob('*.tmp')) diff --git a/tests/unit/connection/test_pool.py b/tests/unit/connection/test_pool.py index 4b1a21b..e1bf4fa 100644 --- a/tests/unit/connection/test_pool.py +++ b/tests/unit/connection/test_pool.py @@ -248,3 +248,63 @@ async def test_evict_unknown_server_is_noop(): await pool.evict('nope') pool.emit_metric.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_get_normalizes_server_id_to_str(): + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + c1 = await pool.get(1, {'url': 'opc.tcp://h:4840', 'use_security': False}) + c2 = await pool.get('1', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + assert c1 is c2 + assert fake_client.connect.await_count == 1 + + +@pytest.mark.asyncio +async def test_get_with_different_cfg_reuses_client_and_warns(): + pool = make_pool() + fake_client = AsyncMock() + with patch('ingestor.connection.pool.Client', return_value=fake_client): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + client = await pool.get('srv', {'url': 'opc.tcp://10.0.0.5:4840', 'use_security': False}) + + assert client is fake_client + assert fake_client.connect.await_count == 1 + pool.logger.warning.assert_called_once() + assert 'url' in pool.logger.warning.call_args.args[0] + + +@pytest.mark.asyncio +async def test_secured_connect_failure_invalidates_cert_cache(): + from unittest.mock import MagicMock + + cert_provider = MagicMock() + cert_provider.fetch = AsyncMock(return_value=('cert.pem', 'key.pem')) + pool = make_pool(cert_provider=cert_provider) + fake_client = AsyncMock() + fake_client.connect.side_effect = RuntimeError('BadCertificateUntrusted') + with ( + patch('ingestor.connection.pool.Client', return_value=fake_client), + patch('ingestor.connection.pool.apply_security', AsyncMock()), + ): + with pytest.raises(RuntimeError): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': True}) + + cert_provider.invalidate.assert_called_once() + + +@pytest.mark.asyncio +async def test_insecure_connect_failure_keeps_cert_cache(): + from unittest.mock import MagicMock + + cert_provider = MagicMock() + pool = make_pool(cert_provider=cert_provider) + fake_client = AsyncMock() + fake_client.connect.side_effect = RuntimeError('unreachable') + with patch('ingestor.connection.pool.Client', return_value=fake_client): + with pytest.raises(RuntimeError): + await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False}) + + cert_provider.invalidate.assert_not_called() diff --git a/tests/unit/engines/test_subscription.py b/tests/unit/engines/test_subscription.py index 6a07a64..6ab7cb9 100644 --- a/tests/unit/engines/test_subscription.py +++ b/tests/unit/engines/test_subscription.py @@ -255,3 +255,46 @@ async def test_unsubscribe_tolerates_dead_connection(): assert 'slot1' not in eng.subscriptions eng.pool.unpin.assert_called_once_with('1') eng.logger.warning.assert_called_once() + + +async def _subscribed_engine(slots): + """Engine with one subscription per slot: {slot: [addresses]}.""" + eng = make_engine() + eng.pool = MagicMock( + get=AsyncMock( + return_value=MagicMock(create_subscription=AsyncMock(return_value=AsyncMock())) + ) + ) + for slot, addresses in slots.items(): + await eng.create_subscription('1', {'subscription_period_ms': 500}, slot) + nodes = { + a: {'frequency': '1000', 'names': [{'collection': 'c', 'tag_name': a}]} + for a in addresses + } + await eng.subscribe(slot, nodes, 10) + return eng + + +@pytest.mark.asyncio +async def test_unsubscribe_forgets_the_slot_nodes(): + eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=3']}) + + await eng.unsubscribe('s1') + + # no more cycle checks / tag counting for addresses nobody reads anymore + assert set(eng.nodes) == {'ns=2;i=3'} + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_TAGS_SUBSCRIBED, + method='set', + value=1, + tags={'pod_id': eng.pod_id, 'server_name': '1'}, + ) + + +@pytest.mark.asyncio +async def test_unsubscribe_keeps_node_shared_with_another_slot(): + eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=2']}) + + await eng.unsubscribe('s1') + + assert set(eng.nodes) == {'ns=2;i=2'} diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index de6580c..c7dfa23 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -314,7 +314,9 @@ async def test_update_opc_servers(metrics, ingestor_manager): assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'} assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'} - ingestor_manager.remove_server.assert_any_call('server4') + # 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 @@ -738,3 +740,55 @@ async def test_update_slot_config_keeps_last_good_config_when_malformed(ingestor 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 diff --git a/tests/unit/managers/test_server_reader.py b/tests/unit/managers/test_server_reader.py index 97b9cf6..b2cc54f 100644 --- a/tests/unit/managers/test_server_reader.py +++ b/tests/unit/managers/test_server_reader.py @@ -127,3 +127,18 @@ async def test_shutdown_unsubscribes_all_slots_evicts_and_zeroes_metrics(): value=0, tags={'pod_id': 'pod', 'server_name': '1'}, ) + + +@pytest.mark.asyncio +async def test_shutdown_without_evict_keeps_the_shared_connection(): + """Slot moved to another pod: drop the subscriptions, keep the pooled + connection for NATS writes/gRPC (review #7).""" + reader, pool = make_reader() + reader.engine.subscriptions.update({'1': MagicMock()}) + + await reader.shutdown(evict=False) + + reader.engine.unsubscribe.assert_awaited_once_with('1') + pool.evict.assert_not_awaited() + for c in reader.engine.emit_metric.await_args_list: + assert c.kwargs['metric_object'] is not metrics.OPC_CONNECTION_STATUS diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 04482a2..8343fe7 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -69,3 +69,19 @@ def test_bool_parsing_edge_cases(monkeypatch, value, expected): cfg = GatewayConfig.from_env() assert cfg.grpc_enabled is expected + + +def test_certs_from_env_ignores_malformed_numeric_vars(monkeypatch): + import pytest + + monkeypatch.setenv('GRPC_PORT', '') + monkeypatch.setenv('IDLE_CONN_MAX_S', 'abc') + monkeypatch.setenv('CERT_BUCKET', 'my-bucket') + monkeypatch.setenv('CERT_CACHE_DIR', 'certs-cache') + + cfg = GatewayConfig.certs_from_env() + + assert cfg.cert_bucket == 'my-bucket' + assert cfg.cert_cache_dir == 'certs-cache' + with pytest.raises(ValueError): + GatewayConfig.from_env() # the full config still fails, only the runtime uses it