Merge pull request #38 from Aignosi/rc/ingestor2

SIENTIAPDE-1973: OPC Gateway — RC2 (GatewayRuntime wiring)

.env.example

@@ -15,3 +15,32 @@ MONGODB_USERNAME="mongo_user"
 MONGODB_PASSWORD="mongo_password"
 MONGODB_DATABASE="sientia"
 
+# OPC Gateway
+NATS_URL="nats://localhost:4222"
+NATS_WRITE_SUBJECT="opc.write"
+NATS_QUEUE="gateway-writers"
+GRPC_ENABLED="false"
+GRPC_PORT="50051"
+CERT_BUCKET="sientia-dataops"
+CERT_CACHE_DIR="/tmp/opc-certs"
+WRITE_TIMESTAMP="true"
+WRITE_QUALITY="true"
+WRITE_MIN_PAUSE_S="0.0"
+IDLE_CONN_MAX_S="300.0"
+
+# OPC Gateway - MinIO (client cert storage, read by GatewayRuntime/CertProvider)
+MINIO_ENDPOINT="localhost:9000"
+MINIO_ACCESS_KEY="minioadmin"
+MINIO_SECRET_KEY="minioadmin"
+MINIO_SECURE="false"
+
+# OPC Gateway - Postgres (write_status updates, read by GatewayRuntime/DataSink
+# via sientia_do.utils.connectors_config.build_postgres_config)
+POSTGRES_HOST="localhost"
+POSTGRES_PORT="5432"
+POSTGRES_USER="sientia"
+POSTGRES_PASSWORD="sientia"
+POSTGRES_DBNAME="sientia"
+POSTGRES_MIN_CONNECTIONS="5"
+POSTGRES_MAX_CONNECTIONS="20"
+

README.md

@@ -49,6 +49,27 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
 - **Data Manager**: Manages data persistence (MongoDB)
 - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
 
+## 🚪 OPC Gateway (SIENTIAPDE-1972)
+
+Alongside the passive ingestion pipeline above, this service also runs an OPC
+Gateway for on-demand and command-driven interactions with OPC UA servers:
+
+- **NATS write consumer**: consumes write commands from `opc.write` (see
+  `NATS_WRITE_SUBJECT`) and executes them against the target OPC server
+  (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`).
+- **On-demand gRPC `ReadTag`**: a gRPC server (`ingestor/adapters/grpc_server.py`,
+  `ingestor/proto/opc_gateway.proto`) for synchronous, single-tag reads.
+- **Scheduled active reads**: a polling read engine (`ingestor/engines/active_read.py`)
+  for tags that need periodic reads outside the subscription pipeline;
+  orchestrator-side wiring lands in SIENTIAPDE-1974.
+- **Certificate management**: client certs for secure OPC connections are
+  fetched from MinIO + a Mongo `platform_certificates` collection
+  (`ingestor/connection/certs.py`) rather than shipped in the image.
+
+See `docs/deploy-notes.md` for the deployment-facing details (env vars,
+health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in
+`.env.example` for the full list of new environment variables.
+
 ## 📋 Prerequisites
 
 - Python 3.11+
@@ -253,6 +274,12 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 - `ingestor_tag_written_count`: Data write operations
 - `redis_operations_total`: Redis operation count
 
+### OPC Gateway Metrics
+- `opc_write_total`: Active write attempts (NATS-driven), by result
+- `opc_write_latency_seconds`: Active write latency
+- `opc_active_read_total`: Scheduled/on-demand read attempts, by result
+- `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...)
+
 ## ⚙️ Configuration
 
 ### Environment Variables
@@ -273,6 +300,10 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 | `MONGODB_PASSWORD` | MongoDB password | `sientia` | No |
 | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
 
+See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`,
+`CERT_BUCKET`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the
+gateway-specific environment variables.
+
 ### OPC Server Configuration
 
 OPC servers are configured through Redis with the following structure. `cert_path` / `private_key_path` are only required for secured connections (see [Certificate Generation](#-certificate-generation)):

ingestor/adapters/grpc_server.py

@@ -17,7 +17,9 @@ class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer):
         self.active_read = active_read
         self.logger = logger
         # ponytail: slot_lookup wiring (address -> names from the current slot)
-        # lands in Task 18 (app wiring); optional callable keeps this servicer usable now.
+        # lands at the SIENTIAPDE-1974 integration -- `names` only exist in slot v2,
+        # which the Orchestrator produces after 1974. Until then ReadTag never
+        # persists (persisted=False); optional callable keeps this servicer usable now.
         self.slot_lookup = slot_lookup
 
     async def ReadTag(self, request, context):

ingestor/app.py

@@ -7,7 +7,16 @@ from time import time
 
 from prometheus_client import start_http_server
 
+try:  # optional: local dev reads .env; in-cluster env comes from Helm (no .env file)
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:
+    pass
+
 import ingestor.metrics as metrics
+from ingestor.config import GatewayConfig
+from ingestor.gateway_runtime import GatewayRuntime
 from ingestor.ingestor import Ingestor
 
 exit_signal = Event()
@@ -48,10 +57,36 @@ async def main():  # NOSONAR
         exit_signal.set()
     ingestor.logger.info('Ingestor prepared. Starting main loop.')
 
+    # Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor
+    # got far enough to have a data_manager (needed by CertProvider/DataSink) --
+    # skipped entirely if prepare failed before that point. Construction can hit
+    # the network (MinIO bucket check, Postgres engine): a failure there must not
+    # kill the read path, so the app continues without the gateway runtime.
+    runtime: GatewayRuntime | None = None
+    if ingestor.ingestor_manager is not None:
+        try:
+            runtime = GatewayRuntime(
+                config=GatewayConfig.from_env(),
+                data_manager=ingestor.ingestor_manager.data_manager,
+                logger=ingestor.logger,
+                notification_handler=ingestor.notification_handler,
+                metrics_controller=ingestor.metrics_controller,
+                metadata=ingestor.metadata,
+            )
+            await runtime.start()
+        except Exception as e:
+            runtime = None
+            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()
+            ingestor.logger.error(
+                f'GatewayRuntime failed to initialize, continuing without gateway: {e}'
+            )
+
     while not exit_signal.is_set():  # NOSONAR
         start_time = time()  # Start loop timer
         try:
             await ingestor.loop()
+            if runtime is not None:
+                await runtime.idle_sweep()
             metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc()  # Increment loop counter
 
             # Use asyncio.sleep instead of exit_signal.wait for better async compatibility
@@ -70,6 +105,8 @@ async def main():  # NOSONAR
             duration = time() - start_time
             metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration)
 
+    if runtime is not None:
+        await runtime.stop()
     await ingestor.shutdown()
     metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
 

ingestor/config.py

@@ -0,0 +1,46 @@
+"""Gateway-specific env config (NATS/gRPC/certs/write flags).
+
+Separate from ingestor.py's inline os.getenv (Redis/Mongo/TTLs) — this module
+covers only the new gateway wiring consumed by app.py in Task 18.
+"""
+
+from dataclasses import dataclass
+from os import getenv
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in ('true', '1', 'yes')
+
+
+@dataclass
+class GatewayConfig:
+    nats_url: str = 'nats://localhost:4222'
+    nats_write_subject: str = 'opc.write'
+    nats_queue: str = 'gateway-writers'
+    grpc_enabled: bool = False
+    grpc_port: int = 50051
+    cert_bucket: str = 'sientia-dataops'
+    cert_cache_dir: str = '/tmp/opc-certs'  # noqa: S108 # nosec B108 - default is spec'd, overridable via env
+    write_timestamp: bool = True
+    write_quality: bool = True
+    write_min_pause_s: float = 0.0
+    idle_conn_max_s: float = 300.0
+
+    @classmethod
+    def from_env(cls) -> 'GatewayConfig':
+        return cls(
+            nats_url=getenv('NATS_URL', cls.nats_url),
+            nats_write_subject=getenv('NATS_WRITE_SUBJECT', cls.nats_write_subject),
+            nats_queue=getenv('NATS_QUEUE', cls.nats_queue),
+            grpc_enabled=_env_bool('GRPC_ENABLED', cls.grpc_enabled),
+            grpc_port=int(getenv('GRPC_PORT', cls.grpc_port)),
+            cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
+            cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
+            write_timestamp=_env_bool('WRITE_TIMESTAMP', cls.write_timestamp),
+            write_quality=_env_bool('WRITE_QUALITY', cls.write_quality),
+            write_min_pause_s=float(getenv('WRITE_MIN_PAUSE_S', cls.write_min_pause_s)),
+            idle_conn_max_s=float(getenv('IDLE_CONN_MAX_S', cls.idle_conn_max_s)),
+        )

ingestor/gateway_runtime.py

@@ -0,0 +1,156 @@
+"""GatewayRuntime (Task 18 F4 capstone): builds and owns the write/active-read/gRPC
+object graph and wires it into the app's lifecycle.
+
+Reuses the existing ingestor's logger/notification_handler/metrics_controller and
+its already-connected DataManager (so CertProvider shares the existing Mongo
+connection instead of opening a second one -- MongoDBRepository.find_one has the
+exact (collection_name, filters, metadata) signature CertProvider.fetch calls).
+
+# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974
+# integration; when swapping, build ONE SubscriptionEngine per OPC server (its
+# non_receive_count is engine-wide -- see plan ledger F1 carry-forward).
+"""
+
+from os import getenv
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.repository.minio_repository import MinioRepository
+from sientia_do.temporal.activities.postgres import Postgres
+from sientia_do.utils.connectors_config import build_postgres_config
+
+from ingestor.adapters.grpc_server import GrpcServer
+from ingestor.adapters.nats_consumer import NatsWriteConsumer
+from ingestor.config import GatewayConfig, _env_bool
+from ingestor.connection.certs import CertProvider
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.engines.active_read import ActiveReadEngine
+from ingestor.engines.write import WriteEngine
+from ingestor.io.data_sink import DataSink
+from ingestor.managers.data_manager import DataManager
+
+
+class GatewayRuntime(SientiaMonitoring):
+    """Owns the gateway write/active-read/gRPC object graph. Resilient to NATS
+    being unreachable at startup: the read path (existing subscription loop,
+    gRPC on-demand reads) must keep working even if the write path can't
+    connect yet -- so a NATS connect failure is logged/notified, not raised,
+    and retried from `idle_sweep()` on each main-loop iteration."""
+
+    def __init__(
+        self,
+        config: GatewayConfig,
+        data_manager: DataManager,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        metadata: dict | None = None,
+    ):
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+        self.config = config
+        self.metadata = metadata or {}
+        self._nats_started = False
+        self._nats_notified = False  # notify once per outage, not per 5s retry
+
+        minio = MinioRepository(
+            endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
+            access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
+            secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
+            secure=_env_bool('MINIO_SECURE', True),
+            bucket=config.cert_bucket,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+        cert_provider = CertProvider(
+            mongo=data_manager.mongo_repository,
+            minio=minio,
+            cache_dir=config.cert_cache_dir,
+            bucket=config.cert_bucket,
+            metadata=self.metadata,
+        )
+        self.pool = OpcConnectionPool(
+            cert_provider, logger, notification_handler, metrics_controller
+        )
+
+        postgres = Postgres(
+            **build_postgres_config(),
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+        self.sink = DataSink(data_manager, postgres)
+
+        self.write_engine = WriteEngine(
+            self.pool,
+            self.sink,
+            logger,
+            notification_handler,
+            metrics_controller,
+            write_timestamp=config.write_timestamp,
+            write_quality=config.write_quality,
+            min_pause_s=config.write_min_pause_s,
+        )
+        self.nats_consumer = NatsWriteConsumer(
+            config.nats_url,
+            self.write_engine,
+            logger,
+            subject=config.nats_write_subject,
+            queue=config.nats_queue,
+        )
+        self.active_read = ActiveReadEngine(
+            self.pool, self.sink, logger, notification_handler, metrics_controller
+        )
+
+        # slot_lookup=None for now: address -> names resolution needs slot v2
+        # (1974); until then ReadTag responses just aren't persisted.
+        self.grpc_server = (
+            GrpcServer(self.active_read, logger, port=config.grpc_port, slot_lookup=None)
+            if config.grpc_enabled
+            else None
+        )
+
+    async def start(self) -> None:
+        await self._start_nats()
+        if self.grpc_server is not None:
+            await self.grpc_server.start()
+        self.logger.info(
+            f'GatewayRuntime started: nats_started={self._nats_started} '
+            f'grpc_enabled={self.grpc_server is not None}'
+        )
+
+    async def stop(self) -> None:
+        if self.grpc_server is not None:
+            await self.grpc_server.stop()
+        if self._nats_started:
+            await self.nats_consumer.stop()
+            self._nats_started = False
+        await self.pool.close_all()
+
+    async def idle_sweep(self) -> None:
+        await self.pool.release_idle(self.config.idle_conn_max_s)
+        if not self._nats_started:
+            await self._start_nats()
+
+    async def _start_nats(self) -> None:
+        if self._nats_started:
+            return
+        try:
+            await self.nats_consumer.start()
+            self._nats_started = True
+            self._nats_notified = False
+        except Exception as e:
+            self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}')
+            if self._nats_notified:
+                return
+            self._nats_notified = True
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id='GATEWAY_NATS_CONSUMER_START_FAILED',
+                message=f'Failed to start NATS write consumer: {e}',
+                block='gateway_runtime',
+                level=NotificationLevel.ERROR,
+            )

requirements-dev.txt

@@ -17,6 +17,7 @@ pytest-asyncio>=0.21.0   # Async test support (already in main requirements)
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell
 ipdb>=0.13.13            # IPython debugger
+python-dotenv>=1.0    # local dev only: app/seed read .env if present (in-cluster env comes from Helm)
 
 # gRPC code generation (regenerate ingestor/proto/*_pb2*.py) — not needed
 # at runtime: the generated stubs are committed. Kept out of requirements.txt

scripts/seed_e2e.py

@@ -0,0 +1,161 @@
+"""Seed test data for the OPC Gateway e2e validation session (SIENTIAPDE-1972).
+
+Populates:
+  1. Mongo `connectors` + `connector_tags` (each tag with `metadata: {}` — required
+     for documentation, per the feature owner).
+  2. Redis slot v1 (`slot:opc_tags:1`) so the legacy subscription path collects them.
+  3. Postgres test predictions table (with the `write_status` column) + one row the
+     NATS write test can UPDATE (correlated by timestamp + model_id).
+
+Idempotent: connectors/tags are upserted, the slot is overwritten, the table uses
+CREATE IF NOT EXISTS + ON CONFLICT DO NOTHING.
+
+Usage (from the repo root, sientia-dataops conda env):
+  python scripts/seed_e2e.py --print-only          # dry-run: show docs/SQL, touch nothing
+  python scripts/seed_e2e.py                       # seed Mongo + Redis + Postgres
+
+Env (same names the gateway/app use):
+  MONGODB_URL/MONGODB_USERNAME/MONGODB_PASSWORD/MONGODB_DATABASE
+  REDIS_HOST/REDIS_PORT  POSTGRES_HOST/PORT/USER/PASSWORD/DBNAME
+  OPC_SERVER_URL (default opc.tcp://localhost:49320 — KepServer default port)
+
+# ponytail: plain sync pymongo/redis/sqlalchemy, no sientia_do async plumbing —
+# this is a test fixture, not production code.
+"""
+
+import argparse
+import json
+import os
+import sys
+
+try:  # optional: local dev reads .env; in-cluster env comes from Helm
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:
+    pass
+
+OPC_URL = os.getenv('OPC_SERVER_URL', 'opc.tcp://localhost:49320')
+
+CONNECTOR = {
+    'connector_id': 9001,
+    'type': 'opc_ua',
+    'name': 'e2e-kepserver',
+    'url': OPC_URL,
+    'server_uri': 'urn:e2e:gateway:test',
+    'subscription_period_ms': 1000,
+    'use_security': False,  # round 1; flip after certs are in sientia-dataops
+    'status': 'Active',
+    'description': 'OPC Gateway e2e validation connector (SIENTIAPDE-1972)',
+}
+
+# metadata: {} on every tag — feature owner asked test tags to carry the field.
+TAGS = [
+    {'tag_id': 9101, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.Ramp1',
+     'tag_name': 'e2e_ramp', 'category': 'Read', 'data_type': 'float', 'unit': '-',
+     'status': 'Active', 'description': 'e2e read tag', 'metadata': {}},
+    {'tag_id': 9102, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.User1',
+     'tag_name': 'e2e_write_target', 'category': 'Write', 'data_type': 'float', 'unit': '-',
+     'status': 'Active', 'description': 'e2e write tag', 'metadata': {}},
+]
+
+# Slot v1 (legacy path the gateway still runs until SIENTIAPDE-1974 delivers slot v2).
+SLOT_V1 = {
+    'e2e-kepserver': {
+        'name': 'e2e-kepserver',
+        'server_id': '9001',
+        'url': OPC_URL,
+        'server_uri': 'urn:e2e:gateway:test',
+        'subscription_period_ms': 1000,
+        'tags': {
+            TAGS[0]['address']: {
+                'tag_name': TAGS[0]['tag_name'],
+                'frequency': '5000',
+                'topics': ['raw_e2e-gateway-test'],
+            }
+        },
+    }
+}
+
+PREDICTIONS_SQL = """
+CREATE SCHEMA IF NOT EXISTS sientia_data;
+CREATE TABLE IF NOT EXISTS sientia_data.e2e_predictions (
+    model_id text NOT NULL,
+    prediction double precision,
+    prediction_confidence double precision,
+    timestamp text NOT NULL,
+    write_status text,
+    PRIMARY KEY (model_id, timestamp)
+);
+INSERT INTO sientia_data.e2e_predictions (model_id, prediction, prediction_confidence, timestamp)
+VALUES ('9001', 42.5, 0.97, '2026-07-15T12:00:00Z')
+ON CONFLICT DO NOTHING;
+"""
+
+# The matching NATS write command for the row above (paste into `nats pub opc.write '...'`).
+NATS_CMD = {
+    'msg_id': 'e2e-0001', 'server_id': '9001', 'addr': TAGS[1]['address'],
+    'value': 42.5, 'data_type': 'float', 'tag_type': 'prediction',
+    'model_id': '9001', 'model_name': 'e2e-model',
+    'schema': 'sientia_data', 'table_name': 'e2e_predictions',
+    'timestamp': '2026-07-15T12:00:00Z',
+    'url': OPC_URL, 'use_security': False, 'server_uri': 'urn:e2e:gateway:test',
+}
+
+
+def seed_mongo():
+    from pymongo import MongoClient
+
+    url = os.getenv('MONGODB_URL', 'localhost:27017')
+    user, pwd = os.getenv('MONGODB_USERNAME', 'sientia'), os.getenv('MONGODB_PASSWORD', 'sientia')
+    db = MongoClient(f'mongodb://{user}:{pwd}@{url}')[os.getenv('MONGODB_DATABASE', 'sientia')]
+    db['connectors'].replace_one({'connector_id': CONNECTOR['connector_id']}, CONNECTOR, upsert=True)
+    for tag in TAGS:
+        db['connector_tags'].replace_one({'tag_id': tag['tag_id']}, tag, upsert=True)
+    print(f'mongo: upserted 1 connector + {len(TAGS)} tags (metadata: {{}})')
+
+
+def seed_redis():
+    import redis
+
+    r = redis.Redis(host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', '6379')))
+    r.set('slot:opc_tags:1', json.dumps(SLOT_V1))
+    print('redis: wrote slot:opc_tags:1 (v1)')
+
+
+def seed_postgres():
+    from sqlalchemy import create_engine, text
+
+    host, port = os.getenv('POSTGRES_HOST', 'localhost'), os.getenv('POSTGRES_PORT', '5432')
+    user, pwd = os.getenv('POSTGRES_USER', 'sientia'), os.getenv('POSTGRES_PASSWORD', 'sientia')
+    dbname = os.getenv('POSTGRES_DBNAME', 'sientia')
+    engine = create_engine(f'postgresql://{user}:{pwd}@{host}:{port}/{dbname}')
+    with engine.begin() as conn:
+        for stmt in filter(str.strip, PREDICTIONS_SQL.split(';')):
+            conn.execute(text(stmt))
+    print('postgres: sientia_data.e2e_predictions ready (write_status column + 1 row)')
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--print-only', action='store_true', help='show docs/SQL, touch nothing')
+    args = parser.parse_args()
+
+    if args.print_only:
+        print('connector:', json.dumps(CONNECTOR, indent=2))
+        print('tags:', json.dumps(TAGS, indent=2))
+        print('slot v1:', json.dumps(SLOT_V1, indent=2))
+        print('sql:', PREDICTIONS_SQL)
+        print('nats write cmd:', json.dumps(NATS_CMD, indent=2))
+        return 0
+
+    seed_mongo()
+    seed_redis()
+    seed_postgres()
+    print('\nNATS write command for the session:')
+    print(f"nats pub opc.write '{json.dumps(NATS_CMD)}'")
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/functional/conftest.py

@@ -1,8 +1,5 @@
 # import subprocess
 # from time import sleep
-# from typing import Generator
-# import uuid
-# from kafka import KafkaConsumer
 # import pytest
 # from redis import Redis
 
@@ -32,16 +29,3 @@
 #     # Limpa o banco de dados após os testes
 #     redis.flushdb()
 #     redis.close()
-
-
-# def kafka_searcher(topic) -> Generator[KafkaConsumer, None, None]:
-#     consumer = KafkaConsumer(
-#         topic,
-#         bootstrap_servers="localhost:9092",
-#         group_id=f"test-group-{uuid.uuid4()}",
-#         auto_offset_reset="earliest",   # Começa a consumir apenas mensagens novas
-#         enable_auto_commit=True,
-#     )
-
-#     yield consumer
-#     consumer.close()

tests/functional/test_single_node.py

@@ -2,8 +2,6 @@
 # import subprocess
 # from time import sleep
 
-# from tests.functional.conftest import kafka_searcher
-
 
 # new_data = {
 #     "slot:opc_tags:1": {
@@ -69,14 +67,6 @@
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
 
-#     # Check if data is in Kafka
-
-#     kafka = next(kafka_searcher('test_topic_1'))
-#     sleep(1)
-#     messages = kafka.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in Kafka, but got none."
-
 
 # def test_simple_double_slot(redis_client):
 
@@ -90,12 +80,6 @@
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
 
-#     # Check if data is in Kafka
-#     kafka1 = next(kafka_searcher('test_topic_double_slot1'))
-#     messages = kafka1.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in test_topic_double_slot1, but got none."
-
 #     new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [
 #         'test_topic_double_slot2']
 #     redis_client.set("slot:opc_tags:2",
@@ -107,12 +91,3 @@
 #     assert redis_client.get("lease:opc_tags:2") == 'ingestor'
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
-
-#     # Check if data is in Kafka
-#     kafka2 = next(kafka_searcher('test_topic_double_slot2'))
-#     messages = kafka2.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in test_topic_double_slot2, but got none."
-
-#     messages = kafka1.poll(timeout_ms=10000)
-#     assert messages, "Expected messages in test_topic_double_slot1, but got none."

tests/unit/test_app.py

@@ -43,6 +43,15 @@ def mock_app_env(monkeypatch):
     monkeypatch.setattr(app, 'Ingestor', mocks['Ingestor'])
     monkeypatch.setattr(asyncio, 'sleep', mocks['asyncio_sleep'])
 
+    mocks['GatewayConfig'] = MagicMock()
+    mocks['GatewayRuntime'] = MagicMock()
+    monkeypatch.setattr(app, 'GatewayConfig', mocks['GatewayConfig'])
+    monkeypatch.setattr(app, 'GatewayRuntime', mocks['GatewayRuntime'])
+    mock_runtime_instance = mocks['GatewayRuntime'].return_value
+    mock_runtime_instance.start = AsyncMock()
+    mock_runtime_instance.idle_sweep = AsyncMock()
+    mock_runtime_instance.stop = AsyncMock()
+
     monkeypatch.setattr(
         app.metrics.APP_UP,
         'labels',
@@ -77,6 +86,7 @@ def mock_app_env(monkeypatch):
     mock_ingestor_instance = mocks['Ingestor'].return_value
     mock_ingestor_instance.poll_interval = 0.01
     mock_ingestor_instance.logger = MagicMock()
+    mock_ingestor_instance.ingestor_manager = MagicMock()
 
     # Make async methods async mocks
     mock_ingestor_instance.prepare_ingestor = AsyncMock()
@@ -124,6 +134,21 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys):
     mock_ingestor_instance.shutdown.assert_called_once()
     mock_ingestor_instance.logger.info.assert_any_call('Main loop exit_signaled.')
 
+    # GatewayRuntime (Task 18): built once prepare_ingestor succeeds, started before
+    # the loop, idle-swept each iteration, stopped on shutdown.
+    mock_app_env['GatewayRuntime'].assert_called_once_with(
+        config=mock_app_env['GatewayConfig'].from_env.return_value,
+        data_manager=mock_ingestor_instance.ingestor_manager.data_manager,
+        logger=mock_ingestor_instance.logger,
+        notification_handler=mock_ingestor_instance.notification_handler,
+        metrics_controller=mock_ingestor_instance.metrics_controller,
+        metadata=mock_ingestor_instance.metadata,
+    )
+    mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value
+    mock_runtime_instance.start.assert_awaited_once()
+    mock_runtime_instance.idle_sweep.assert_awaited_once()
+    mock_runtime_instance.stop.assert_awaited_once()
+
     captured = capsys.readouterr()
     assert 'Prometheus server started on port 9090.' in captured.out
 
@@ -297,12 +322,16 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
     mock_exit_signal = mock_app_env['mock_exit_signal']
 
     mock_ingestor_instance.prepare_ingestor.side_effect = Exception('Preparation failed')
+    mock_ingestor_instance.ingestor_manager = None  # prepare failed before assignment
     mock_exit_signal.is_set.side_effect = [False, True]
 
     with pytest.raises(OsExitCalledError) as excinfo:
         await app.main()
     assert excinfo.value.code == 0
 
+    # GatewayRuntime must not be built when prepare_ingestor never got a data_manager
+    mock_app_env['GatewayRuntime'].assert_not_called()
+
     # Verify error metrics were incremented
     app.metrics.APP_ERRORS_TOTAL.labels.assert_called_with(pod_id='test_pod')
     mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once()
@@ -317,3 +346,34 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
     mock_ingestor_instance.logger.error.assert_called_once_with(
         'Failed to prepare ingestor: Preparation failed'
     )
+
+
+@pytest.mark.asyncio
+async def test_main_gateway_runtime_init_failure_keeps_read_path_alive(mock_app_env):
+    """GatewayRuntime construction hitting the network (MinIO/Postgres) and failing
+    must not kill the app: the main loop (read path) still runs and shuts down cleanly."""
+    mock_ingestor_instance = mock_app_env['Ingestor'].return_value
+    mock_exit_signal = mock_app_env['mock_exit_signal']
+
+    mock_app_env['GatewayRuntime'].side_effect = Exception('MinIO unreachable')
+    mock_exit_signal.is_set.side_effect = [False, True]
+    mock_app_env['time_time'].side_effect = [10.0, 11.0]
+
+    with pytest.raises(OsExitCalledError) as excinfo:
+        await app.main()
+    assert excinfo.value.code == 0
+
+    # Loop still ran and shutdown was clean, without any runtime calls
+    mock_ingestor_instance.loop.assert_called_once()
+    mock_ingestor_instance.shutdown.assert_called_once()
+    mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value
+    mock_runtime_instance.idle_sweep.assert_not_called()
+    mock_runtime_instance.stop.assert_not_called()
+
+    # Failure was logged and counted, but did not set exit_signal
+    assert any(
+        'GatewayRuntime failed to initialize' in str(c)
+        for c in mock_ingestor_instance.logger.error.call_args_list
+    )
+    mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once()
+    mock_exit_signal.set.assert_not_called()

tests/unit/test_config.py

@@ -0,0 +1,71 @@
+import pytest
+
+from ingestor.config import GatewayConfig
+
+_ENV_VARS = (
+    'NATS_URL',
+    'NATS_WRITE_SUBJECT',
+    'NATS_QUEUE',
+    'GRPC_ENABLED',
+    'GRPC_PORT',
+    'CERT_BUCKET',
+    'CERT_CACHE_DIR',
+    'WRITE_TIMESTAMP',
+    'WRITE_QUALITY',
+    'WRITE_MIN_PAUSE_S',
+    'IDLE_CONN_MAX_S',
+)
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    for var in _ENV_VARS:
+        monkeypatch.delenv(var, raising=False)
+
+
+def test_from_env_defaults():
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.nats_url == 'nats://localhost:4222'
+    assert cfg.nats_write_subject == 'opc.write'
+    assert cfg.nats_queue == 'gateway-writers'
+    assert cfg.grpc_enabled is False
+    assert cfg.grpc_port == 50051
+    assert cfg.cert_bucket == 'sientia-dataops'
+    assert cfg.cert_cache_dir == '/tmp/opc-certs'  # noqa: S108
+    assert cfg.write_timestamp is True
+    assert cfg.write_quality is True
+    assert cfg.write_min_pause_s == 0.0
+    assert cfg.idle_conn_max_s == 300.0
+
+
+def test_from_env_overrides(monkeypatch):
+    monkeypatch.setenv('GRPC_ENABLED', 'true')
+    monkeypatch.setenv('GRPC_PORT', '6000')
+    monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0.5')
+    monkeypatch.setenv('WRITE_QUALITY', 'false')
+
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.grpc_enabled is True
+    assert cfg.grpc_port == 6000
+    assert cfg.write_min_pause_s == 0.5
+    assert cfg.write_quality is False
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('0', False),
+        ('no', False),
+        ('FALSE', False),
+        ('1', True),
+        ('YES', True),
+    ],
+)
+def test_bool_parsing_edge_cases(monkeypatch, value, expected):
+    monkeypatch.setenv('GRPC_ENABLED', value)
+
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.grpc_enabled is expected

tests/unit/test_gateway_runtime.py

@@ -0,0 +1,172 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ingestor import gateway_runtime as gr
+from ingestor.config import GatewayConfig
+
+_PATCHED_CLASSES = (
+    'MinioRepository',
+    'CertProvider',
+    'OpcConnectionPool',
+    '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,
+    }
+    patched['OpcConnectionPool'].return_value.close_all = AsyncMock()
+    patched['OpcConnectionPool'].return_value.release_idle = AsyncMock()
+    patched['NatsWriteConsumer'].return_value.start = AsyncMock()
+    patched['NatsWriteConsumer'].return_value.stop = AsyncMock()
+    patched['GrpcServer'].return_value.start = AsyncMock()
+    patched['GrpcServer'].return_value.stop = AsyncMock()
+    return patched
+
+
+def make_runtime(**config_overrides):
+    return gr.GatewayRuntime(
+        config=GatewayConfig(**config_overrides),
+        data_manager=MagicMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+
+def test_builds_object_graph_with_config_flags(mocks):
+    runtime = make_runtime(
+        write_timestamp=False,
+        write_quality=False,
+        write_min_pause_s=0.25,
+        cert_bucket='my-bucket',
+        cert_cache_dir='certs-cache',
+        grpc_enabled=False,
+    )
+
+    mocks['OpcConnectionPool'].assert_called_once()
+    mocks['DataSink'].assert_called_once()
+
+    mocks['WriteEngine'].assert_called_once()
+    _, kwargs = mocks['WriteEngine'].call_args
+    assert kwargs['write_timestamp'] is False
+    assert kwargs['write_quality'] is False
+    assert kwargs['min_pause_s'] == 0.25
+
+    _, cert_kwargs = mocks['CertProvider'].call_args
+    assert cert_kwargs['bucket'] == 'my-bucket'
+    assert cert_kwargs['cache_dir'] == 'certs-cache'
+
+    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_and_closes_pool(mocks):
+    runtime = make_runtime(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()
+    mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_stop_skips_nats_stop_if_never_started(mocks):
+    runtime = make_runtime()
+    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['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_idle_sweep_releases_idle_connections(mocks):
+    runtime = make_runtime(idle_conn_max_s=42.0)
+    runtime._nats_started = True  # isolate from the retry path
+
+    await runtime.idle_sweep()
+
+    mocks['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0)