gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #41 from Aignosi/feature-rc/SIENTIAPDE-1974-read-v2

Browse Source 
SIENTIAPDE-1974: leitura v2 — slot v2 via SubscriptionEngine, certificado OPC do MinIO
This commit is contained in:
Eduardo Rios
2026-10-06 15:21:53 -03:00
committed by GitHub
parent 4c85a9c5c2 b2d106e5c4
commit ee02e481bd
22 changed files with 1637 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CLAUDE.md
View File

@@ -40,7 +40,9 @@ app.py (asyncio loop, signals, Prometheus server)
-> Ingestor (ingestor.py) — lifecycle: prepare_ingestor() -> loop() -> shutdown()
-> IngestorManager (managers/ingestor_manager.py) — central coordinator
-> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry
-> OpcManager (managers/opc_manager.py), one instance per OPC server — connection, subscriptions, datachange callbacks
-> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool
(READ_ENGINE=legacy: OpcManager (managers/opc_manager.py), the rc3 reader kept as plan B until the end of SIENTIAPDE-1974)
-> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO
-> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish
```
@@ -56,7 +58,7 @@ app.py (asyncio loop, signals, Prometheus server)
Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives).
**OpcManager** owns one OPC UA `Client` connection (asyncua), handles security policy setup (cert/key-based, `SecurityPolicyBasic256`), subscriptions per tag group, and `datachange_notification` callbacks that push into `data_queue` for `DataManager` to consume. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles`).
**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.

109
README.md
View File

@@ -123,9 +123,9 @@ health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in
# Ensure services are accessible on localhost with appropriate ports
```
6. **Generate an OPC UA client certificate** (only if connecting to a secured OPC UA server)
6. **Register the OPC UA client certificate** (only if connecting to a secured OPC UA server)
See [Certificate Generation](#-certificate-generation) below.
See [OPC Client Certificate](#-opc-client-certificate) below.
## Usage
@@ -193,37 +193,50 @@ pytest --cov=ingestor --cov-report=html
The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage.
## 🔐 Certificate Generation
## 🔐 OPC Client Certificate
OPC UA servers that require secure connections (e.g. KEPServer with `Basic256` policy) validate the ingestor via an X.509 client certificate. Use `scripts/generate-and-push-opc-cert.sh` to generate that certificate and push it to the Gitea repo the pod clones on boot.
OPC UA servers that require secure connections validate the gateway with a single,
shared platform X.509 client certificate. Every OPC connection — subscription reads,
NATS writes and gRPC reads — uses it with `Basic256Sha256` (SIENTIAPDE-1974).
```bash
GITEA_PASSWORD='<senha-do-gitea>' ./scripts/generate-and-push-opc-cert.sh
```
The certificate is **never** read from the repository or the image. On connect, the
`CertProvider` (`ingestor/connection/certs.py`):
1. reads the single document of the Mongo collection `platform_certificates`:
`{"client_cert": {"minio_object_name": "..."}, "private_key": {"minio_object_name": "..."}}`;
2. downloads both objects from the MinIO bucket `CERT_BUCKET` (default
`streamlit-connectors`) into `CERT_CACHE_DIR` (default `/tmp/opc-certs`).
Environment variables (all optional except `GITEA_PASSWORD`):
Requirements for a server to accept the connection:
- the server's slot config has `use_security: true` (a slot v1 server with a
`cert_path` is treated the same way; the path itself is ignored);
- the certificate's SAN URI equals the server's `server_uri` in the slot — it is sent
as the client `application_uri` (KEPServer rejects a mismatch with
`BadCertificateUriInvalid`);
- the server exposes a `Basic256Sha256` endpoint and trusts the certificate (e.g.
KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*, then restart
its runtime).
| Variable | Description | Default |
|----------|-------------|---------|
| `GITEA_URL` | Gitea in-cluster URL | `https://git.sientia.ai` |
| `GITEA_USER` | Gitea user | `gitea_admin` |
| `GITEA_PASSWORD` | Gitea password (**required**) | — |
| `GITEA_REPO` | Repo the ingestor pod clones | `gitea_admin/sientia-dataops-opc-ingestor` |
| `BRANCH` | Branch to push the cert to | `main` |
| `APP_URI` | SAN URI of the cert — **must match** the `uri` field of the corresponding document in the `OPC_servers` Mongo collection, since the ingestor uses it as `application_uri` (KEPServer rejects with `BadCertificateUriInvalid` on mismatch) | `urn:sientia:opc-ingestor` |
| `CN` | Certificate common name | `sientia-opc-ingestor` |
| `DAYS` | Certificate validity (days) | `3650` |
| `FORCE` | Set to `1` to overwrite an existing cert already pushed to the repo | unset |
MinIO is only contacted when a certificate is first needed, so a MinIO outage only
affects secured servers, which are retried on the next main-loop iteration.
What the script does:
1. Generates a self-signed RSA-2048 X.509 cert + key with `openssl`, with the SAN URI, `keyUsage`, and `extendedKeyUsage` extensions OPC UA / KEPServer require.
2. Clones the ingestor's Gitea repo, copies the cert/key (`.pem`) and a `.der` copy (for manual import) into `certs/`, commits, and pushes.
3. Prints the paths the pod will see after cloning (`/app/certs/opc-ingestor-cert.pem`, `/app/certs/opc-ingestor-key.pem`) and the next manual steps.
### Plan B: legacy read engine (`READ_ENGINE=legacy`)
After pushing, you still need to:
- Set `cert_path` / `private_key_path` (and `uri`) on the server's document in the `OPC_servers` collection so `OpcManager` picks them up (see [OPC Server Configuration](#opc-server-configuration)).
- Restart the ingestor deployment so it re-clones the repo: `kubectl -n sientia rollout restart deployment/opc-ingestor`.
- On first connection, the server rejects the new cert — trust it in the OPC UA server's console (e.g. KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*) and restart its runtime.
Until the end of SIENTIAPDE-1974 the legacy reader is kept for validation and as a
fallback. `READ_ENGINE` selects the read path (default `v2`; an invalid value falls
back to `v2` with a warning):
| `READ_ENGINE` | Reader | Slot | Certificate |
|---|---|---|---|
| `v2` (default) | `ServerReader` + `SubscriptionEngine` over the shared pool | v1 or v2 | MinIO + Mongo (`CertProvider`), `Basic256Sha256` |
| `legacy` | the rc3 `OpcManager`, unchanged, with its own connection | **v1 only** (a v2 slot is rejected and notified) | files from the repository the pod clones (`cert_path` / `private_key_path` in the slot), `Basic256` |
Writes and gRPC always use the pool (MinIO certificate), whatever the read engine.
The legacy engine needs the certificate files in the Gitea branch the pod clones;
`scripts/generate-and-push-opc-cert.sh` generates them and pushes them to
`certs/` (`GITEA_PASSWORD='<senha>' BRANCH=<branch> ./scripts/generate-and-push-opc-cert.sh`).
Each version deploy creates a new Gitea branch without `certs/`, so they have to be
copied again for the legacy engine. Remove the legacy engine (OpcManager, its tests
and the script) once it is unused and the test suite passes the same without it.
## 🧪 Testing
@@ -306,7 +319,12 @@ 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)):
OPC servers are configured by the Orchestrator through Redis, one key per slot
(`slot:opc_tags:<n>`). The gateway reads the **slot v2** structure (contract C1):
one entry per address, with one `names` item per collection / tag name / model, so
the same address can feed several models under different names. For secured
connections set `use_security: true` (the certificate comes from MinIO, see
[OPC Client Certificate](#-opc-client-certificate)):
```json
{
@@ -315,30 +333,30 @@ OPC servers are configured through Redis with the following structure. `cert_pat
"name": "server1",
"server_id": "1",
"url": "opc.tcp://localhost:4841",
"server_uri": "http://opcua-server.simulator",
"cert_path": "/app/certs/opc-ingestor-cert.pem",
"private_key_path": "/app/certs/opc-ingestor-key.pem",
"server_uri": "urn:sientia:opc-ingestor",
"use_security": true,
"subscription_period_ms": 7500,
"tags": {
"ns=2;i=2": {
"aggr_func": "avg",
"data_range": [
-100,
100
],
"frequency": "15000",
"server_id": "1",
"tag_address": "ns=2;i=2",
"tag_name": "Counter",
"topics": [
"raw_scouter-opcua-orchestrated-pipeline",
"raw_scouter-basic-sum-model"
"active_read": {"enabled": false},
"names": [
{"collection": "raw_scouter-opcua-orchestrated-pipeline", "tag_name": "Counter", "model_name": "m1", "model_id": "1"},
{"collection": "raw_scouter-basic-sum-model", "tag_name": "Contador", "model_name": "m2", "model_id": "2"}
]
}
}
}
}
}
```
The legacy **slot v1** (each tag with `tag_name` + `topics`, the server with
`cert_path`) is still accepted and converted on read (`ingestor/io/slots.py`): one
`names` item per topic, and `use_security` is true when `cert_path` is set (the path
itself is ignored). This lets the gateway and the Orchestrator be deployed in any
order. Documents written from a v1 slot have no `model_id`, exactly as before.
## 🔧 Development
### Project Structure
@@ -348,13 +366,14 @@ sientia-dataops-opc-ingestor/
│ ├── managers/ # Component managers
│ │ ├── data_manager.py # Data persistence and export
│ │ ├── ingestor_manager.py # Main coordination
│ │ ├── opc_manager.py # OPC UA server management
│ │ └── resource_manager.py # Resource coordination
│ │ ├── opc_manager.py # Legacy reader, plan B (READ_ENGINE=legacy)
│ │ ├── resource_manager.py # Resource coordination
│ │ └── server_reader.py # OPC UA server reads (SubscriptionEngine + pool)
│ ├── app.py # Main application entry point
│ ├── ingestor.py # Core ingestor logic
│ └── metrics.py # Prometheus metrics definitions
├── simulator/ # OPC simulation and testing tools
├── scripts/ # Ops scripts (e.g. OPC UA cert generation)
├── scripts/ # Ops scripts (e2e seed; legacy OPC cert generation)
├── tests/ # Test suite
└── requirements.txt # Python dependencies
```

9
ingestor/app.py
View File

@@ -58,16 +58,17 @@ async def main(): # NOSONAR
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.
# got far enough to have an ingestor_manager (its data_manager and connection
# pool are shared) -- skipped entirely if prepare failed before that point. The
# read path (subscriptions) lives in the ingestor_manager and is already running:
# a failure here only disables writes/gRPC, the app continues without the 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,
pool=ingestor.ingestor_manager.pool,
logger=ingestor.logger,
notification_handler=ingestor.notification_handler,
metrics_controller=ingestor.metrics_controller,

14
ingestor/config.py
View File

@@ -22,13 +22,25 @@ class GatewayConfig:
nats_queue: str = 'gateway-writers'
grpc_enabled: bool = False
grpc_port: int = 50051
cert_bucket: str = 'sientia-dataops'
# Legacy frontend bucket the platform already provisions (SIENTIAPDE-1974 D7);
# the design's `sientia-dataops` bucket does not exist in the platform MinIO.
cert_bucket: str = 'streamlit-connectors'
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 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(

97
ingestor/connection/certs.py
View File

@@ -1,12 +1,29 @@
import asyncio
import os
from os import getenv
from pathlib import Path
from sientia_do.repository.minio_repository import MinioRepository
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
@@ -14,16 +31,84 @@ 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:
"""Builds the MinIO client on first use. MinioRepository's constructor hits the
network (bucket check), so building it eagerly would let a MinIO outage at pod
start take the whole read path down. Built lazily, only the servers that need a
certificate fail — and they are retried on the next loop."""
def __init__(self, factory):
self._factory = factory
self._client = None
async def download_file(self, object_name: str, bucket: str, metadata: dict):
if self._client is None:
self._client = self._factory()
return await self._client.download_file(object_name, bucket, metadata)
def build_cert_provider(
mongo, config, logger, notification_handler, metrics_controller, metadata=None
) -> CertProvider:
"""CertProvider shared by the read path (IngestorManager) and the write path
(GatewayRuntime): one certificate source — MinIO + Mongo — for every connection."""
def minio_factory() -> MinioRepository:
return 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,
)
return CertProvider(
mongo=mongo,
minio=LazyMinio(minio_factory),
cache_dir=config.cert_cache_dir,
bucket=config.cert_bucket,
metadata=metadata,
)

72
ingestor/connection/pool.py
View File

@@ -26,6 +26,12 @@ class OpcConnectionPool(SientiaMonitoring):
self.cert_provider = cert_provider
self.clock = MonotonicClock()
self._connections: dict[str, tuple[Client, float]] = {}
# Active subscriptions per server. A subscribed connection only receives
# 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] = {}
@@ -41,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:
@@ -63,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},
@@ -70,13 +99,23 @@ 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
async def _connect(self, server_cfg: dict) -> Client:
client = Client(server_cfg['url'], timeout=10)
# Identify the pod on the OPC server, as the legacy OpcManager did: asyncua
# sends `name` as ApplicationName and "<description> Session<n>" as the
# session name, so each pod's session is recognizable in the server console.
client.name = self.pod_id
client.description = self.pod_id
pod_uri = self.pod_id.replace('-', ':')
client.product_uri = pod_uri
client.application_uri = pod_uri
server_uri = server_cfg.get('server_uri')
if server_uri:
# Must match the certificate's SAN URI (KEPServer: BadCertificateUriInvalid).
client.application_uri = server_uri
if server_cfg.get('use_security'):
cert, key = await self.cert_provider.fetch()
@@ -96,15 +135,39 @@ class OpcConnectionPool(SientiaMonitoring):
tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url},
)
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
else:
self._pins.pop(server_id, None)
async def release_idle(self, max_idle_s: float):
"""Disconnect and drop clients unused for longer than `max_idle_s`."""
"""Disconnect and drop clients unused for longer than `max_idle_s`,
except the ones pinned by an active subscription."""
now = self.clock.elapsed()
idle_ids = [
sid for sid, (_, last_used) in self._connections.items() if now - last_used > max_idle_s
sid
for sid, (_, last_used) in self._connections.items()
if now - last_used > max_idle_s and not self._pins.get(sid)
]
for server_id in idle_ids:
await self._evict(server_id)
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)
async def close_all(self):
"""Disconnect every pooled client (e.g. on shutdown)."""
# list() snapshots the keys: _evict() pops from self._connections,
@@ -114,6 +177,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,
@@ -122,7 +186,7 @@ class OpcConnectionPool(SientiaMonitoring):
async def _disconnect_with_retries(self, client: Client, server_id: str) -> None:
"""Tries 5 times to disconnect, with a delay of 100ms x try (mirrors
OpcManager.disconnection_fallback)."""
the legacy OpcManager.disconnection_fallback)."""
for i in range(5):
try:
await client.disconnect()

35
ingestor/engines/subscription.py
View File

@@ -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__(
@@ -57,6 +60,7 @@ class SubscriptionEngine(SientiaMonitoring):
raise
self._clients[name] = client
self._servers[name] = server_id
self.pool.pin(server_id)
self.logger.info(f'Subscription {name} created on {server_id}.')
await self.emit_metric(
metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
@@ -74,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)
@@ -96,11 +101,35 @@ class SubscriptionEngine(SientiaMonitoring):
if not self.subscriptions.get(subscription):
self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
return
await self.subscriptions[subscription].delete()
del self.subscriptions[subscription]
sub = self.subscriptions.pop(subscription)
self._clients.pop(subscription, None)
self._servers.pop(subscription, None)
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:
# The connection may already be gone (lost server); the subscription is
# 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

42
ingestor/gateway_runtime.py
View File

@@ -2,30 +2,25 @@
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).
its already-connected DataManager. The OpcConnectionPool (and its CertProvider,
MinIO + Mongo) belongs to the read path in IngestorManager (SIENTIAPDE-1974):
reads start in prepare_ingestor, before this runtime exists, and must keep running
if anything here fails. The runtime only borrows the pool for writes and gRPC.
# 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).
# ponytail: scheduled_tick + gRPC slot_lookup wiring land in the next 1974 RC.
"""
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.config import GatewayConfig
from ingestor.connection.pool import OpcConnectionPool
from ingestor.engines.active_read import ActiveReadEngine
from ingestor.engines.write import WriteEngine
@@ -44,6 +39,7 @@ class GatewayRuntime(SientiaMonitoring):
self,
config: GatewayConfig,
data_manager: DataManager,
pool: OpcConnectionPool,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
@@ -55,26 +51,7 @@ class GatewayRuntime(SientiaMonitoring):
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
)
self.pool = pool # owned by IngestorManager (read path); closed there
postgres = Postgres(
**build_postgres_config(),
@@ -128,7 +105,8 @@ class GatewayRuntime(SientiaMonitoring):
if self._nats_started:
await self.nats_consumer.stop()
self._nats_started = False
await self.pool.close_all()
# The pool is not closed here: IngestorManager.shutdown() owns it and runs
# after this, once its readers have dropped their subscriptions.
async def idle_sweep(self) -> None:
await self.pool.release_idle(self.config.idle_conn_max_s)

19
ingestor/io/data_sink.py
View File

@@ -21,15 +21,16 @@ class DataSink:
await self.data_manager.emit_metric(**kwargs)
async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None:
"""Publish one MongoDB document per name entry and record its tag value metric."""
"""Publish one MongoDB document per name entry and record its tag value metric.
`model_id` is only written when the name carries one: names converted from a
slot v1 have none, and their documents must stay identical to the legacy ones."""
for name in names:
doc = {
'tag': tag,
'name': name['tag_name'],
'model_id': name['model_id'],
'timestamp': timestamp,
'value': value,
}
doc = {'tag': tag, 'name': name['tag_name']}
if 'model_id' in name:
doc['model_id'] = name['model_id']
doc['timestamp'] = timestamp
doc['value'] = value
await self.data_manager.publish(name['collection'], doc)
await self.emit_metric(
metric_object=metrics.OPC_TAG_VALUE,
@@ -38,7 +39,7 @@ class DataSink:
tags={
'pod_id': self.data_manager.pod_id,
'tag_name': name['tag_name'],
'model_id': name['model_id'],
'model_id': name.get('model_id', ''),
},
)

38
ingestor/io/slots.py
View File

@@ -1,25 +1,47 @@
"""Parse the slot v2 structure written by the Orchestrator to Redis (contract C1)."""
"""Parse the slot written by the Orchestrator to Redis into the slot v2 structure (contract C1).
Both shapes are accepted so the gateway and the Orchestrator can be deployed in
any order (SIENTIAPDE-1974):
- v2: each tag carries `names` (one entry per collection / tag name / model);
- v1 (legacy): each tag carries `tag_name` + `topics`, and the server carries the
file-based `cert_path`. v1 is converted to v2: one `names` entry per topic, and
`use_security` is true when the server had a certificate path (the certificate
itself now always comes from MinIO via CertProvider, never from that path).
"""
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]:
names = tag.get('names')
if names:
return names
topics = tag.get('topics')
if topics and tag.get('tag_name'):
return [{'collection': topic, 'tag_name': tag['tag_name']} for topic in topics]
raise ValueError(f'tag {address!r} has no names')
def _parse_tag(address: str, tag: dict) -> dict:
names = tag.get('names')
if not names:
raise ValueError(f'tag {address!r} has no names')
return {
'frequency': tag['frequency'],
'names': names,
'names': _parse_names(address, tag),
'active_read': tag.get('active_read', {'enabled': False}),
}
def _parse_server(server: dict) -> dict:
def _parse_server(name: str, server: dict) -> dict:
parsed = {field: server[field] for field in SERVER_FIELDS if field in server}
parsed['name'] = server.get('name', name)
if 'use_security' not in parsed:
parsed['use_security'] = bool(server.get('cert_path'))
parsed['tags'] = {addr: _parse_tag(addr, tag) for addr, tag in server['tags'].items()}
return parsed
def parse_slot(raw: dict) -> dict:
"""Normalize a raw slot v2 dict into `{server_name: {..., tags: {addr: tag}}}`."""
return {name: _parse_server(server) for name, server in raw.items()}
"""Normalize a raw slot (v1 or v2) into `{server_name: {..., tags: {addr: tag}}}`."""
return {name: _parse_server(name, server) for name, server in raw.items()}

208
ingestor/managers/ingestor_manager.py
View File

@@ -1,5 +1,6 @@
import traceback
from copy import deepcopy
from os import getenv
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
@@ -8,9 +9,21 @@ from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
import ingestor.metrics as metrics
from ingestor.config import GatewayConfig
from ingestor.connection.certs import build_cert_provider
from ingestor.connection.pool import OpcConnectionPool
from ingestor.io.data_sink import DataSink
from ingestor.io.slots import parse_slot
from ingestor.managers.data_manager import DataManager
from ingestor.managers.opc_manager import OpcManager
from ingestor.managers.resource_manager import ResourceManager
from ingestor.managers.server_reader import ServerReader
# READ_ENGINE (SIENTIAPDE-1974): 'v2' = ServerReader/SubscriptionEngine over the shared
# pool, certificate from MinIO (default). 'legacy' = plan B: the untouched rc3
# OpcManager, slot v1 only, certificate files from the Gitea clone (cert_path). Kept
# for validation until the end of 1974; remove once unused and the tests pass without it.
READ_ENGINES = ('v2', 'legacy')
class IngestorManager(SientiaMonitoring):
@@ -47,7 +60,10 @@ class IngestorManager(SientiaMonitoring):
Attributes:
data_manager (DataManager): Manages data persistence
opc_managers (dict): Dictionary of OPC managers keyed by server name
opc_managers (dict): ServerReader per OPC server, keyed by server name
pool (OpcConnectionPool): OPC connections shared by reads, writes and gRPC;
owned here because reads start in prepare_ingestor, before GatewayRuntime
sink (DataSink): read egress (fan-out to every slot v2 `names` entry)
resource_manager (ResourceManager): Manages Redis-based resource coordination
number_of_slots (int): Total number of slots configured in the system
poll_interval (int): Main loop polling interval
@@ -68,6 +84,8 @@ class IngestorManager(SientiaMonitoring):
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
gateway_config: GatewayConfig | None = None,
read_engine: str | None = None,
):
redis_host: str = redis_data['host']
redis_port: int = int(redis_data['port'])
@@ -90,6 +108,26 @@ class IngestorManager(SientiaMonitoring):
metrics_controller=metrics_controller,
)
self.opc_managers: dict = {}
# 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.certs_from_env()
self.cert_provider = build_cert_provider(
mongo=self.data_manager.mongo_repository,
config=config,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
metadata=metadata,
)
self.pool = OpcConnectionPool(
self.cert_provider, logger, notification_handler, metrics_controller
)
self.sink = DataSink(self.data_manager, None)
self._server_cert_warned: set[str] = set()
self.read_engine = self._resolve_read_engine(
read_engine or getenv('READ_ENGINE') or 'v2', logger
)
self.resource_manager = ResourceManager(
host=redis_host,
port=redis_port,
@@ -109,36 +147,19 @@ class IngestorManager(SientiaMonitoring):
self.metadata = metadata
async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
"""
Initializes an OPC Manager instance using the provided server configuration.
@staticmethod
def _resolve_read_engine(value: str, logger: Logger) -> str:
engine = value.strip().lower()
if engine not in READ_ENGINES:
logger.warning(f'Invalid READ_ENGINE {value!r}, using v2 ({"|".join(READ_ENGINES)})')
engine = 'v2'
logger.info(f'OPC read engine: {engine}')
return engine
This method creates and configures an OPC Manager for a specific OPC UA server,
establishing the connection and preparing it for tag subscriptions.
Args:
server_config (dict): A dictionary containing the OPC server configuration.
Expected keys include:
- 'name' (str): The name of the OPC server.
- 'url' (str): The URL of the OPC server.
- 'server_uri' (str): The URI of the OPC server.
- 'cert_path' (str, optional): Path to the client certificate file.
- 'private_key_path' (str, optional): Path to the private key file.
- 'server_cert_path' (str, optional): Path to the server certificate file.
Returns:
OpcManager | None: An initialized OpcManager instance if successful,
otherwise None if an error occurs during initialization.
Raises:
Exception: If OPC manager initialization fails, the error is logged and
a notification is sent, but the method returns None to allow
the system to continue operating with other servers.
"""
try:
self.logger.info(f'Initializing OpcManager at {server_config["url"]}')
manager = OpcManager(
def _build_reader(self, server_config: dict) -> ServerReader | OpcManager:
if self.read_engine == 'legacy':
# Plan B: the rc3 OpcManager, unchanged (own connection, cert files).
return OpcManager(
name=server_config['name'],
url=server_config['url'],
subscription_period_ms=server_config['subscription_period_ms'],
@@ -152,6 +173,41 @@ class IngestorManager(SientiaMonitoring):
server_cert_path=server_config.get('server_cert_path'),
metrics_controller=self.metrics_controller,
)
return ServerReader(
name=server_config['name'],
server_config=server_config,
pool=self.pool,
sink=self.sink,
logger=self.logger,
notification_handler=self.notification_handler,
metrics_controller=self.metrics_controller,
metadata=self.metadata,
)
async def initialize_opc_from_config(
self, server_config: dict
) -> ServerReader | OpcManager | None:
"""
Initializes the reader of one OPC UA server from its slot configuration.
With READ_ENGINE=v2 (default) creates a ServerReader and opens its pooled
connection; with `use_security`, the certificate comes from MinIO. With
READ_ENGINE=legacy (plan B) creates the rc3 OpcManager, which connects on its
own with the certificate files of the slot v1 (`cert_path`).
Args:
server_config (dict): The server configuration of the slot, without the
'tags' key. v2: 'name', 'server_id', 'url', 'server_uri',
'use_security', 'subscription_period_ms'. legacy: the slot v1 fields.
Returns:
ServerReader | OpcManager | None: The connected reader, or None if the
connection fails (logged and notified, so the other servers keep working).
"""
try:
self.logger.info(f'Initializing OPC reader at {server_config["url"]}')
manager = self._build_reader(server_config)
manager.config = server_config
await manager.connect()
@@ -176,23 +232,33 @@ class IngestorManager(SientiaMonitoring):
Gracefully shuts down the IngestorManager and all its components.
This method ensures proper cleanup of:
- All OPC manager instances and their connections
- All OPC readers, their subscriptions and the pooled connections
- Data manager connections and resources
- Active subscriptions and server connections
The shutdown process is performed asynchronously to allow proper cleanup
of all managed resources before termination.
"""
for _server_name, server in self.opc_managers.items():
await server.shutdown()
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:
if self.read_engine == 'legacy':
# OpcManager owns its connection: shutdown always disconnects it.
await self.opc_managers[server].shutdown()
else:
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)
@@ -258,9 +324,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,
@@ -419,14 +486,18 @@ class IngestorManager(SientiaMonitoring):
Notes:
- If a slot is leased but its details cannot be retrieved
(i.e., `get_tag_slot` returns None), that slot is skipped.
(i.e., `get_tag_slot` returns None) or are malformed, that slot is skipped.
- Slot details are normalized to the slot v2 structure (parse_slot).
"""
acquired = {}
for i in range(1, self.number_of_slots + 1):
if await self.resource_manager.lease_tag(str(i)):
self.logger.info(f'Leased slot {i}')
slots = await self.resource_manager.get_tag_slot(str(i))
raw = await self.resource_manager.get_tag_slot(str(i))
if raw is None:
continue
slots = self.parse_slot_config(str(i), raw)
if slots is None:
continue
acquired[str(i)] = slots
@@ -518,16 +589,75 @@ class IngestorManager(SientiaMonitoring):
removed_slots: list[str] = []
for slot, _slot_config in self.managed_tags.items():
await self.resource_manager.renew_tag_lease(slot)
update = await self.resource_manager.get_tag_slot(slot)
if update is None:
raw = await self.resource_manager.get_tag_slot(slot)
if raw is None:
removed_slots.append(slot)
continue
update = self.parse_slot_config(slot, raw)
if update is None:
# Malformed config: keep reading with the last good one.
continue
self.managed_tags[slot] = update
for slot in removed_slots:
self.managed_tags.pop(slot, None)
def parse_slot_config(self, slot: str, raw: dict) -> dict | None:
"""
Normalizes the raw slot read from Redis to the slot v2 structure.
Accepts slot v1 (`topics`) and v2 (`names`), so the gateway and the
Orchestrator can be deployed in any order (SIENTIAPDE-1974). With
READ_ENGINE=legacy the raw slot v1 is kept as is (what OpcManager reads), and
a slot v2 is rejected.
Returns:
dict | None: The parsed slot, or None when the config is malformed
(logged and notified).
"""
try:
if self.read_engine == 'legacy':
self._validate_slot_v1(raw)
return raw
parsed = parse_slot(raw)
except (AttributeError, KeyError, TypeError, ValueError) as e:
self.send_notification(
metadata=self.metadata,
notification_id=f'OPC_SLOT_CONFIG_ERROR_{slot}',
message=f'Slot {slot} has an invalid configuration: {e!r}',
block='ingestor_manager',
level=NotificationLevel.ERROR,
)
self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}')
return None
self._warn_server_cert_path(raw)
return parsed
@staticmethod
def _validate_slot_v1(raw: dict) -> None:
"""READ_ENGINE=legacy: OpcManager only understands slot v1 (`tag_name` +
`topics` per tag); a v2 slot (`names`) would fail in its datachange callback."""
for name, server in raw.items():
for address, tag in server['tags'].items():
if 'topics' not in tag or 'tag_name' not in tag or 'frequency' not in tag:
raise ValueError(
f'READ_ENGINE=legacy needs a slot v1 (tag_name + topics); '
f'server {name!r}, tag {address!r} is not'
)
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:
"""
Releases the leases associated with the specified slot IDs.

107
ingestor/managers/server_reader.py Normal file
View File

@@ -0,0 +1,107 @@
"""Per-server read path driven by the IngestorManager (SIENTIAPDE-1974).
Replaces the legacy OpcManager behind the same interface the IngestorManager
already uses (connect / create_subscription / subscribe / unsubscribe /
check_cycles / check_opc_listenning / shutdown), so the slot lease loop is
unchanged. The difference is underneath:
- the connection comes from the shared OpcConnectionPool, so the certificate is
fetched from MinIO by CertProvider (Basic256Sha256) — never from a file shipped
in the Gitea repository;
- readings go through SubscriptionEngine, which fans each value out to every
`names` entry of the slot v2 (one document per collection / tag name / model).
"""
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
import ingestor.metrics as metrics
from ingestor.connection.pool import OpcConnectionPool
from ingestor.engines.subscription import SubscriptionEngine
from ingestor.io.data_sink import DataSink
from ingestor.io.slots import CONNECTION_FIELDS
class ServerReader:
"""Reads one OPC server: one SubscriptionEngine (its `non_receive_count` is
engine-wide, hence one per server) with one subscription per leased slot."""
def __init__(
self,
name: str,
server_config: dict,
pool: OpcConnectionPool,
sink: DataSink,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
metadata: dict | None = None,
):
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.pool = pool
self.logger = logger
self.config: dict | None = None
self.engine = SubscriptionEngine(
pool, sink, logger, notification_handler, metrics_controller, metadata
)
def __str__(self):
return f'ServerReader(name={self.name}, server_id={self.server_id}, url={self.url})'
@property
def subscriptions(self) -> dict:
return self.engine.subscriptions
async def connect(self):
"""Open (or reuse) the pooled connection; raises if the server is unreachable
or the certificate can't be fetched, like OpcManager.connect did."""
await self.pool.get(self.server_id, self.server_cfg)
async def create_subscription(self, slot: str):
await self.engine.create_subscription(self.server_id, self.server_cfg, slot)
async def subscribe(self, slot: str, nodes: dict, collect_period: int):
await self.engine.subscribe(slot, nodes, collect_period)
async def unsubscribe(self, slot: str):
await self.engine.unsubscribe(slot)
async def check_cycles(self):
await self.engine.check_cycles()
async def check_opc_listenning(self) -> bool:
return await self.engine.check_opc_listenning(self.server_id)
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."""
# list() snapshots the keys: unsubscribe() pops from engine.subscriptions, so
# iterating the dict directly would raise "changed size during iteration".
for slot in list(self.engine.subscriptions): # NOSONAR
await self.engine.unsubscribe(slot)
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',
value=0,
tags={'pod_id': self.engine.pod_id, 'server_name': self.server_id},
)

120
tests/unit/connection/test_certs.py
View File

@@ -26,3 +26,123 @@ async def test_fetch_returns_local_paths(tmp_path):
)
assert (tmp_path / 'client_cert_c.pem').read_bytes() == b'cert-bytes'
assert (tmp_path / 'private_key_k.pem').read_bytes() == b'key-bytes'
@pytest.mark.asyncio
async def test_lazy_minio_builds_client_once_on_first_download():
from unittest.mock import MagicMock
from ingestor.connection.certs import LazyMinio
client = AsyncMock()
client.download_file.return_value = b'bytes'
factory = MagicMock(return_value=client)
minio = LazyMinio(factory)
factory.assert_not_called() # no network at construction
assert await minio.download_file('a.pem', 'bucket', {}) == b'bytes'
await minio.download_file('b.pem', 'bucket', {})
factory.assert_called_once()
assert client.download_file.await_count == 2
def test_build_cert_provider_is_lazy_and_uses_config(monkeypatch):
from unittest.mock import MagicMock
from ingestor.config import GatewayConfig
from ingestor.connection import certs
minio_cls = MagicMock()
monkeypatch.setattr(certs, 'MinioRepository', minio_cls)
mongo = MagicMock()
provider = certs.build_cert_provider(
mongo=mongo,
config=GatewayConfig(cert_bucket='streamlit-connectors', cert_cache_dir='certs-cache'),
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=MagicMock(),
)
assert provider.mongo is mongo
assert provider.bucket == 'streamlit-connectors'
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'))

160
tests/unit/connection/test_pool.py
View File

@@ -182,3 +182,163 @@ async def test_concurrent_get_same_server_connects_once():
assert c1 is c2 is clients[0]
assert ctor.call_count == 1
assert clients[0].connect.await_count == 1
@pytest.mark.asyncio
async def test_release_idle_keeps_pinned_connection():
"""A subscribed connection only gets callbacks (never get()), so it looks idle:
the pin keeps release_idle from dropping it (SIENTIAPDE-1974 G4)."""
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})
pool.pin('srv')
await pool.release_idle(max_idle_s=0)
assert 'srv' in pool._connections
fake_client.disconnect.assert_not_awaited()
@pytest.mark.asyncio
async def test_release_idle_evicts_after_last_unpin():
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})
pool.pin('srv')
pool.pin('srv') # two slots subscribed on the same server
pool.unpin('srv')
await pool.release_idle(max_idle_s=0)
assert 'srv' in pool._connections # one subscription still active
pool.unpin('srv')
await pool.release_idle(max_idle_s=0)
assert 'srv' not in pool._connections
fake_client.disconnect.assert_awaited_once()
def test_unpin_never_goes_negative():
pool = make_pool()
pool.unpin('srv')
pool.pin('srv')
assert pool._pins == {'srv': 1}
@pytest.mark.asyncio
async def test_evict_disconnects_now_and_next_get_reconnects():
pool = make_pool()
first, second = AsyncMock(), AsyncMock()
with patch('ingestor.connection.pool.Client', side_effect=[first, second]):
await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
await pool.evict('srv')
client = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
first.disconnect.assert_awaited_once()
assert client is second
@pytest.mark.asyncio
async def test_evict_unknown_server_is_noop():
pool = make_pool()
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()
@pytest.mark.asyncio
async def test_connect_identifies_the_pod_on_the_server():
"""Session/application name = pod id, as the legacy OpcManager did, so each pod's
session is recognizable in the KEPServer console."""
pool = make_pool()
fake_client = MagicMock(connect=AsyncMock())
with patch('ingestor.connection.pool.Client', return_value=fake_client):
await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
assert fake_client.name == pool.pod_id
assert fake_client.description == pool.pod_id
assert fake_client.product_uri == pool.pod_id.replace('-', ':')
assert fake_client.application_uri == pool.pod_id.replace('-', ':')
@pytest.mark.asyncio
async def test_connect_server_uri_still_wins_for_application_uri():
pool = make_pool()
fake_client = MagicMock(connect=AsyncMock())
with patch('ingestor.connection.pool.Client', return_value=fake_client):
await pool.get(
'srv',
{
'url': 'opc.tcp://h:4840',
'use_security': False,
'server_uri': 'urn:sientia:opc-ingestor',
},
)
# must match the certificate SAN URI (BadCertificateUriInvalid otherwise)
assert fake_client.application_uri == 'urn:sientia:opc-ingestor'
assert fake_client.name == pool.pod_id

97
tests/unit/engines/test_subscription.py
View File

@@ -201,3 +201,100 @@ async def test_check_opc_listenning_triggers_reconnect_at_15():
assert result is True
assert eng.send_notification_async.call_count == 2
@pytest.mark.asyncio
async def test_create_subscription_pins_pool_connection():
eng = make_engine()
eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock()))
await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
eng.pool.pin.assert_called_once_with('1')
@pytest.mark.asyncio
async def test_create_subscription_failure_does_not_pin():
eng = make_engine()
client = AsyncMock()
client.create_subscription.side_effect = RuntimeError('boom')
eng.pool = MagicMock(get=AsyncMock(return_value=client))
with pytest.raises(RuntimeError):
await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
eng.pool.pin.assert_not_called()
@pytest.mark.asyncio
async def test_unsubscribe_unpins_pool_connection():
eng = make_engine()
eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock()))
await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
await eng.unsubscribe('slot1')
eng.pool.unpin.assert_called_once_with('1')
assert 'slot1' not in eng.subscriptions
@pytest.mark.asyncio
async def test_unsubscribe_tolerates_dead_connection():
"""A lost server's subscription can't be deleted on the server side; it must
still be dropped locally (and unpinned) so the slot can be re-subscribed."""
eng = make_engine()
sub = AsyncMock()
sub.delete.side_effect = ConnectionError('closed')
client = AsyncMock()
client.create_subscription.return_value = sub
eng.pool = MagicMock(get=AsyncMock(return_value=client))
await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
await eng.unsubscribe('slot1') # must not raise
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'}

23
tests/unit/io/test_data_sink_publish.py
View File

@@ -67,3 +67,26 @@ async def test_publish_reading_emits_tag_value_metric_per_name():
value=42,
tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'},
)
@pytest.mark.asyncio
async def test_publish_reading_without_model_id_keeps_legacy_doc():
"""Names converted from a slot v1 carry no model_id: the document must be the
same as the legacy OpcManager one (SIENTIAPDE-1974 NF2)."""
sink, dm = make_sink()
names = [{'collection': 'raw_x', 'tag_name': 'Counter'}]
await sink.publish_reading(
tag='ns=2;i=2', value=42, timestamp='2026-07-13 00:00:00+0000', names=names
)
collection, doc = dm.publish.await_args_list[0].args
assert collection == 'raw_x'
assert doc == {
'tag': 'ns=2;i=2',
'name': 'Counter',
'timestamp': '2026-07-13 00:00:00+0000',
'value': 42,
}
_, kwargs = sink.emit_metric.await_args
assert kwargs['tags']['model_id'] == ''

99
tests/unit/io/test_slots.py
View File

@@ -62,3 +62,102 @@ def test_parse_slot_rejects_tag_without_names():
with pytest.raises((KeyError, ValueError)):
parse_slot(raw)
RAW_V1 = {
'kepserver': {
'name': 'kepserver',
'server_id': '1',
'url': 'opc.tcp://kepserver:49320',
'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,
'subscription_period_ms': 15000.0,
'tags': {
'ns=2;s=A.TargetSin': {
'aggr_func': 'lts',
'frequency': 30000,
'server_id': '1',
'tag_address': 'ns=2;s=A.TargetSin',
'tag_name': 'TargetSin',
'topics': ['raw_scouter-sin-approx', 'raw_other-model'],
}
},
}
}
def test_parse_slot_converts_v1_topics_to_names():
"""Slot v1 (current Orchestrator) is read as v2: one name per topic, no model_id,
so the documents stay identical to the legacy ones (SIENTIAPDE-1974 G1)."""
server = parse_slot(RAW_V1)['kepserver']
tag = server['tags']['ns=2;s=A.TargetSin']
assert tag['names'] == [
{'collection': 'raw_scouter-sin-approx', 'tag_name': 'TargetSin'},
{'collection': 'raw_other-model', 'tag_name': 'TargetSin'},
]
assert tag['frequency'] == 30000
assert tag['active_read'] == {'enabled': False}
def test_parse_slot_v1_cert_path_means_security_but_path_is_dropped():
server = parse_slot(RAW_V1)['kepserver']
assert server['use_security'] is True
assert server['name'] == 'kepserver'
assert server['server_id'] == '1'
assert server['subscription_period_ms'] == 15000.0
# the certificate always comes from MinIO now: no file path survives the parse
assert 'cert_path' not in server
assert 'private_key_path' not in server
def test_parse_slot_v1_without_cert_path_is_insecure():
raw = {
's': {
'server_id': '2',
'url': 'u',
'cert_path': None,
'tags': {'ns=2;i=1': {'tag_name': 't', 'frequency': '1000', 'topics': ['raw_c']}},
}
}
server = parse_slot(raw)['s']
assert server['use_security'] is False
assert server['name'] == 's' # falls back to the slot key
def test_parse_slot_v2_explicit_use_security_wins():
raw = {
's': {
'server_id': '1',
'url': 'u',
'use_security': False,
'cert_path': '/legacy/path',
'tags': {
'ns=2;i=1': {
'frequency': '1000',
'names': [{'collection': 'c', 'tag_name': 't', 'model_id': '1'}],
}
},
}
}
assert parse_slot(raw)['s']['use_security'] is False
def test_parse_slot_rejects_v1_tag_without_tag_name():
raw = {
's': {
'server_id': '1',
'url': 'u',
'tags': {'ns=2;i=1': {'frequency': '1', 'topics': ['c']}},
}
}
import pytest
with pytest.raises(ValueError):
parse_slot(raw)

378
tests/unit/managers/test_ingestor_manager.py
View File

@@ -4,6 +4,9 @@ from unittest.mock import AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
from ingestor.config import GatewayConfig
from ingestor.connection.pool import OpcConnectionPool
from ingestor.io.slots import parse_slot
from ingestor.managers.ingestor_manager import IngestorManager
metadata = {
@@ -16,6 +19,21 @@ metadata = {
}
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
@@ -52,12 +70,12 @@ def ingestor_manager(data_manager_mock, resource_manager_mock):
return ingestor
@patch('ingestor.managers.ingestor_manager.OpcManager')
@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, opc_manager_mock
notification_handler_mock, resource_manager_mock, data_manager_mock, server_reader_mock
):
ingestor = IngestorManager(
redis_data={'host': 'localhost', 'port': 6379},
@@ -72,7 +90,7 @@ def test___init__(
metrics_controller=MagicMock(),
)
opc_manager_mock.assert_not_called()
server_reader_mock.assert_not_called()
data_manager_mock.assert_called_once_with(
mongo_connection_string='mongodb://localhost:27017',
mongo_database='sientia',
@@ -100,59 +118,86 @@ def test___init__(
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
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.OpcManager')
async def test_initialize_opc_from_config(opc_manager, ingestor_manager):
server_config = {
'name': 'server1',
'url': 'opc.tcp://localhost:4840',
'subscription_period_ms': 1000,
'server_uri': 'http://opcua-server.simulator',
'cert_path': '/path/to/cert',
'private_key_path': '/path/to/private_key',
'server_cert_path': '/path/to/server_cert',
'pod_id': 'test_pod',
}
opc_manager.return_value = MagicMock(connect=AsyncMock())
result = await ingestor_manager.initialize_opc_from_config(server_config)
opc_manager.assert_called_once_with(
name=server_config['name'],
url=server_config['url'],
data_manager=ingestor_manager.data_manager,
logger=ingestor_manager.logger,
subscription_period_ms=server_config['subscription_period_ms'],
server_uri=server_config['server_uri'],
notification_handler=ingestor_manager.notification_handler,
cert_path=server_config['cert_path'],
private_key_path=server_config['private_key_path'],
server_cert_path=server_config['server_cert_path'],
@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=ingestor_manager.metrics_controller,
metrics_controller=MagicMock(),
gateway_config=GatewayConfig(cert_bucket='my-bucket', cert_cache_dir='certs-cache'),
)
assert result == opc_manager.return_value
result.connect.assert_called_once()
assert ingestor.cert_provider.bucket == 'my-bucket'
assert ingestor.cert_provider.cache_dir == 'certs-cache'
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.OpcManager')
@patch('ingestor.managers.ingestor_manager.traceback')
async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager):
@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,
'server_uri': 'http://opcua-server.simulator',
'cert_path': '/path/to/cert',
'private_key_path': '/path/to/private_key',
'server_cert_path': '/path/to/server_cert',
}
ingestor_manager.logger.error = MagicMock()
opc_manager.side_effect = Exception('Initialization error')
# 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)
@@ -174,11 +219,14 @@ 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()
@@ -218,9 +266,8 @@ async def test_remove_server_not_found(ingestor_manager):
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.OpcManager')
@patch('ingestor.managers.ingestor_manager.metrics')
async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
async def test_update_opc_servers(metrics, ingestor_manager):
manager1 = MagicMock(config={'config': 'config1'})
manager2 = MagicMock(config={'config': 'config2'})
manager3 = MagicMock(config={'config': 'config3'})
@@ -267,7 +314,9 @@ async def test_update_opc_servers(metrics, opc_manager, 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
@@ -431,25 +480,47 @@ async def test_get_number_of_slots_empty(metrics, ingestor_manager):
@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={'tags': ['tag1']})
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()
assert result == {'1': {'tags': ['tag1']}}
# 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=[{'tags': ['tag1']}, {'tags': ['tag2']}]
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': {'tags': ['tag1']}, '2': {'tags': ['tag2']}}
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
@@ -497,20 +568,20 @@ async def test_unsubscribe_slot(ingestor_manager):
@mark.asyncio
async def test_update_slot_config(ingestor_manager):
ingestor_manager.managed_tags = {
'slot1': {'config': 'old_config'},
'slot2': {'config': 'new_config'},
'slot3': {'config': 'old_config'},
'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=[{'config': 'updated_config'}, {'config': 'new_config'}, None]
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'] == {'config': 'updated_config'}
assert ingestor_manager.managed_tags['slot2'] == {'config': 'new_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')
@@ -652,3 +723,202 @@ async def test_subscribe_to_tags(ingestor_manager):
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 == {}

164
tests/unit/managers/test_server_reader.py Normal file
View File

@@ -0,0 +1,164 @@
from unittest.mock import AsyncMock, MagicMock, call
import pytest
import ingestor.metrics as metrics
from ingestor.managers.server_reader import ServerReader
SERVER_CONFIG = {
'name': 'kepserver',
'server_id': 1, # the slot may carry it as int; the pool keys by str
'url': 'opc.tcp://kepserver:49320',
'server_uri': 'urn:sientia:opc-ingestor',
'use_security': True,
'subscription_period_ms': 15000.0,
}
def make_reader():
pool = MagicMock(get=AsyncMock(), evict=AsyncMock())
reader = ServerReader(
name='kepserver',
server_config=SERVER_CONFIG,
pool=pool,
sink=AsyncMock(),
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
metadata={},
)
reader.engine = MagicMock(
subscriptions={},
pod_id='pod',
create_subscription=AsyncMock(),
subscribe=AsyncMock(),
unsubscribe=AsyncMock(),
check_cycles=AsyncMock(),
check_opc_listenning=AsyncMock(return_value=False),
emit_metric=AsyncMock(),
)
return reader, pool
def test_server_cfg_keeps_only_connection_fields():
reader, _ = make_reader()
assert reader.server_id == '1'
assert reader.server_cfg == {
'url': 'opc.tcp://kepserver:49320',
'server_uri': 'urn:sientia:opc-ingestor',
'use_security': True,
'subscription_period_ms': 15000.0,
}
@pytest.mark.asyncio
async def test_connect_opens_pooled_connection():
reader, pool = make_reader()
await reader.connect()
# use_security=True: the pool fetches the certificate from MinIO (CertProvider)
pool.get.assert_awaited_once_with('1', reader.server_cfg)
@pytest.mark.asyncio
async def test_connect_propagates_failure():
reader, pool = make_reader()
pool.get.side_effect = ConnectionError('unreachable')
with pytest.raises(ConnectionError):
await reader.connect()
@pytest.mark.asyncio
async def test_subscription_calls_delegate_to_engine_per_slot():
reader, _ = make_reader()
nodes = {'ns=2;i=2': {'frequency': '1000', 'names': [{'collection': 'raw_a', 'tag_name': 'a'}]}}
await reader.create_subscription('1')
await reader.subscribe('1', nodes, 10)
await reader.unsubscribe('1')
reader.engine.create_subscription.assert_awaited_once_with('1', reader.server_cfg, '1')
reader.engine.subscribe.assert_awaited_once_with('1', nodes, 10)
reader.engine.unsubscribe.assert_awaited_once_with('1')
@pytest.mark.asyncio
async def test_supervision_delegates_to_engine():
reader, _ = make_reader()
reader.engine.check_opc_listenning.return_value = True
await reader.check_cycles()
lost = await reader.check_opc_listenning()
reader.engine.check_cycles.assert_awaited_once()
reader.engine.check_opc_listenning.assert_awaited_once_with('1')
assert lost is True
def test_subscriptions_are_the_engine_ones():
reader, _ = make_reader()
reader.engine.subscriptions['1'] = MagicMock()
assert reader.subscriptions is reader.engine.subscriptions
@pytest.mark.asyncio
async def test_shutdown_unsubscribes_all_slots_evicts_and_zeroes_metrics():
reader, pool = make_reader()
reader.engine.subscriptions.update({'1': MagicMock(), '2': MagicMock()})
await reader.shutdown()
reader.engine.unsubscribe.assert_has_awaits([call('1'), call('2')])
# Dropping the pooled connection makes the next initialization reconnect.
pool.evict.assert_awaited_once_with('1')
reader.engine.emit_metric.assert_any_await(
metric_object=metrics.OPC_CONNECTION_STATUS,
method='set',
value=0,
tags={'pod_id': 'pod', 'server_name': '1', 'server_url': 'opc.tcp://kepserver:49320'},
)
reader.engine.emit_metric.assert_any_await(
metric_object=metrics.OPC_TAGS_SUBSCRIBED,
method='set',
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
@pytest.mark.asyncio
async def test_shutdown_unsubscribes_every_slot_while_engine_pops_them():
"""The real unsubscribe() pops from engine.subscriptions: shutdown must iterate a
snapshot (list()), or the loop breaks with "dictionary changed size during
iteration" and the remaining slots stay subscribed (SonarQube S7504 does not apply)."""
reader, pool = make_reader()
subs = {'1': MagicMock(), '2': MagicMock(), '3': MagicMock()}
reader.engine.subscriptions = subs
async def pop(slot):
subs.pop(slot)
reader.engine.unsubscribe = AsyncMock(side_effect=pop)
await reader.shutdown()
assert subs == {}
assert reader.engine.unsubscribe.await_count == 3

6
tests/unit/test_app.py
View File

@@ -139,6 +139,7 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys):
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,
pool=mock_ingestor_instance.ingestor_manager.pool,
logger=mock_ingestor_instance.logger,
notification_handler=mock_ingestor_instance.notification_handler,
metrics_controller=mock_ingestor_instance.metrics_controller,
@@ -350,8 +351,9 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
@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."""
"""GatewayRuntime construction failing (e.g. Postgres config) must not kill the
app: the read path lives in the ingestor_manager, so the main loop still runs and
shuts down cleanly."""
mock_ingestor_instance = mock_app_env['Ingestor'].return_value
mock_exit_signal = mock_app_env['mock_exit_signal']

18
tests/unit/test_config.py
View File

@@ -31,7 +31,7 @@ def test_from_env_defaults():
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_bucket == 'streamlit-connectors'
assert cfg.cert_cache_dir == '/tmp/opc-certs' # noqa: S108
assert cfg.write_timestamp is True
assert cfg.write_quality is True
@@ -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

43
tests/unit/test_gateway_runtime.py
View File

@@ -6,9 +6,6 @@ from ingestor import gateway_runtime as gr
from ingestor.config import GatewayConfig
_PATCHED_CLASSES = (
'MinioRepository',
'CertProvider',
'OpcConnectionPool',
'Postgres',
'build_postgres_config',
'DataSink',
@@ -36,8 +33,10 @@ def mocks(monkeypatch):
'min_connections': 5,
'max_connections': 20,
}
patched['OpcConnectionPool'].return_value.close_all = AsyncMock()
patched['OpcConnectionPool'].return_value.release_idle = AsyncMock()
# 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['GrpcServer'].return_value.start = AsyncMock()
@@ -45,10 +44,13 @@ def mocks(monkeypatch):
return patched
def make_runtime(**config_overrides):
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(),
@@ -57,26 +59,26 @@ def make_runtime(**config_overrides):
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,
cert_bucket='my-bucket',
cert_cache_dir='certs-cache',
grpc_enabled=False,
)
mocks['OpcConnectionPool'].assert_called_once()
# 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()
_, kwargs = mocks['WriteEngine'].call_args
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
_, cert_kwargs = mocks['CertProvider'].call_args
assert cert_kwargs['bucket'] == 'my-bucket'
assert cert_kwargs['cache_dir'] == 'certs-cache'
args, _ = mocks['ActiveReadEngine'].call_args
assert args[0] is mocks['pool']
mocks['GrpcServer'].assert_not_called()
assert runtime.grpc_server is None
@@ -137,20 +139,21 @@ async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks):
@pytest.mark.asyncio
async def test_stop_stops_started_components_and_closes_pool(mocks):
runtime = make_runtime(grpc_enabled=True)
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()
mocks['OpcConnectionPool'].return_value.close_all.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()
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'))
@@ -159,14 +162,14 @@ async def test_stop_skips_nats_stop_if_never_started(mocks):
await runtime.stop()
nats_mock.stop.assert_not_called()
mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
mocks['pool'].close_all.assert_not_called()
@pytest.mark.asyncio
async def test_idle_sweep_releases_idle_connections(mocks):
runtime = make_runtime(idle_conn_max_s=42.0)
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['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0)
mocks['pool'].release_idle.assert_awaited_once_with(42.0)