From 4ee672a715e3cc87d009cc1ca9ee09b78621fbd4 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 29 Oct 2026 17:31:15 -0300 Subject: [PATCH] SIENTIAPDE-2202 SIENTIAPDE-2202 Add gateway end-to-end test suite (e2e/) End-to-end suite at the repo root, following the Laborious/Orchestrator pattern: numbered scenarios per topic in e2e/scenarios.md (1:1 with test_scenario_* functions), scenario inputs as JSON, Postgres DDL mirrored from production, `integration` marker, excluded from the default pytest run. Every dependency runs for real: MongoDB, Redis, PostgreSQL, MinIO and NATS via testcontainers, OPC UA servers in process (asyncua). Pods are built from the environment like ingestor/app.py and driven one loop iteration at a time. 69 scenarios across 7 topics: slot leasing and failover, subscription reads, NATS writes and write_status (incl. the SIENTIAPDE-2202 timestamp/quality matrix, checked on the server and on the wire), gRPC ReadTag, active reads, client certificates/security, pod lifecycle. Replaces tests/functional (legacy slot v1 tests, fully commented out) and tests/functional/test_write_e2e.py (now scenarios 3.1.x). Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 +- README.md | 9 +- e2e/README.md | 48 ++ {tests/functional => e2e}/__init__.py | 0 e2e/conftest.py | 355 ++++++++++++ e2e/db_schema.sql | 33 ++ e2e/helpers.py | 243 ++++++++ e2e/opc_server.py | 134 +++++ e2e/scenario_inputs/connector_plant_a.json | 10 + e2e/scenario_inputs/slot_active_read.json | 25 + e2e/scenario_inputs/slot_data_types.json | 28 + e2e/scenario_inputs/slot_multi_name.json | 20 + e2e/scenario_inputs/slot_secured.json | 19 + e2e/scenario_inputs/slot_single_server.json | 19 + e2e/scenario_inputs/slot_two_servers.json | 36 ++ e2e/scenario_inputs/slot_v1_invalid.json | 10 + e2e/scenario_inputs/write_command.json | 12 + e2e/scenarios.md | 584 ++++++++++++++++++++ e2e/test_active_read.py | 84 +++ e2e/test_certificates.py | 165 ++++++ e2e/test_grpc_read_tag.py | 113 ++++ e2e/test_lifecycle.py | 93 ++++ e2e/test_slot_leasing.py | 193 +++++++ e2e/test_subscription_reads.py | 266 +++++++++ e2e/test_writes.py | 386 +++++++++++++ pyproject.toml | 7 +- requirements-dev.txt | 2 + tests/functional/conftest.py | 31 -- tests/functional/test_single_node.py | 93 ---- tests/functional/test_write_e2e.py | 173 ------ 30 files changed, 2893 insertions(+), 304 deletions(-) create mode 100644 e2e/README.md rename {tests/functional => e2e}/__init__.py (100%) create mode 100644 e2e/conftest.py create mode 100644 e2e/db_schema.sql create mode 100644 e2e/helpers.py create mode 100644 e2e/opc_server.py create mode 100644 e2e/scenario_inputs/connector_plant_a.json create mode 100644 e2e/scenario_inputs/slot_active_read.json create mode 100644 e2e/scenario_inputs/slot_data_types.json create mode 100644 e2e/scenario_inputs/slot_multi_name.json create mode 100644 e2e/scenario_inputs/slot_secured.json create mode 100644 e2e/scenario_inputs/slot_single_server.json create mode 100644 e2e/scenario_inputs/slot_two_servers.json create mode 100644 e2e/scenario_inputs/slot_v1_invalid.json create mode 100644 e2e/scenario_inputs/write_command.json create mode 100644 e2e/scenarios.md create mode 100644 e2e/test_active_read.py create mode 100644 e2e/test_certificates.py create mode 100644 e2e/test_grpc_read_tag.py create mode 100644 e2e/test_lifecycle.py create mode 100644 e2e/test_slot_leasing.py create mode 100644 e2e/test_subscription_reads.py create mode 100644 e2e/test_writes.py delete mode 100644 tests/functional/conftest.py delete mode 100644 tests/functional/test_single_node.py delete mode 100644 tests/functional/test_write_e2e.py diff --git a/CLAUDE.md b/CLAUDE.md index f39e914..602597a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -17,9 +17,9 @@ pip install -r requirements.txt -r requirements-dev.txt ./run_local.sh # loads .env, runs python -m ingestor.app ./run_coverage.sh # pytest --cov=ingestor --cov-report=html, opens report -pytest # all tests +pytest # unit tests (testpaths = tests) pytest tests/unit # unit only -pytest tests/functional # functional only (spins up real-ish clients, see conftest.py) +pytest e2e/ -o testpaths=e2e -m integration # E2E (Docker: testcontainers + in-process OPC servers) pytest tests/unit/test_ingestor.py::TestClass::test_name # single test pytest --cov=ingestor --cov-report=html # coverage report -> htmlcov/ @@ -72,6 +72,6 @@ Env vars documented in README ("Configuration" section) and `.env.example`. Nota ## Testing notes - `tests/unit/managers/` mirrors `ingestor/managers/` 1:1. -- `tests/functional/` exercises more end-to-end paths against `conftest.py` fixtures. +- `e2e/` is the end-to-end suite (Laborious/Orchestrator pattern): real MongoDB/Redis/Postgres/MinIO/NATS via testcontainers, in-process asyncua servers, pods built like `app.py` (`e2e/helpers.Pod`). Scenarios are numbered by topic in `e2e/scenarios.md`, 1:1 with `test_scenario____*` β€” keep both in sync. Marked `integration`, not collected by the default `pytest`. - Coverage config (`pyproject.toml` / `.coveragerc`) excludes `ingestor/app.py` (thin entrypoint, signal wiring) β€” don't chase coverage there. - `ruff` complexity cap is `max-complexity = 15` (mccabe) β€” factor out branches in manager methods before that. diff --git a/README.md b/README.md index 9cadccb..7170579 100644 --- a/README.md +++ b/README.md @@ -287,11 +287,14 @@ pytest --cov=ingestor pytest --cov=ingestor --cov-report=html ``` -### Functional Tests +### End-to-end Tests +Every dependency runs for real (MongoDB, Redis, PostgreSQL, MinIO and NATS via testcontainers; +OPC UA servers in process). Requires Docker; not part of the default `pytest` run. ```bash -# Run functional tests -pytest tests/functional/ +pytest e2e/ -o testpaths=e2e -m integration ``` +Numbered scenarios (slot leasing, reads, writes, gRPC, active reads, certificates, lifecycle): +[`e2e/scenarios.md`](e2e/scenarios.md). Setup and layout: [`e2e/README.md`](e2e/README.md). ## πŸ“Š Monitoring and Metrics diff --git a/e2e/README.md b/e2e/README.md new file mode 100644 index 0000000..72869f0 --- /dev/null +++ b/e2e/README.md @@ -0,0 +1,48 @@ +# OPC Gateway end-to-end tests + +End-to-end tests run every external dependency for real (MongoDB, Redis, PostgreSQL, MinIO and +NATS via testcontainers; OPC UA servers in process with asyncua) and build each gateway pod from +the environment exactly like `ingestor/app.py` does. + +## Requirements + +- Docker (for testcontainers), with the reaper disabled once per machine: + `echo "ryuk.disabled=true" >> ~/.testcontainers.properties` (see [scenarios.md](scenarios.md)). +- Python dev dependencies: `pip install -r requirements.txt -r requirements-dev.txt` + +## Run locally + +```bash +source ./venv/bin/activate +pytest e2e/ -o testpaths=e2e -m integration +``` + +One topic, or stop on first failure: + +```bash +pytest e2e/test_writes.py -o testpaths=e2e -m integration +pytest e2e/ -o testpaths=e2e -m integration -x +``` + +The default `pytest` run only collects `tests/` (`testpaths`), so the E2E suite never runs there. + +## Layout + +| Path | Content | +|---|---| +| `conftest.py` | Containers (session), environment + state reset (per test), factories: `opc_servers`, `pods`, `put_slot`, `publish_command`, `client_certificate`, `opc_wire` | +| `helpers.py` | `Pod` (one gateway pod, `tick()` = one main loop iteration), `load_scenario_input`, `wait_until`/`stays_false`, Mongo/Postgres/Redis queries, DataValue mask | +| `opc_server.py` | In-process OPC UA server (`ns=2;s=` variables, restart, secured mode, received writes) | +| `db_schema.sql` | `sientia_data.predictions` DDL, mirrored from production | +| `scenario_inputs/` | Slots (v2), connector and write command used by the scenarios | +| `test_*.py` | One file per topic, `test_scenario____*` | + +## Coverage (separate from unit tests) + +```bash +COVERAGE_FILE=.coverage.e2e pytest e2e/ -o testpaths=e2e -m integration --cov=ingestor --cov-branch +``` + +## Scenario catalog + +See [scenarios.md](scenarios.md) for the numbered scenarios and which test implements each one. diff --git a/tests/functional/__init__.py b/e2e/__init__.py similarity index 100% rename from tests/functional/__init__.py rename to e2e/__init__.py diff --git a/e2e/conftest.py b/e2e/conftest.py new file mode 100644 index 0000000..3b1a880 --- /dev/null +++ b/e2e/conftest.py @@ -0,0 +1,355 @@ +"""Pytest configuration and fixtures for the gateway E2E tests. + +Every external dependency runs for real: MongoDB, Redis, PostgreSQL, MinIO and NATS +via testcontainers (session-scoped), OPC UA servers in process (``e2e/opc_server.py``, +function-scoped). Gateway code under ``ingestor/**`` is not mocked: pods are built from +the environment exactly like ``ingestor/app.py`` builds them (``e2e/helpers.Pod``). +State is reset before every test. +""" + +import io +import json +import os +from pathlib import Path + +import nats +import pytest +import pytest_asyncio +from asyncua import ua +from asyncua.client import ua_client +from minio import Minio +from pymongo import MongoClient +from redis import Redis +from sqlalchemy import create_engine, text +from testcontainers.community.minio import MinioContainer +from testcontainers.community.mongodb import MongoDbContainer +from testcontainers.community.postgres import PostgresContainer +from testcontainers.community.redis import RedisContainer +from testcontainers.core.container import DockerContainer +from testcontainers.core.wait_strategies import LogMessageWaitStrategy + +from e2e.helpers import CERT_BUCKET, Pod +from e2e.opc_server import OpcTestServer, free_port, make_certificate + +DB_SCHEMA_SQL_PATH = Path(__file__).parent / 'db_schema.sql' +MONGO_USER = MONGO_PASSWORD = 'sientia' +MONGO_DATABASE = 'sientia_e2e' +WRITE_SUBJECT = 'opc.write' + + +# -------------------------------------------------------------------------------------- +# Session infrastructure +# -------------------------------------------------------------------------------------- + + +@pytest.fixture(scope='session') +def mongo_container(): + """MongoDB with root credentials: the gateway always connects as user:password@url.""" + mongo = MongoDbContainer('mongo:7', username=MONGO_USER, password=MONGO_PASSWORD) + mongo.start() + yield mongo + mongo.stop() + + +@pytest.fixture(scope='session') +def redis_container(): + redis = RedisContainer('redis:7-alpine') + redis.start() + yield redis + redis.stop() + + +@pytest.fixture(scope='session') +def postgres_container(): + postgres = PostgresContainer( + 'postgres:15', username='sientia', password='sientia', dbname='sientia' + ) + postgres.start() + yield postgres + postgres.stop() + + +@pytest.fixture(scope='session') +def minio_container(): + """ + MinIO holding the OPC client certificates. Same Bitnami "legacy" image as the + Laborious E2E suite and production: Docker Hub no longer serves ``minio/minio``. + """ + minio = MinioContainer(image='docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3') + minio.with_env('MINIO_ROOT_USER', minio.access_key) + minio.with_env('MINIO_ROOT_PASSWORD', minio.secret_key) + minio.with_kwargs(user='root') + minio.start() + yield minio + minio.stop() + + +NATS_IMAGE = 'nats:2.10-alpine' + + +def nats_server_container(host_port: int | None = None) -> DockerContainer: + """NATS server; ``host_port`` pins the published port (to start it after the gateway).""" + container = DockerContainer(NATS_IMAGE).waiting_for(LogMessageWaitStrategy('Server is ready')) + if host_port is None: + return container.with_exposed_ports(4222) + return container.with_bind_ports(4222, host_port) + + +@pytest.fixture(scope='session') +def nats_container(): + nats_server = nats_server_container() + nats_server.start() + yield nats_server + nats_server.stop() + + +@pytest.fixture(scope='session') +def mongo_client(mongo_container): + port = mongo_container.get_exposed_port(27017) + client = MongoClient(f'mongodb://{MONGO_USER}:{MONGO_PASSWORD}@localhost:{port}') + yield client + client.close() + + +@pytest.fixture(scope='session') +def postgres_engine(postgres_container): + engine = create_engine(postgres_container.get_connection_url()) + yield engine + engine.dispose() + + +@pytest.fixture(scope='session') +def redis_client(redis_container): + client = Redis( + host=redis_container.get_container_host_ip(), + port=int(redis_container.get_exposed_port(6379)), + decode_responses=True, + ) + yield client + client.close() + + +@pytest.fixture(scope='session') +def minio_client(minio_container): + return minio_container.get_client() + + +@pytest.fixture(scope='session') +def nats_url(nats_container) -> str: + host = nats_container.get_container_host_ip() + return f'nats://{host}:{nats_container.get_exposed_port(4222)}' + + +# -------------------------------------------------------------------------------------- +# Per-test environment and state reset +# -------------------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def gateway_env( + monkeypatch, + tmp_path, + mongo_container, + redis_container, + postgres_container, + minio_container, + nats_url, +): + """Point every gateway setting at the containers, with the app's defaults otherwise.""" + minio_cfg = minio_container.get_config() + env = { + 'HOSTNAME': 'pod-a', + 'REDIS_HOST': redis_container.get_container_host_ip(), + 'REDIS_PORT': str(redis_container.get_exposed_port(6379)), + 'LEASE_TTL': '10', + 'HEARTBEAT_TTL': '20', + 'POLL_INTERVAL': '1', + 'MONGODB_URL': f'localhost:{mongo_container.get_exposed_port(27017)}', + 'MONGODB_USERNAME': MONGO_USER, + 'MONGODB_PASSWORD': MONGO_PASSWORD, + 'MONGODB_DATABASE': MONGO_DATABASE, + 'POSTGRES_HOST': postgres_container.get_container_host_ip(), + 'POSTGRES_PORT': str(postgres_container.get_exposed_port(5432)), + 'POSTGRES_USER': 'sientia', + 'POSTGRES_PASSWORD': 'sientia', + 'POSTGRES_DBNAME': 'sientia', + 'POSTGRES_MIN_CONNECTIONS': '1', + 'POSTGRES_MAX_CONNECTIONS': '2', + 'MINIO_ENDPOINT': minio_cfg['endpoint'], + 'MINIO_ACCESS_KEY': minio_cfg['access_key'], + 'MINIO_SECRET_KEY': minio_cfg['secret_key'], + 'MINIO_SECURE': 'false', + 'NATS_URL': nats_url, + 'NATS_WRITE_SUBJECT': WRITE_SUBJECT, + 'NATS_QUEUE': 'gateway-writers', + 'GRPC_ENABLED': 'false', + 'GRPC_PORT': str(free_port()), + 'CERT_CACHE_DIR': str(tmp_path / 'cert-cache'), + 'WRITE_MIN_PAUSE_S': '0', + 'IDLE_CONN_MAX_S': '300', + } + for name, value in env.items(): + monkeypatch.setenv(name, value) + for name in ('REDIS_USERNAME', 'REDIS_PASSWORD', 'WRITE_TIMESTAMP', 'WRITE_QUALITY'): + monkeypatch.delenv(name, raising=False) + return env + + +@pytest.fixture(autouse=True) +def reset_state(redis_client, mongo_client, postgres_engine, minio_client): + """Known-empty baseline: Redis flushed, Mongo database dropped, Postgres schema + recreated from ``e2e/db_schema.sql``, certificate bucket emptied.""" + redis_client.flushdb() + mongo_client.drop_database(MONGO_DATABASE) + with postgres_engine.begin() as conn: + conn.execute(text('DROP SCHEMA IF EXISTS sientia_data CASCADE')) + conn.exec_driver_sql(DB_SCHEMA_SQL_PATH.read_text(encoding='utf-8')) + if minio_client.bucket_exists(CERT_BUCKET): + for obj in minio_client.list_objects(CERT_BUCKET, recursive=True): + minio_client.remove_object(CERT_BUCKET, obj.object_name) + else: + minio_client.make_bucket(CERT_BUCKET) + yield + + +@pytest.fixture +def mongo_db(mongo_client): + return mongo_client[MONGO_DATABASE] + + +# -------------------------------------------------------------------------------------- +# Factories: OPC servers, pods, slots, certificates, NATS commands +# -------------------------------------------------------------------------------------- + + +@pytest_asyncio.fixture +async def opc_servers(tmp_path): + """ + Factory of started in-process OPC servers. + + ``await opc_servers('plant-a', {'Temperature': 20.0}, secured=False)`` declares the + variables (``{name: value}`` or ``{name: (value, data_type, writable)}``) and starts it. + """ + started: dict[str, OpcTestServer] = {} + + async def start(name: str, variables: dict, secured: bool = False) -> OpcTestServer: + server = OpcTestServer(name, cert_dir=tmp_path / 'server-certs', secured=secured) + for var, spec in variables.items(): + value, data_type, writable = spec if isinstance(spec, tuple) else (spec, 'float', True) + server.add_variable(var, value, data_type, writable) + await server.start() + started[name] = server + return server + + yield start + for server in started.values(): + await server.stop() + + +@pytest_asyncio.fixture +async def pods(monkeypatch): + """ + Factory of started gateway pods: ``await pods('pod-a', runtime=True, **env)``. + + Extra keyword arguments are environment overrides applied before the pod is built + (e.g. ``WRITE_TIMESTAMP='false'``). Every pod is stopped at teardown. + """ + started: list[Pod] = [] + + async def start(name: str = 'pod-a', runtime: bool = True, **env: str) -> Pod: + for key, value in env.items(): + monkeypatch.setenv(key, value) + pod = Pod(name, runtime=runtime) + started.append(pod) + return await pod.start() + + yield start + for pod in reversed(started): + await pod.stop() + + +@pytest.fixture +def put_slot(redis_client): + """Write a slot v2 to Redis (``slot:opc_tags:``), as the Orchestrator does.""" + + def put(number: int, slot: dict) -> None: + redis_client.set(f'slot:opc_tags:{number}', json.dumps(slot)) + + return put + + +@pytest_asyncio.fixture +async def publish_command(nats_url): + """Publish C2 write commands (dicts) or raw payloads (bytes) on the write subject.""" + client = await nats.connect(nats_url) + + async def publish(payload) -> None: + data = payload if isinstance(payload, bytes) else json.dumps(payload).encode() + await client.publish(WRITE_SUBJECT, data) + await client.flush() + + yield publish + await client.drain() + + +@pytest_asyncio.fixture +async def client_certificate(tmp_path, minio_client: Minio, mongo_db): + """ + Create an OPC client certificate the way the platform stores it: cert (DER) and + private key (PEM) uploaded to MinIO, and an ``opc_certificates`` document (active, + kind client). Returns the certificate in the slot format (contract C1). + """ + + async def create( + certificate_id: int, + upload: bool = True, + status: str = 'active', + app_uri: str | None = None, + ) -> dict: + app_uri = app_uri or f'urn:sientia:e2e:client:{certificate_id}' + cert, key = await make_certificate( + tmp_path / 'client-certs', f'client-{certificate_id}', app_uri, server=False + ) + objects = { + 'cert': {'bucket': CERT_BUCKET, 'object_name': f'{certificate_id}/client.der'}, + 'private_key': {'bucket': CERT_BUCKET, 'object_name': f'{certificate_id}/client.pem'}, + } + if upload: + for path, field in ((cert, 'cert'), (key, 'private_key')): + data = path.read_bytes() + minio_client.put_object( + CERT_BUCKET, objects[field]['object_name'], io.BytesIO(data), len(data) + ) + mongo_db['opc_certificates'].insert_one( + { + 'certificate_id': certificate_id, + 'kind': 'client', + 'status': status, + 'application_uri': app_uri, + **objects, + } + ) + return {'certificate_id': certificate_id, 'application_uri': app_uri, **objects} + + return create + + +@pytest.fixture +def cert_cache_dir() -> Path: + return Path(os.environ['CERT_CACHE_DIR']) + + +@pytest.fixture +def opc_wire(monkeypatch): + """Serialized WriteRequests, exactly as the gateway's OPC client sends them.""" + sent: list[bytes] = [] + original = ua_client.struct_to_binary + + def capture(obj): + data = original(obj) + if isinstance(obj, ua.WriteRequest): + sent.append(data) + return data + + monkeypatch.setattr(ua_client, 'struct_to_binary', capture) + return sent diff --git a/e2e/db_schema.sql b/e2e/db_schema.sql new file mode 100644 index 0000000..eba2b9a --- /dev/null +++ b/e2e/db_schema.sql @@ -0,0 +1,33 @@ +-- ============================================================================= +-- E2E test database schema for the ``sientia_data`` namespace. +-- +-- Mirrors the production DDL one-to-one so any change in production can be +-- pasted directly into this file. The conftest fixture drops the schema and +-- loads this SQL into the testcontainers Postgres before every test. +-- +-- The gateway only touches ``sientia_data.predictions``: it updates +-- ``write_status`` (contract C3, SIENTIAPDE-1975) on the row the Laborious +-- inserted. Same DDL as ``sientia-dataops-laborious/e2e/db_schema.sql``. +-- ============================================================================= + +CREATE SCHEMA IF NOT EXISTS sientia_data; + +-- ----------------------------------------------------------------------------- +-- sientia_data.predictions +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.predictions ( + model_id int4 NOT NULL, + prediction numeric NULL, + prediction_confidence numeric NOT NULL, + response_time numeric NOT NULL, + prediction_status text NOT NULL, + "timestamp" timestamptz NOT NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + "comments" text NULL, + metadata_id varchar(24) NULL, + write_status text NULL, + CONSTRAINT unique_model_id_timestamp + UNIQUE (model_id, "timestamp"), + CONSTRAINT chk_predictions_write_status + CHECK (write_status IN ('pending', 'ok', 'error', 'session_bad')) +); diff --git a/e2e/helpers.py b/e2e/helpers.py new file mode 100644 index 0000000..4b5a137 --- /dev/null +++ b/e2e/helpers.py @@ -0,0 +1,243 @@ +"""Shared helpers for the gateway E2E tests (OPC UA + Redis + MongoDB + Postgres + MinIO + NATS).""" + +import asyncio +import copy +import inspect +import json +import os +import time +from collections.abc import Awaitable, Callable +from pathlib import Path +from typing import Any + +from asyncua import ua +from asyncua.ua.ua_binary import Primitives, nodeid_to_binary +from pymongo.database import Database +from sqlalchemy import text +from sqlalchemy.engine import Engine + +from ingestor.config import GatewayConfig +from ingestor.gateway_runtime import GatewayRuntime +from ingestor.ingestor import Ingestor + +SCENARIO_INPUTS_DIR = Path(__file__).parent / 'scenario_inputs' + +# Collections the gateway reads (connectors, opc_certificates) or writes +# (readings, notification_queue). Dropped before every test with the whole database. +NOTIFICATIONS = 'notification_queue' +CERT_BUCKET = 'opc-certificates' +PREDICTIONS_TABLE = 'predictions' + + +def _resolve_markers(value: Any, servers: dict) -> Any: + """Replace ``@server:`` with the URL of that in-process OPC server.""" + if isinstance(value, dict): + return {k: _resolve_markers(v, servers) for k, v in value.items()} + if isinstance(value, list): + return [_resolve_markers(v, servers) for v in value] + if isinstance(value, str) and value.startswith('@server:'): + return servers[value.removeprefix('@server:')].url + return value + + +def load_scenario_input(name: str, servers: dict | None = None, **overrides: Any) -> Any: + """ + Load a JSON scenario input from ``e2e/scenario_inputs``. + + Args: + name: File name (e.g. ``slot_single_server.json``). + servers: ``{name: OpcTestServer}`` used to resolve ``@server:`` markers. + **overrides: Top-level keys replaced in the loaded dict. + + Return: + Any: The resolved JSON content. + """ + data = json.loads((SCENARIO_INPUTS_DIR / name).read_text(encoding='utf-8')) + data = _resolve_markers(data, servers or {}) + if overrides: + data = {**data, **copy.deepcopy(overrides)} + return data + + +async def wait_until( + predicate: Callable[[], Any | Awaitable[Any]], timeout: float = 15.0, interval: float = 0.1 +) -> Any: + """ + Poll ``predicate`` (sync or async) until it returns a truthy value. + + Sleeps on the event loop between attempts, so OPC subscription callbacks and NATS + handlers keep running while the test waits. + + Raises: + AssertionError: When the predicate is still falsy after ``timeout`` seconds. + """ + deadline = time.monotonic() + timeout + while True: + result = predicate() + if inspect.isawaitable(result): + result = await result + if result: + return result + if time.monotonic() > deadline: + raise AssertionError(f'condition not met within {timeout}s') + await asyncio.sleep(interval) + + +async def stays_false( + predicate: Callable[[], Any | Awaitable[Any]], duration: float = 1.5, interval: float = 0.1 +) -> bool: + """True when ``predicate`` never becomes truthy during ``duration`` seconds.""" + deadline = time.monotonic() + duration + while time.monotonic() < deadline: + result = predicate() + if inspect.isawaitable(result): + result = await result + if result: + return False + await asyncio.sleep(interval) + return True + + +class Pod: + """ + One gateway pod, wired exactly as ``ingestor/app.py`` does (minus the Prometheus + HTTP server and the sleep between iterations): ``Ingestor.prepare_ingestor``, then + ``GatewayRuntime`` on its data manager and connection pool. ``tick`` runs one main + loop iteration (``Ingestor.loop`` + ``GatewayRuntime.idle_sweep``). + + Every component reads its settings from the environment at construction time, so + the caller sets the env (``HOSTNAME`` included) before ``start``. + """ + + def __init__(self, name: str, runtime: bool = True): + self.name = name + self.with_runtime = runtime + self.ingestor: Ingestor | None = None + self.runtime: GatewayRuntime | None = None + + async def start(self) -> 'Pod': + os.environ['HOSTNAME'] = self.name + self.ingestor = Ingestor() + await self.ingestor.prepare_ingestor() + if self.with_runtime: + manager = self.ingestor.ingestor_manager + assert manager is not None + self.runtime = GatewayRuntime( + config=GatewayConfig.from_env(), + data_manager=manager.data_manager, + pool=manager.pool, + logger=self.ingestor.logger, + notification_handler=self.ingestor.notification_handler, + metrics_controller=self.ingestor.metrics_controller, + metadata=self.ingestor.metadata, + ) + await self.runtime.start() + consumer = self.runtime.nats_consumer + if consumer.nc is not None: + # PING/PONG: the server has processed the SUB, so a command published + # right after start is not lost for lack of a subscriber. + await consumer.nc.flush() + return self + + @property + def manager(self): + assert self.ingestor is not None and self.ingestor.ingestor_manager is not None + return self.ingestor.ingestor_manager + + @property + def slots(self) -> set[str]: + """Slots this pod currently manages.""" + return set(self.manager.managed_tags) + + async def tick(self, times: int = 1) -> None: + assert self.ingestor is not None + for _ in range(times): + await self.ingestor.loop() + if self.runtime is not None: + await self.runtime.idle_sweep() + + async def stop(self) -> None: + if self.runtime is not None: + await self.runtime.stop() + self.runtime = None + if self.ingestor is not None: + await self.ingestor.shutdown() + self.ingestor.notification_handler.shutdown() + self.ingestor = None + + +def redis_json(redis_client, key: str) -> Any: + """Value stored by the gateway's RedisRepository (JSON-encoded), or None.""" + raw = redis_client.get(key) + return None if raw is None else json.loads(raw) + + +def readings(db: Database, collection: str, **query: Any) -> list[dict]: + """Documents the read path persisted into ``collection`` (oldest first).""" + return list(db[collection].find(query, {'_id': 0}).sort('inserted_at', 1)) + + +def notifications(db: Database, prefix: str = '') -> list[dict]: + """Notifications sent by the gateway whose ``notification_id`` starts with ``prefix``.""" + return [ + n + for n in db[NOTIFICATIONS].find({}, {'_id': 0}) + if str(n.get('notification_id', '')).startswith(prefix) + ] + + +def insert_prediction(engine: Engine, model_id: int, timestamp: str, write_status: str | None): + """One ``sientia_data.predictions`` row, as the Laborious inserts it before publishing C2.""" + with engine.begin() as conn: + conn.execute( + text( + 'INSERT INTO sientia_data.predictions (model_id, prediction, ' + 'prediction_confidence, response_time, prediction_status, "timestamp", ' + 'write_status) VALUES (:model_id, 1.5, 0.9, 0.1, :status, :ts, :write_status)' + ), + { + 'model_id': model_id, + 'status': 'success', + 'ts': timestamp, + 'write_status': write_status, + }, + ) + + +def write_status(engine: Engine, model_id: int, timestamp: str) -> str | None: + """``write_status`` of a prediction row (``''`` when the row does not exist).""" + with engine.begin() as conn: + row = conn.execute( + text( + 'SELECT write_status FROM sientia_data.predictions ' + 'WHERE model_id = :model_id AND "timestamp" = :ts' + ), + {'model_id': model_id, 'ts': timestamp}, + ).first() + return '' if row is None else row[0] + + +def prediction_count(engine: Engine) -> int: + with engine.begin() as conn: + return conn.execute(text('SELECT count(*) FROM sientia_data.predictions')).scalar_one() + + +# DataValue encoding mask bits (OPC UA Part 6, 5.2.2.17). +STATUS_CODE_BIT = 0x02 +SOURCE_TIMESTAMP_BIT = 0x04 + + +def datavalue_mask(request: bytes, addr: str) -> int: + """ + Encoding mask of the DataValue written to ``addr`` in a serialized WriteRequest. + + An omitted StatusCode decodes as Good on the server (OPC UA default), so whether + the gateway sent the quality is only visible on the wire. WriteValue = NodeId, + AttributeId, IndexRange, DataValue: the byte right after the first three is the mask. + """ + prefix = ( + nodeid_to_binary(ua.NodeId.from_string(addr)) + + Primitives.UInt32.pack(ua.AttributeIds.Value) + + Primitives.String.pack(None) + ) + return request[request.index(prefix) + len(prefix)] diff --git a/e2e/opc_server.py b/e2e/opc_server.py new file mode 100644 index 0000000..8da3652 --- /dev/null +++ b/e2e/opc_server.py @@ -0,0 +1,134 @@ +"""In-process OPC UA servers for the E2E suite (asyncua `Server`). + +Each `OpcTestServer` listens on its own localhost port and exposes string NodeIds +under namespace 2 (`ns=2;s=`), so a slot can address them the same way the +Orchestrator addresses a real server. Writes received from the gateway are recorded +(decoded `WriteValue`s) through the server's PreWrite callback. +""" + +import socket +import time +from pathlib import Path + +from asyncua import Server, ua +from asyncua.common.callback import CallbackType +from asyncua.crypto.cert_gen import setup_self_signed_certificate +from cryptography.x509.oid import ExtendedKeyUsageOID + +NAMESPACE_URI = 'urn:sientia:e2e' + +_VARIANT_TYPES = { + 'float': ua.VariantType.Float, + 'double': ua.VariantType.Double, + 'int': ua.VariantType.Int32, + 'bool': ua.VariantType.Boolean, + 'string': ua.VariantType.String, +} + + +def free_port() -> int: + with socket.socket() as s: + s.bind(('127.0.0.1', 0)) + return s.getsockname()[1] + + +def address(name: str) -> str: + """NodeId string of a variable created by `OpcTestServer.add_variable`.""" + return f'ns=2;s={name}' + + +async def make_certificate( + folder: Path, name: str, app_uri: str, server: bool +) -> tuple[Path, Path]: + """Self-signed application certificate (DER) + private key (PEM) with `app_uri` as + the SAN URI, as asyncua expects from both ends of a secure channel.""" + folder.mkdir(parents=True, exist_ok=True) + cert, key = folder / f'{name}.der', folder / f'{name}.pem' + usage = ExtendedKeyUsageOID.SERVER_AUTH if server else ExtendedKeyUsageOID.CLIENT_AUTH + await setup_self_signed_certificate( + key, cert, app_uri, 'localhost', [usage], {'commonName': name} + ) + return cert, key + + +class OpcTestServer: + """One asyncua server. `secured=True` exposes only Basic256Sha256 SignAndEncrypt.""" + + def __init__(self, name: str, cert_dir: Path | None = None, secured: bool = False): + self.name = name + self.port = free_port() + self.secured = secured + self.cert_dir = cert_dir + self.server: Server | None = None + self.variables: dict[str, tuple[object, str, bool]] = {} + self.writes: list[ua.WriteValue] = [] + self.write_times: list[float] = [] # time.monotonic() of each received write + + @property + def url(self) -> str: + return f'opc.tcp://127.0.0.1:{self.port}' + + def add_variable( + self, name: str, value: object, data_type: str = 'float', writable: bool = True + ) -> str: + """Declare a variable (created on `start`, and again on every restart).""" + self.variables[name] = (value, data_type, writable) + return address(name) + + async def start(self) -> None: + server = Server() + await server.init() + server.set_endpoint(self.url) + server.set_server_name(self.name) + if self.secured: + assert self.cert_dir is not None, 'a secured server needs a cert_dir' + cert, key = await make_certificate( + self.cert_dir, f'server-{self.name}', f'urn:sientia:e2e:{self.name}', True + ) + await server.load_certificate(str(cert)) + await server.load_private_key(str(key)) + server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt]) + idx = await server.register_namespace(NAMESPACE_URI) + assert idx == 2, f'unexpected namespace index {idx}' + for name, (value, data_type, writable) in self.variables.items(): + node = await server.nodes.objects.add_variable( + address(name), name, value, _VARIANT_TYPES[data_type] + ) + if writable: + await node.set_writable() + server.subscribe_server_callback(CallbackType.PreWrite, self._record_write) + await server.start() + self.server = server + + async def stop(self) -> None: + if self.server is not None: + await self.server.stop() + self.server = None + + async def restart(self) -> None: + """Stop and start again on the same port (values reset to their initial ones).""" + await self.stop() + await self.start() + + async def set(self, name: str, value: object) -> None: + """Change a value server-side (raises a data change for subscribed clients).""" + assert self.server is not None + _, data_type, _ = self.variables[name] + node = self.server.get_node(address(name)) + await node.write_value(ua.Variant(value, _VARIANT_TYPES[data_type])) + + async def read(self, name: str) -> ua.DataValue: + assert self.server is not None + return await self.server.get_node(address(name)).read_data_value() + + def session_count(self) -> int: + """Client sessions currently open on this server.""" + assert self.server is not None + return len(self.server.iserver._external_sessions) + + async def _record_write(self, event, _dispatcher) -> None: + if not event.is_external: # the server updates its own nodes (ServerStatus, ...) + return + for value in event.request_params.NodesToWrite: + self.writes.append(value) + self.write_times.append(time.monotonic()) diff --git a/e2e/scenario_inputs/connector_plant_a.json b/e2e/scenario_inputs/connector_plant_a.json new file mode 100644 index 0000000..ede48e7 --- /dev/null +++ b/e2e/scenario_inputs/connector_plant_a.json @@ -0,0 +1,10 @@ +{ + "connector_id": "1", + "type": "opc_ua", + "name": "plant-a", + "url": "@server:plant-a", + "status": "Active", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None", "client_certificate_id": null}, + "legacy_opc_server_id": "1" +} diff --git a/e2e/scenario_inputs/slot_active_read.json b/e2e/scenario_inputs/slot_active_read.json new file mode 100644 index 0000000..356329e --- /dev/null +++ b/e2e/scenario_inputs/slot_active_read.json @@ -0,0 +1,25 @@ +{ + "plant-a": { + "name": "plant-a", + "server_id": "1", + "url": "@server:plant-a", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Level": { + "frequency": 1000, + "active_read": {"enabled": true, "cycle_multiple": 1}, + "names": [ + {"collection": "active_tank", "tag_name": "tank_level", "model_id": "30"}, + {"collection": "active_mixer", "tag_name": "mixer_inlet_level", "model_id": "40"} + ] + }, + "ns=2;s=Temperature": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [{"collection": "active_tank", "tag_name": "tank_temperature", "model_id": "30"}] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_data_types.json b/e2e/scenario_inputs/slot_data_types.json new file mode 100644 index 0000000..5f2866d --- /dev/null +++ b/e2e/scenario_inputs/slot_data_types.json @@ -0,0 +1,28 @@ +{ + "plant-a": { + "name": "plant-a", + "server_id": "1", + "url": "@server:plant-a", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Temperature": { + "frequency": 1000, + "names": [{"collection": "raw_types", "tag_name": "temperature", "model_id": "10"}] + }, + "ns=2;s=Counter": { + "frequency": 1000, + "names": [{"collection": "raw_types", "tag_name": "counter", "model_id": "10"}] + }, + "ns=2;s=Running": { + "frequency": 1000, + "names": [{"collection": "raw_types", "tag_name": "running", "model_id": "10"}] + }, + "ns=2;s=Batch": { + "frequency": 1000, + "names": [{"collection": "raw_types", "tag_name": "batch", "model_id": "10"}] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_multi_name.json b/e2e/scenario_inputs/slot_multi_name.json new file mode 100644 index 0000000..3dca638 --- /dev/null +++ b/e2e/scenario_inputs/slot_multi_name.json @@ -0,0 +1,20 @@ +{ + "plant-a": { + "name": "plant-a", + "server_id": "1", + "url": "@server:plant-a", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Pressure": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_furnace", "tag_name": "furnace_pressure", "model_id": "10", "model_name": "furnace"}, + {"collection": "raw_boiler", "tag_name": "boiler_inlet_pressure", "model_id": "20", "model_name": "boiler"} + ] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_secured.json b/e2e/scenario_inputs/slot_secured.json new file mode 100644 index 0000000..5f4b4e8 --- /dev/null +++ b/e2e/scenario_inputs/slot_secured.json @@ -0,0 +1,19 @@ +{ + "plant-s": { + "name": "plant-s", + "server_id": "7", + "url": "@server:plant-s", + "subscription_period_ms": 100, + "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256"}, + "certificate": null, + "tags": { + "ns=2;s=Temperature": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_secured", "tag_name": "secured_temperature", "model_id": "70"} + ] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_single_server.json b/e2e/scenario_inputs/slot_single_server.json new file mode 100644 index 0000000..ee2b7c5 --- /dev/null +++ b/e2e/scenario_inputs/slot_single_server.json @@ -0,0 +1,19 @@ +{ + "plant-a": { + "name": "plant-a", + "server_id": "1", + "url": "@server:plant-a", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Temperature": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_plant_a", "tag_name": "temperature", "model_id": "10", "model_name": "furnace"} + ] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_two_servers.json b/e2e/scenario_inputs/slot_two_servers.json new file mode 100644 index 0000000..a858095 --- /dev/null +++ b/e2e/scenario_inputs/slot_two_servers.json @@ -0,0 +1,36 @@ +{ + "plant-a": { + "name": "plant-a", + "server_id": "1", + "url": "@server:plant-a", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Temperature": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_plant_a", "tag_name": "temperature", "model_id": "10", "model_name": "furnace"} + ] + } + } + }, + "plant-b": { + "name": "plant-b", + "server_id": "2", + "url": "@server:plant-b", + "subscription_period_ms": 100, + "security": {"mode": "None", "policy": "None"}, + "certificate": null, + "tags": { + "ns=2;s=Flow": { + "frequency": 1000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_plant_b", "tag_name": "flow", "model_id": "20", "model_name": "boiler"} + ] + } + } + } +} diff --git a/e2e/scenario_inputs/slot_v1_invalid.json b/e2e/scenario_inputs/slot_v1_invalid.json new file mode 100644 index 0000000..689b1a4 --- /dev/null +++ b/e2e/scenario_inputs/slot_v1_invalid.json @@ -0,0 +1,10 @@ +{ + "plant-a": { + "name": "plant-a", + "url": "@server:plant-a", + "server_uri": "http://opcua-server.simulator", + "tags": { + "ns=2;s=Temperature": {"tag_name": "Temperature", "frequency": 1000, "topics": ["raw_plant_a"]} + } + } +} diff --git a/e2e/scenario_inputs/write_command.json b/e2e/scenario_inputs/write_command.json new file mode 100644 index 0000000..fa02249 --- /dev/null +++ b/e2e/scenario_inputs/write_command.json @@ -0,0 +1,12 @@ +{ + "server_id": "1", + "addr": "ns=2;s=Setpoint", + "value": 1.25, + "data_type": "float", + "tag_type": "prediction", + "model_id": "2", + "model_name": "sin-approx", + "schema": "sientia_data", + "table_name": "predictions", + "timestamp": "2026-10-13T12:00:00-03:00" +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md new file mode 100644 index 0000000..5020d1a --- /dev/null +++ b/e2e/scenarios.md @@ -0,0 +1,584 @@ +# E2E Scenario Documentation β€” OPC Gateway + +Functional reference for the gateway end-to-end scenarios: slot leasing, subscription reads, +NATS writes, gRPC reads, active reads, certificates and the pod lifecycle. Tests live under +`e2e/`, are marked with `@pytest.mark.integration` and run with: + +```bash +pytest e2e/ -o testpaths=e2e -m integration +``` + +**Every section below maps 1:1 to an actual test function** (or to one `pytest.param` id of a +parametrized test) β€” this file describes what the suite verifies today. Adding a scenario here +without a matching test (or vice versa) is a doc bug; fix the mismatch instead of leaving it +for later. + +## Execution Context + +### One-time setup: disable the testcontainers reaper + +Same as the Laborious and Orchestrator suites. Once per machine: + +```bash +echo "ryuk.disabled=true" >> ~/.testcontainers.properties +``` + +Under Docker Desktop on Linux the reaper ("Ryuk") cannot mount the Docker socket and every +test errors at setup (`ERROR`, not `FAILED`). Every fixture stops its own containers in +teardown; after a killed run, clean up by hand: + +```bash +docker ps -a --filter "label=org.testcontainers=true" -q | xargs -r docker rm -f +``` + +### What runs for real + +| Dependency | How | Scope | +|---|---|---| +| MongoDB (`mongo:7`, user/password) | testcontainers | session | +| Redis (`redis:7-alpine`) | testcontainers | session | +| PostgreSQL (`postgres:15`, `e2e/db_schema.sql`) | testcontainers | session | +| MinIO (`bitnamilegacy/minio`, client certificates) | testcontainers | session | +| NATS (`nats:2.10-alpine`) | testcontainers | session | +| OPC UA servers | in process (`e2e/opc_server.py`, asyncua `Server`) | per test | + +- Gateway code under `ingestor/**` is not mocked. A pod (`e2e/helpers.Pod`) is built from the + environment exactly like `ingestor/app.py` builds it β€” `Ingestor.prepare_ingestor`, then + `GatewayRuntime` on the same data manager and connection pool β€” minus the Prometheus HTTP + server. `Pod.tick()` runs one main loop iteration (`Ingestor.loop` + `GatewayRuntime.idle_sweep`) + without the `POLL_INTERVAL` sleep, so every scenario controls exactly how many iterations run. +- The environment (`gateway_env` in `e2e/conftest.py`) points every setting at the containers; + scenarios override single variables per pod (`LEASE_TTL`, `WRITE_TIMESTAMP`, ...). +- State is reset before every test: Redis `FLUSHDB`, Mongo database dropped, Postgres schema + recreated from `e2e/db_schema.sql`, certificate bucket emptied. +- Inputs live in `e2e/scenario_inputs/*.json` and are loaded with `load_scenario_input`; + `@server:` markers resolve to the URL of that test's in-process OPC server. +- Default `POLL_INTERVAL=1` and tag `frequency=1000`: one OPC "cycle" per loop iteration. +- Not covered by the default `pytest` run (`testpaths = ["tests"]`) nor by the unit coverage. + `--strict-markers` is on: a test without `@pytest.mark.integration` is not collected, and a run + that collects nothing still exits 0 β€” check the collected count. + +--- + +## 1. Slot Leasing and Load Balancing +Source: `e2e/test_slot_leasing.py` + +### 1.1.1 Single Pod Leases the Slot and Heartbeats +**Summary**: A lone pod takes the only slot on its first loop iteration. + +**Description**: +- Slot 1 reads `plant-a`; one pod starts and runs one iteration. + +**Expected Outcome**: +- The pod manages slot `1`; `lease:opc_tags:1 = "pod-a"` with TTL ≀ `LEASE_TTL` (10 s). +- `heartbeat:ingestor:pod-a = 1` with TTL ≀ `HEARTBEAT_TTL` (20 s). +- Readings of the slot reach MongoDB. + +### 1.1.2 Idle Pod Takes a Slot Created After Startup +**Summary**: A pod without slots picks up a slot the Orchestrator creates later. + +**Description**: +- The pod starts with Redis empty and runs one iteration (nothing to lease). +- Slot 1 is written; the pod runs another iteration. + +**Expected Outcome**: +- The pod manages slot `1`, holds its lease and its readings reach MongoDB. + +### 1.2.1 Two Pods Split Two Slots +**Summary**: The rebalancing hands one slot to each pod. + +**Description**: +- Slot 1 reads `plant-a`, slot 2 reads `plant-b`. `pod-a` starts alone, then `pod-b` starts. +- Iterations: `pod-a`, `pod-b`, `pod-a`, `pod-b`. + +**Expected Outcome**: +- Alone, `pod-a` takes both slots; `pod-b` finds them leased. +- `pod-a` sees a pod without slots and gives slot `2` back (unsubscribing first); `pod-b` takes it. +- Leases: slot 1 β†’ `pod-a`, slot 2 β†’ `pod-b`; each pod only has the reader of its own server. +- New values of both servers reach MongoDB. + +### 1.2.2 Extra Pod Stays Idle +**Summary**: With more pods than slots, no slot is read twice. + +**Description**: +- Two slots balanced between `pod-a` and `pod-b`; `pod-c` joins and every pod iterates twice. + +**Expected Outcome**: +- `pod-c` manages nothing; `pod-a` and `pod-b` hold disjoint slots covering `{1, 2}`. +- Leases belong to `pod-a` and `pod-b` only. + +### 1.2.3 Failover After Lease Expiry +**Summary**: A slot left by a dead pod is taken over once its lease expires. + +**Description**: +- `pod-a` (`LEASE_TTL=2`) reads slot 1; `pod-b` runs idle. +- `pod-a` stops without releasing the lease (crash-like). + +**Expected Outcome**: +- The lease still names `pod-a` right after the stop, then expires. +- On its next iteration `pod-b` leases slot `1` and new values reach MongoDB. + +### 1.3.1 Deleted Slot Stops the Reads +**Summary**: Removing a slot from Redis unsubscribes it. + +**Description**: +- A pod reads slot 1; the slot key is deleted; the pod iterates. + +**Expected Outcome**: +- The pod manages no slot and has no OPC reader. +- A value changed afterwards is not persisted. + +### 1.3.2 Invalid Slot Is Notified and Not Read +**Summary**: A slot v1 (`topics`, no `security`/`names`) is rejected. + +**Description**: +- Slot 1 holds the legacy v1 shape (`slot_v1_invalid.json`). + +**Expected Outcome**: +- Notification `OPC_SLOT_CONFIG_ERROR_1` in `notification_queue`. +- The pod manages nothing; no reading is persisted. + +### 1.3.3 Fixed Slot Is Read Once Its Lease Expires +**Summary**: Fixing an invalid slot takes effect after the lease of the failed acquisition expires. + +**Description**: +- `pod-a` (`LEASE_TTL=2`) leases the invalid slot 1 (the lease is kept, the slot is not managed). +- The slot is fixed; the pod iterates before and after the lease expires. + +**Expected Outcome**: +- Before expiry the pod still manages nothing (its own lease blocks the re-acquisition). +- After expiry the pod manages slot `1` and its readings reach MongoDB. + +--- + +## 2. Subscription Reads (OPC UA β†’ MongoDB) +Source: `e2e/test_subscription_reads.py` + +### 2.1.1 Value Change Is Persisted +**Summary**: Each data change becomes one document in the name's collection. + +**Description**: +- Slot 1 subscribes `ns=2;s=Temperature` of `plant-a` into `raw_plant_a`; the value changes. + +**Expected Outcome**: +- The initial value and the new one are persisted. +- Document fields: `tag` (address), `name` (`tag_name`), `model_id`, `value`, `timestamp` + (source timestamp, `YYYY-MM-DD HH:MM:SS-0300`) and `inserted_at` β€” nothing else. + +### 2.1.2 One Address Fans Out to Every Name +**Summary**: An address with several `names` is written to each collection. + +**Description**: +- `ns=2;s=Pressure` has two names: `raw_furnace`/`furnace_pressure`/model 10 and + `raw_boiler`/`boiler_inlet_pressure`/model 20 (`slot_multi_name.json`). + +**Expected Outcome**: +- One document per collection for the change, each with its own `name`/`model_id`, the same + `tag` and the same `timestamp`. + +### 2.1.3 Two Servers in One Slot +**Summary**: A slot can hold several OPC servers. + +**Description**: +- Slot 1 holds `plant-a` and `plant-b` (`slot_two_servers.json`); both values change. + +**Expected Outcome**: +- Both changes are persisted; the pod has one reader per server. + +### 2.1.4 Values Keep Their OPC Type +**Summary**: Double, Int32, Boolean and String values are stored with their native type. + +**Description**: +- Four tags of different OPC types change value (`slot_data_types.json`). + +**Expected Outcome**: +- Each document stores the value with the matching Python/BSON type (`float`, `int`, `bool`, `str`). + +### 2.2.1 Tag Added to the Slot Is Subscribed +**Summary**: A new tag in the slot is subscribed on the next iteration. + +**Description**: +- The Orchestrator adds `ns=2;s=Level` to slot 1; the pod iterates. + +**Expected Outcome**: +- Changes of the new tag and of the existing one are persisted. + +### 2.2.2 Tag Removed from the Slot Is No Longer Read +**Summary**: A tag removed from the slot stops being persisted. + +**Description**: +- Slot 1 reads `Temperature` and `Level`; `Level` is removed; the pod iterates. + +**Expected Outcome**: +- `Temperature` keeps being persisted; a new `Level` value is not. + +### 2.2.3 Name Added to a Tag Receives Its Readings +**Summary**: A new name on an existing tag starts receiving the readings. + +**Description**: +- A second name (`raw_dryer`/`dryer_temperature`/model 50) is added to `Temperature`. + +**Expected Outcome**: +- The next change is persisted in both collections. + +### 2.2.4 Server URL Change Reconnects to the New Server +**Summary**: Changing the connection settings of a server re-initializes its reader. + +**Description**: +- Slot 1 points `plant-a` to a second server (same tags) and the pod iterates. + +**Expected Outcome**: +- Values of the new server are persisted; the session on the old server is closed. +- Changes on the old server are no longer persisted. + +### 2.3.1 Unreachable Server Does Not Stop the Others +**Summary**: A server that cannot be reached is dropped; the other servers of the slot keep reading. + +**Description**: +- Slot 1 holds `plant-a` and `plant-b`, whose URL has nothing listening. + +**Expected Outcome**: +- Notification `OPC_CONNECTION_ERROR_plant-b`; the pod only has the `plant-a` reader. +- `plant-a` values are persisted. + +### 2.3.2 Tag Without Data Raises a Warning +**Summary**: A tag without data changes for 5 cycles is notified. + +**Description**: +- The value does not change for 5 iterations. + +**Expected Outcome**: +- One WARNING notification `TAG_ns=2;s=Temperature:temperature_LISTENNING_STOPPED`. + +### 2.3.3 Silent Server Is Reconnected After 15 Cycles +**Summary**: A server without any data change for 15 cycles is treated as silently dead. + +**Description**: +- The value does not change for 15 iterations, then the pod iterates once more. + +**Expected Outcome**: +- Notifications `OPC_LISTENNING_STOPPED__1` and `OPC_CONNECTION_RETRY__1`; the reader is dropped + (pool connection evicted). +- The next iteration re-initializes the reader with a new connection; changes are persisted again. + +### 2.3.4 Server Restart Resumes the Reads +**Summary**: A server that goes down and comes back is read again. + +**Description**: +- The server stops; the pod iterates 15 times (reader dropped) and once more while it is down; + the server starts again on the same port; the pod iterates twice. + +**Expected Outcome**: +- While down, the reconnection fails with notification `OPC_CONNECTION_ERROR_plant-a`. +- Once back, the reader is re-initialized and changes are persisted again. + +--- + +## 3. Writes (NATS C2 β†’ OPC UA β†’ Postgres `write_status` C3) +Source: `e2e/test_writes.py` + +Common setup: `plant-a` with a writable `Setpoint`, its connector in `connectors` +(`connector_plant_a.json`, `connector_id "1"`) and a `pending` row in `sientia_data.predictions` +(model 2, `2026-10-13T12:00:00-03:00`). Commands come from `write_command.json` and are published +on `opc.write`; the gateway answers through the row's `write_status`. + +### 3.1 DataValue Parts: Timestamp and Quality (SIENTIAPDE-2202) +**Summary**: The value is always written; the prediction timestamp (`SourceTimestamp`) and a Good +`StatusCode` are written unless turned off. Each flag comes from the command, else from the env +(`WRITE_TIMESTAMP` / `WRITE_QUALITY`), else `true`. + +**Description**: +- `test_scenario_3_1_datavalue_parts`, one case per id. The timestamp is checked on the value the + server stored; the quality on the DataValue encoding mask of the WriteRequest the gateway sent + (an omitted StatusCode decodes as Good on the server, so only the wire tells them apart). + +| Id | Env | Command flags | SourceTimestamp | StatusCode | +|---|---|---|---|---| +| `3_1_1_defaults_on` | β€” | β€” | sent | sent | +| `3_1_2_timestamp_off` | `WRITE_TIMESTAMP=false` | β€” | not sent | sent | +| `3_1_3_quality_off` | `WRITE_QUALITY=false` | β€” | sent | not sent | +| `3_1_4_both_off` | both `false` | β€” | not sent | not sent | +| `3_1_5_command_overrides_env` | `WRITE_TIMESTAMP=true` | `write_timestamp: false` | not sent | sent | +| `3_1_6_command_turns_on_over_env` | both `false` | both `true` | sent | sent | + +**Expected Outcome**: +- `write_status = ok`; the server holds `1.25`. +- When sent, the stored `SourceTimestamp` equals the command `timestamp`; otherwise it is empty. + +### 3.2.1 Pending Row Becomes Ok +**Summary**: A successful write records `ok`. + +**Expected Outcome**: +- `write_status = ok`; the server holds the command value. + +### 3.2.2 Null write_status Becomes Ok +**Summary**: Rows without a status (inserted before the column existed) are updated too. + +**Expected Outcome**: +- `write_status` goes from NULL to `ok`. + +### 3.2.3 Worse Outcome Is Kept +**Summary**: `ok` never replaces a worse outcome (`ok` < `error` < `session_bad`). + +**Description**: +- Parametrized: the row starts as `error` or `session_bad`; a successful command arrives. + +**Expected Outcome**: +- The OPC write happens, the row keeps its worse status. + +### 3.2.4 Failure Replaces Ok +**Summary**: A worse outcome replaces a better one. + +**Description**: +- A successful command records `ok`; a second command for the same row fails (conversion). + +**Expected Outcome**: +- `write_status = error`. + +### 3.2.5 Missing Row Is Not Created +**Summary**: The gateway only updates rows; it never inserts one. + +**Description**: +- The command targets model 3, which has no prediction row. + +**Expected Outcome**: +- The OPC write happens; no row for model 3 exists afterwards. + +### 3.3.1 Unknown Server Is an Error +**Summary**: A `server_id` with no live connection and no connector fails the write. + +**Expected Outcome**: +- `write_status = error`; nothing is written to the server. + +### 3.3.2 Unusable Connector Is an Error +**Summary**: Only active `opc_ua` connectors with a URL and full security are used. + +**Description**: +- Parametrized: connector `inactive`, `not_opc_ua` (`type: modbus`), `no_policy`, `no_url`. + +**Expected Outcome**: +- `write_status = error`; nothing is written to the server. + +### 3.3.3 Invalid Command Is an Error Without OPC Write +**Summary**: Commands that cannot become a DataValue are rejected before touching the server. + +**Description**: +- Parametrized: `conversion` (`"abc"` as `int`), `unknown_data_type` (`decimal128`), `missing_addr`. + +**Expected Outcome**: +- `write_status = error`; nothing is written to the server. + +### 3.3.4 Bad Timestamp Skips the Write and Keeps the Row +**Summary**: A non-ISO `timestamp` (with the timestamp flag on) fails before the OPC write. + +**Description**: +- The command carries `13/10/2026 12:00`; a marker command is used to know it was handled. + +**Expected Outcome**: +- Only the marker reaches the server; the row stays `pending` β€” the status cannot be recorded + either, since the row is found by that same timestamp. + +### 3.3.5 OPC Rejection Is an Error +**Summary**: A write the server refuses is recorded as `error`. + +**Description**: +- Parametrized: `not_writable` (read-only `Locked`), `unknown_node` (`ns=2;s=Missing`). + +**Expected Outcome**: +- `write_status = error`; `Locked` keeps its value. + +### 3.3.6 Malformed JSON Is Dropped and the Consumer Keeps Going +**Summary**: A payload that is not JSON is discarded without stopping the consumer. + +**Expected Outcome**: +- The next valid command is written and recorded as `ok`. + +### 3.4.1 Write Reuses the Read Connection +**Summary**: A server read by the pod is written through the same pooled connection. + +**Description**: +- `plant-a` is only known through slot 1 (no connector in Mongo). + +**Expected Outcome**: +- `write_status = ok`; the server has a single session. + +### 3.4.2 Write-Only Server Gets Its Own Connection +**Summary**: A server only known as a connector is connected on the first write. + +**Expected Outcome**: +- No session before the command; one session after; the pool holds the connector's settings. + +### 3.5.1 Minimum Pause Between Writes +**Summary**: `WRITE_MIN_PAUSE_S` spaces consecutive writes. + +**Description**: +- `WRITE_MIN_PAUSE_S=0.5`; three commands are published back to back. + +**Expected Outcome**: +- The server receives the three writes at least ~0.5 s apart. + +### 3.5.2 Queue Group Writes Each Command Once +**Summary**: Pods share the `gateway-writers` queue: each command is handled by a single pod. + +**Description**: +- Two pods with the runtime; six commands. + +**Expected Outcome**: +- The server receives exactly six writes. + +### 3.6.1 NATS Down at Startup Connects Later +**Summary**: The runtime starts without NATS and connects from the main loop once NATS is up. + +**Description**: +- `NATS_URL` points to a port with nothing listening; the pod starts and iterates once. +- A NATS server starts on that port; the pod iterates again. + +**Expected Outcome**: +- One `GATEWAY_NATS_CONSUMER_START_FAILED` notification for the whole outage (not one per retry). +- After NATS starts, the consumer connects and a published command is recorded as `ok`. + +--- + +## 4. On-demand gRPC ReadTag +Source: `e2e/test_grpc_read_tag.py` + +Common setup: `GRPC_ENABLED=true`; slot 1 reads `plant-a`, so the pod's pool holds a connection. + +### 4.1.1 Reads a Tag of a Connected Server +**Expected Outcome**: +- `value` (as text), `opc_status_code = Good`, `source_timestamp` / `server_timestamp` in + `YYYY-MM-DD HH:MM:SS-0300`. +- `persisted = false` and no collection is created: the slot lookup of `ReadTag` is not wired yet. + +### 4.1.2 Reads the Current Value +**Expected Outcome**: +- After a server-side change, `ReadTag` returns the new value. + +### 4.2.1 Unknown Address Is an Internal Error +**Expected Outcome**: +- gRPC `INTERNAL` whose details carry `BadNodeIdUnknown`. + +### 4.2.2 Server Not Connected by the Reads Is an Internal Error +**Summary**: `ReadTag` has no connection settings of its own: it only reads servers the read path +already connected (documents the current limitation). + +**Description**: +- `plant-a` only exists as a connector; no slot. + +**Expected Outcome**: +- gRPC `INTERNAL`; no session is opened on the server. + +### 4.2.3 gRPC Disabled by Default +**Expected Outcome**: +- Without `GRPC_ENABLED`, nothing listens on `GRPC_PORT` (`UNAVAILABLE`/`DEADLINE_EXCEEDED`). + +--- + +## 5. Scheduled Active Reads +Source: `e2e/test_active_read.py` + +`GatewayRuntime` builds `ActiveReadEngine` on the pod's pool and sink, but no loop calls +`scheduled_tick` yet; these scenarios call it on the running pod's engine with the parsed slot +(`slot_active_read.json`). + +### 5.1.1 Enabled Tag Is Read into Every Name +**Expected Outcome**: +- `Level` (`active_read.enabled`) is persisted in `active_tank` (model 30) and `active_mixer` + (model 40); `Temperature` (disabled) is not read. + +### 5.1.2 cycle_multiple Reads Every Nth Tick +**Description**: +- `cycle_multiple = 3`; six ticks. + +**Expected Outcome**: +- Persisted documents after each tick: `0, 0, 1, 1, 1, 2`. + +### 5.2.1 Unreachable Server Does Not Stop the Tick +**Expected Outcome**: +- The tick does not raise; the unreachable server's tag is not persisted, the other server's is. + +### 5.2.2 Active Read Uses the Slot Connection Settings +**Expected Outcome**: +- The first tick opens the connection with the slot settings; later ticks reuse it (one session); + each tick persists the current value. + +--- + +## 6. Client Certificates and OPC Security +Source: `e2e/test_certificates.py` + +Common setup: `plant-s` exposes only Basic256Sha256 SignAndEncrypt. Client certificates are created +like the platform stores them: cert (DER) and key (PEM) in MinIO bucket `opc-certificates`, plus an +`opc_certificates` document (`kind: client`). + +### 6.1.1 Secured Read with Certificate from MinIO +**Expected Outcome**: +- Readings are persisted; the certificate is cached as `/7/cert.der` and `key.pem`. +- No connection error is notified. + +### 6.1.2 Secured Write Resolved from Connectors and Certificates +**Description**: +- No slot: the connector (`client_certificate_id: 7`) is joined with `opc_certificates`. + +**Expected Outcome**: +- `write_status = ok`; the server holds the value. + +### 6.2.1 Certificate Missing in MinIO +**Expected Outcome**: +- Notification `OPC_CONNECTION_ERROR_plant-s`; no reader; nothing cached; no readings. + +### 6.2.2 Unsupported Security Policy +**Expected Outcome**: +- Notification `OPC_CONNECTION_ERROR_plant-s` whose message names the unsupported mode/policy. + +### 6.2.3 No Certificate Cannot Reach a Secured-only Server +**Description**: +- The slot has `certificate: null` (connect without security, D5). + +**Expected Outcome**: +- Notification `OPC_CONNECTION_ERROR_plant-s`; no readings. + +### 6.2.4 Inactive Certificate Write Fails +**Description**: +- The connector's certificate is `revoked`: the write connects without security. + +**Expected Outcome**: +- `write_status = error`; nothing is written to the server. + +### 6.3.1 Certificate Rotation Reconnects with the New One +**Description**: +- Slot 1 switches from certificate 7 to certificate 8; the pod iterates. + +**Expected Outcome**: +- The pooled connection is replaced and opened with certificate 8 (cached under `8/`). +- Readings continue; the server ends with a single session. + +--- + +## 7. Pod Lifecycle +Source: `e2e/test_lifecycle.py` + +### 7.1.1 Shutdown Closes Sessions and Stops Consuming +**Expected Outcome**: +- After `GatewayRuntime.stop` + `Ingestor.shutdown`, the OPC server has no session. +- A command published afterwards is not handled (row stays `pending`); value changes are not persisted. + +### 7.1.2 Idle Write Connection Is Closed and Reopened +**Description**: +- `IDLE_CONN_MAX_S=1`; `plant-a` is read (slot), `plant-b` only written (connector `2`). + +**Expected Outcome**: +- After the idle time, `idle_sweep` closes the `plant-b` session; the subscribed `plant-a` + connection is pinned and stays open. +- The next write to `plant-b` reconnects and is recorded as `ok`. + +### 7.2.1 Redis Unreachable Fails the Startup +**Expected Outcome**: +- `prepare_ingestor` raises a connection error (`app.py` then stops the process). + +### 7.2.2 Restarted Pod Takes Its Slot Back +**Description**: +- `pod-a` (`LEASE_TTL=2`) stops; a new `pod-a` starts after the lease expired. + +**Expected Outcome**: +- The restarted pod leases slot `1` and its readings reach MongoDB. diff --git a/e2e/test_active_read.py b/e2e/test_active_read.py new file mode 100644 index 0000000..3b01072 --- /dev/null +++ b/e2e/test_active_read.py @@ -0,0 +1,84 @@ +"""E2E scenarios β€” 5. Scheduled active reads. See ``e2e/scenarios.md``. + +``GatewayRuntime`` builds the ``ActiveReadEngine`` with the pod's pool and sink, but no +loop calls ``scheduled_tick`` yet (its wiring is pending in ``gateway_runtime.py``), so +these scenarios call it on the running pod's engine with the parsed slot. +""" + +import pytest + +from e2e.helpers import load_scenario_input, readings, stays_false, wait_until +from e2e.opc_server import free_port +from ingestor.io.slots import parse_slot + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +async def test_scenario_5_1_1_enabled_tag_is_read_into_every_name(opc_servers, pods, mongo_db): + server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0}) + slot = parse_slot(load_scenario_input('slot_active_read.json', {'plant-a': server})) + pod = await pods('pod-a') + + await pod.runtime.active_read.scheduled_tick(slot) + + [tank] = readings(mongo_db, 'active_tank', name='tank_level') + [mixer] = readings(mongo_db, 'active_mixer', name='mixer_inlet_level') + assert tank['value'] == mixer['value'] == 2.5 + assert (tank['model_id'], mixer['model_id']) == ('30', '40') + assert tank['tag'] == 'ns=2;s=Level' + assert readings(mongo_db, 'active_tank', name='tank_temperature') == [] # not enabled + + +async def test_scenario_5_1_2_cycle_multiple_reads_every_nth_tick(opc_servers, pods, mongo_db): + server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0}) + raw = load_scenario_input('slot_active_read.json', {'plant-a': server}) + raw['plant-a']['tags']['ns=2;s=Level']['active_read']['cycle_multiple'] = 3 + slot = parse_slot(raw) + pod = await pods('pod-a') + + counts = [] + for _ in range(6): + await pod.runtime.active_read.scheduled_tick(slot) + counts.append(len(readings(mongo_db, 'active_tank', name='tank_level'))) + + assert counts == [0, 0, 1, 1, 1, 2] + + +async def test_scenario_5_2_1_unreachable_server_does_not_stop_the_tick( + opc_servers, pods, mongo_db +): + server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0}) + raw = load_scenario_input('slot_active_read.json', {'plant-a': server}) + down = {**raw['plant-a'], 'name': 'plant-down', 'server_id': '9'} + down['url'] = f'opc.tcp://127.0.0.1:{free_port()}' + down['tags'] = { + 'ns=2;s=Level': { + **raw['plant-a']['tags']['ns=2;s=Level'], + 'names': [{'collection': 'active_down', 'tag_name': 'down_level', 'model_id': '90'}], + } + } + slot = parse_slot({'plant-down': down, 'plant-a': raw['plant-a']}) + pod = await pods('pod-a') + + await pod.runtime.active_read.scheduled_tick(slot) # does not raise + + assert readings(mongo_db, 'active_down') == [] + await wait_until(lambda: readings(mongo_db, 'active_tank', name='tank_level')) + + +async def test_scenario_5_2_2_active_read_uses_the_slot_connection_settings( + opc_servers, pods, mongo_db +): + server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0}) + slot = parse_slot(load_scenario_input('slot_active_read.json', {'plant-a': server})) + pod = await pods('pod-a') + assert pod.manager.pool.config_for('1') is None + + await pod.runtime.active_read.scheduled_tick(slot) + + assert pod.manager.pool.config_for('1')['url'] == server.url + assert server.session_count() == 1 + await server.set('Level', 3.0) + await pod.runtime.active_read.scheduled_tick(slot) + assert await stays_false(lambda: server.session_count() != 1, duration=0.5) + assert [d['value'] for d in readings(mongo_db, 'active_tank', name='tank_level')] == [2.5, 3.0] diff --git a/e2e/test_certificates.py b/e2e/test_certificates.py new file mode 100644 index 0000000..e7028e4 --- /dev/null +++ b/e2e/test_certificates.py @@ -0,0 +1,165 @@ +"""E2E scenarios β€” 6. Client certificates and OPC security (MinIO + opc_certificates). See ``e2e/scenarios.md``.""" + +import pytest +import pytest_asyncio + +from e2e.helpers import ( + insert_prediction, + load_scenario_input, + notifications, + readings, + stays_false, + wait_until, + write_status, +) + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + +TIMESTAMP = '2026-10-13T12:00:00-03:00' +SECURED = {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'} + + +@pytest_asyncio.fixture +async def secured_server(opc_servers): + """plant-s: only Basic256Sha256 SignAndEncrypt endpoints, accepts any client certificate.""" + return await opc_servers('plant-s', {'Temperature': 20.0, 'Setpoint': 0.0}, secured=True) + + +def secured_slot(server, certificate, **security): + slot = load_scenario_input('slot_secured.json', {'plant-s': server}) + slot['plant-s']['certificate'] = certificate + slot['plant-s']['security'] = {**SECURED, **security} + return slot + + +def secured_connector(server, certificate_id): + return { + 'connector_id': '7', + 'type': 'opc_ua', + 'name': 'plant-s', + 'url': server.url, + 'status': 'Active', + 'subscription_period_ms': 100, + 'security': {**SECURED, 'client_certificate_id': certificate_id}, + } + + +def write_command(): + return load_scenario_input( + 'write_command.json', server_id='7', model_id='2', timestamp=TIMESTAMP + ) + + +# 6.1 Secured connections --------------------------------------------------------------- + + +async def test_scenario_6_1_1_secured_read_with_certificate_from_minio( + secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir +): + certificate = await client_certificate(7) + put_slot(1, secured_slot(secured_server, certificate)) + pod = await pods('pod-a') + + await pod.tick() + + await secured_server.set('Temperature', 21.0) + await wait_until(lambda: readings(mongo_db, 'raw_secured', value=21.0)) + assert sorted(p.name for p in (cert_cache_dir / '7').iterdir()) == ['cert.der', 'key.pem'] + assert notifications(mongo_db, 'OPC_CONNECTION_ERROR') == [] + + +async def test_scenario_6_1_2_secured_write_resolved_from_connectors_and_certificates( + secured_server, client_certificate, pods, mongo_db, postgres_engine, publish_command +): + await client_certificate(7) + mongo_db['connectors'].insert_one(secured_connector(secured_server, 7)) + insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending') + await pods('pod-a') + + await publish_command(write_command()) + + await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'ok') + assert (await secured_server.read('Setpoint')).Value.Value == 1.25 + + +# 6.2 Failures -------------------------------------------------------------------------- + + +async def test_scenario_6_2_1_certificate_missing_in_minio( + secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir +): + certificate = await client_certificate(7, upload=False) + put_slot(1, secured_slot(secured_server, certificate)) + pod = await pods('pod-a') + + await pod.tick() + + assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s') + assert pod.manager.opc_managers == {} + assert not list((cert_cache_dir / '7').glob('*.der')) + assert await stays_false(lambda: readings(mongo_db, 'raw_secured')) + + +async def test_scenario_6_2_2_unsupported_security_policy( + secured_server, client_certificate, pods, put_slot, mongo_db +): + certificate = await client_certificate(7) + put_slot(1, secured_slot(secured_server, certificate, policy='Basic999')) + pod = await pods('pod-a') + + await pod.tick() + + [error] = notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s') + assert 'Unsupported OPC security' in error['message'] + assert pod.manager.opc_managers == {} + + +async def test_scenario_6_2_3_no_certificate_cannot_reach_a_secured_only_server( + secured_server, pods, put_slot, mongo_db +): + put_slot(1, secured_slot(secured_server, None)) + pod = await pods('pod-a') + + await pod.tick() + + assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s') + assert await stays_false(lambda: readings(mongo_db, 'raw_secured')) + + +async def test_scenario_6_2_4_inactive_certificate_write_fails( + secured_server, client_certificate, pods, mongo_db, postgres_engine, publish_command +): + await client_certificate(7, status='revoked') + mongo_db['connectors'].insert_one(secured_connector(secured_server, 7)) + insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending') + await pods('pod-a') + + await publish_command(write_command()) + + await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'error') + assert secured_server.writes == [] + + +# 6.3 Rotation -------------------------------------------------------------------------- + + +async def test_scenario_6_3_1_certificate_rotation_reconnects_with_the_new_one( + secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir +): + first = await client_certificate(7) + put_slot(1, secured_slot(secured_server, first)) + pod = await pods('pod-a') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_secured')) + first_client = pod.manager.pool._connections['7'][0] + + second = await client_certificate(8) + put_slot(1, secured_slot(secured_server, second)) + await pod.tick() + + assert pod.manager.pool._connections['7'][0] is not first_client + assert pod.manager.pool.config_for('7')['certificate']['certificate_id'] == 8 + assert (cert_cache_dir / '8' / 'cert.der').is_file() + await secured_server.set('Temperature', 22.0) + await wait_until(lambda: readings(mongo_db, 'raw_secured', value=22.0)) + await wait_until(lambda: secured_server.session_count() == 1) diff --git a/e2e/test_grpc_read_tag.py b/e2e/test_grpc_read_tag.py new file mode 100644 index 0000000..0be2655 --- /dev/null +++ b/e2e/test_grpc_read_tag.py @@ -0,0 +1,113 @@ +"""E2E scenarios β€” 4. On-demand gRPC ReadTag. See ``e2e/scenarios.md``.""" + +import os +import re + +import grpc +import pytest +import pytest_asyncio + +from e2e.helpers import load_scenario_input, readings, wait_until +from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + +FORMATTED_TIMESTAMP = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}-0300$') + + +@pytest_asyncio.fixture +async def grpc_stub(): + """Client of the gateway's gRPC server (GRPC_PORT of the test environment).""" + channel = grpc.aio.insecure_channel(f'127.0.0.1:{os.environ["GRPC_PORT"]}') + yield opc_gateway_pb2_grpc.OpcGatewayStub(channel) + await channel.close() + + +async def _reading_pod(opc_servers, pods, put_slot, mongo_db): + server = await opc_servers('plant-a', {'Temperature': 20.0, 'Level': 3.25}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + pod = await pods('pod-a', GRPC_ENABLED='true') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + return server, pod + + +async def test_scenario_4_1_1_reads_a_tag_of_a_connected_server( + opc_servers, pods, put_slot, mongo_db, grpc_stub +): + await _reading_pod(opc_servers, pods, put_slot, mongo_db) + persisted_before = mongo_db.list_collection_names() + + response = await grpc_stub.ReadTag( + opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float') + ) + + assert response.value == '3.25' + assert response.opc_status_code == 'Good' + assert FORMATTED_TIMESTAMP.match(response.source_timestamp), response.source_timestamp + assert FORMATTED_TIMESTAMP.match(response.server_timestamp), response.server_timestamp + assert response.persisted is False # no slot lookup wired yet: nothing is written + assert mongo_db.list_collection_names() == persisted_before + + +async def test_scenario_4_1_2_reads_the_current_value( + opc_servers, pods, put_slot, mongo_db, grpc_stub +): + server, _ = await _reading_pod(opc_servers, pods, put_slot, mongo_db) + await server.set('Level', 4.5) + + response = await grpc_stub.ReadTag( + opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float') + ) + + assert response.value == '4.5' + + +async def test_scenario_4_2_1_unknown_address_is_internal_error( + opc_servers, pods, put_slot, mongo_db, grpc_stub +): + await _reading_pod(opc_servers, pods, put_slot, mongo_db) + + with pytest.raises(grpc.aio.AioRpcError) as error: + await grpc_stub.ReadTag( + opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Missing', data_type='float') + ) + + assert error.value.code() == grpc.StatusCode.INTERNAL + assert 'BadNodeIdUnknown' in error.value.details() + + +async def test_scenario_4_2_2_server_not_connected_by_the_reads_is_internal_error( + opc_servers, pods, mongo_db, grpc_stub +): + server = await opc_servers('plant-a', {'Level': 3.25}) + mongo_db['connectors'].insert_one( + load_scenario_input('connector_plant_a.json', {'plant-a': server}) + ) + await pods('pod-a', GRPC_ENABLED='true') # no slot: the pool has no connection to plant-a + + with pytest.raises(grpc.aio.AioRpcError) as error: + await grpc_stub.ReadTag( + opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float') + ) + + assert error.value.code() == grpc.StatusCode.INTERNAL + assert server.session_count() == 0 + + +async def test_scenario_4_2_3_grpc_disabled_by_default( + opc_servers, pods, put_slot, mongo_db, grpc_stub +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + pod = await pods('pod-a') + await pod.tick() + + with pytest.raises(grpc.aio.AioRpcError) as error: + await grpc_stub.ReadTag( + opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Temperature'), + timeout=2, + ) + + assert error.value.code() in (grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED) + assert pod.runtime.grpc_server is None diff --git a/e2e/test_lifecycle.py b/e2e/test_lifecycle.py new file mode 100644 index 0000000..20ff3a6 --- /dev/null +++ b/e2e/test_lifecycle.py @@ -0,0 +1,93 @@ +"""E2E scenarios β€” 7. Pod lifecycle: shutdown, idle connections, startup failures. See ``e2e/scenarios.md``.""" + +import asyncio + +import pytest + +from e2e.helpers import ( + Pod, + insert_prediction, + load_scenario_input, + readings, + stays_false, + wait_until, + write_status, +) +from e2e.opc_server import free_port + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + +TIMESTAMP = '2026-10-13T12:00:00-03:00' + + +async def test_scenario_7_1_1_shutdown_closes_sessions_and_stops_consuming( + opc_servers, pods, put_slot, mongo_db, postgres_engine, publish_command +): + server = await opc_servers('plant-a', {'Temperature': 20.0, 'Setpoint': 0.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending') + pod = await pods('pod-a') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + assert server.session_count() == 1 + + await pod.stop() + + await wait_until(lambda: server.session_count() == 0) + await publish_command(load_scenario_input('write_command.json')) + assert await stays_false(lambda: write_status(postgres_engine, 2, TIMESTAMP) != 'pending') + await server.set('Temperature', 50.0) + assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=50.0)) + + +async def test_scenario_7_1_2_idle_write_connection_is_closed_and_reopened( + opc_servers, pods, put_slot, mongo_db, postgres_engine, publish_command +): + reader = await opc_servers('plant-a', {'Temperature': 20.0}) + writer = await opc_servers('plant-b', {'Setpoint': 0.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': reader})) + connector = load_scenario_input('connector_plant_a.json', {'plant-a': writer}) + mongo_db['connectors'].insert_one({**connector, 'connector_id': '2', 'name': 'plant-b'}) + insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending') + pod = await pods('pod-a', IDLE_CONN_MAX_S='1') + await pod.tick() + + await publish_command(load_scenario_input('write_command.json', server_id='2')) + await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'ok') + assert writer.session_count() == 1 + + await asyncio.sleep(1.5) + await pod.tick() # idle_sweep + + await wait_until(lambda: writer.session_count() == 0) + assert reader.session_count() == 1 # subscribed: pinned, never idle + + insert_prediction(postgres_engine, 3, TIMESTAMP, 'pending') + await publish_command(load_scenario_input('write_command.json', server_id='2', model_id='3')) + await wait_until(lambda: write_status(postgres_engine, 3, TIMESTAMP) == 'ok') + assert writer.session_count() == 1 + + +async def test_scenario_7_2_1_redis_unreachable_fails_the_startup(monkeypatch): + monkeypatch.setenv('REDIS_PORT', str(free_port())) + + with pytest.raises(Exception, match='(?i)connect'): + await Pod('pod-a').start() + + +async def test_scenario_7_2_2_restarted_pod_takes_its_slot_back( + opc_servers, pods, put_slot, mongo_db, redis_client +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + pod = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2') + await pod.tick() + await pod.stop() + + await wait_until(lambda: redis_client.get('lease:opc_tags:1') is None, timeout=5) + restarted = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2') + await restarted.tick() + + assert restarted.slots == {'1'} + await server.set('Temperature', 31.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=31.0)) diff --git a/e2e/test_slot_leasing.py b/e2e/test_slot_leasing.py new file mode 100644 index 0000000..8afd912 --- /dev/null +++ b/e2e/test_slot_leasing.py @@ -0,0 +1,193 @@ +"""E2E scenarios β€” 1. Slot leasing and load balancing (Redis). See ``e2e/scenarios.md``.""" + +import asyncio + +import pytest + +from e2e.helpers import ( + load_scenario_input, + notifications, + readings, + redis_json, + stays_false, + wait_until, +) + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +async def _two_slots(opc_servers, put_slot): + """Slot 1 reads plant-a, slot 2 reads plant-b.""" + plant_a = await opc_servers('plant-a', {'Temperature': 20.0}) + plant_b = await opc_servers('plant-b', {'Flow': 3.0}) + both = load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b}) + put_slot(1, {'plant-a': both['plant-a']}) + put_slot(2, {'plant-b': both['plant-b']}) + return plant_a, plant_b + + +# 1.1 Acquisition ---------------------------------------------------------------------- + + +async def test_scenario_1_1_1_single_pod_leases_the_slot_and_heartbeats( + opc_servers, pods, put_slot, redis_client, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + + pod = await pods('pod-a') + await pod.tick() + + assert pod.slots == {'1'} + assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a' + assert 0 < redis_client.ttl('lease:opc_tags:1') <= 10 + assert redis_json(redis_client, 'heartbeat:ingestor:pod-a') == 1 + assert 0 < redis_client.ttl('heartbeat:ingestor:pod-a') <= 20 + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + +async def test_scenario_1_1_2_idle_pod_takes_a_slot_created_after_startup( + opc_servers, pods, put_slot, redis_client, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + pod = await pods('pod-a') + await pod.tick() + assert pod.slots == set() + + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + await pod.tick() + + assert pod.slots == {'1'} + assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a' + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + +# 1.2 Balancing and failover ------------------------------------------------------------ + + +async def test_scenario_1_2_1_two_pods_split_two_slots( + opc_servers, pods, put_slot, redis_client, mongo_db +): + plant_a, plant_b = await _two_slots(opc_servers, put_slot) + + pod_a = await pods('pod-a') + await pod_a.tick() + assert pod_a.slots == {'1', '2'} # alone: takes every slot + + pod_b = await pods('pod-b') + await pod_b.tick() + assert pod_b.slots == set() # every slot is leased + await pod_a.tick() # pod-a now sees a pod without slot and gives one back + assert pod_a.slots == {'1'} + await pod_b.tick() + assert pod_b.slots == {'2'} + + assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a' + assert redis_json(redis_client, 'lease:opc_tags:2') == 'pod-b' + assert set(pod_a.manager.opc_managers) == {'plant-a'} + assert set(pod_b.manager.opc_managers) == {'plant-b'} + await plant_a.set('Temperature', 25.0) + await plant_b.set('Flow', 4.5) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=25.0)) + await wait_until(lambda: readings(mongo_db, 'raw_plant_b', value=4.5)) + + +async def test_scenario_1_2_2_extra_pod_stays_idle(opc_servers, pods, put_slot, redis_client): + await _two_slots(opc_servers, put_slot) + pod_a, pod_b = await pods('pod-a'), await pods('pod-b') + await pod_a.tick() + await pod_b.tick() + await pod_a.tick() + await pod_b.tick() + + pod_c = await pods('pod-c') + for _ in range(2): + for pod in (pod_a, pod_b, pod_c): + await pod.tick() + + assert pod_c.slots == set() + assert pod_a.slots | pod_b.slots == {'1', '2'} + assert pod_a.slots.isdisjoint(pod_b.slots) + assert { + redis_json(redis_client, 'lease:opc_tags:1'), + redis_json(redis_client, 'lease:opc_tags:2'), + } == { + 'pod-a', + 'pod-b', + } + + +async def test_scenario_1_2_3_failover_after_lease_expiry( + opc_servers, pods, put_slot, redis_client, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + pod_a = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2') + await pod_a.tick() + pod_b = await pods('pod-b') + await pod_b.tick() + assert (pod_a.slots, pod_b.slots) == ({'1'}, set()) + + await pod_a.stop() # crash-like: the lease is not released, it expires + assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a' + await wait_until(lambda: redis_json(redis_client, 'lease:opc_tags:1') is None, timeout=5) + await pod_b.tick() + + assert pod_b.slots == {'1'} + assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-b' + await server.set('Temperature', 30.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=30.0)) + + +# 1.3 Slot changes and invalid slots ---------------------------------------------------- + + +async def test_scenario_1_3_1_deleted_slot_stops_the_reads( + opc_servers, pods, put_slot, redis_client, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + pod = await pods('pod-a') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + redis_client.delete('slot:opc_tags:1') + await pod.tick() + + assert pod.slots == set() + assert pod.manager.opc_managers == {} + await server.set('Temperature', 99.0) + assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=99.0)) + + +async def test_scenario_1_3_2_invalid_slot_is_notified_and_not_read( + opc_servers, pods, put_slot, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_v1_invalid.json', {'plant-a': server})) + + pod = await pods('pod-a') + await pod.tick() + + assert pod.slots == set() + assert notifications(mongo_db, 'OPC_SLOT_CONFIG_ERROR_1') + assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a')) + + +async def test_scenario_1_3_3_fixed_slot_is_read_once_its_lease_expires( + opc_servers, pods, put_slot, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0}) + put_slot(1, load_scenario_input('slot_v1_invalid.json', {'plant-a': server})) + pod = await pods('pod-a', LEASE_TTL='2') + await pod.tick() + assert pod.slots == set() + + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + await pod.tick() + assert pod.slots == set() # still leased by the failed acquisition + + await asyncio.sleep(2.5) + await pod.tick() + assert pod.slots == {'1'} + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) diff --git a/e2e/test_subscription_reads.py b/e2e/test_subscription_reads.py new file mode 100644 index 0000000..fab241e --- /dev/null +++ b/e2e/test_subscription_reads.py @@ -0,0 +1,266 @@ +"""E2E scenarios β€” 2. Subscription reads (OPC UA β†’ MongoDB). See ``e2e/scenarios.md``.""" + +import re + +import pytest + +from e2e.helpers import load_scenario_input, notifications, readings, stays_false, wait_until +from e2e.opc_server import free_port + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + +# DATETIME_FORMAT_WITH_TZ in OPC_TIMEZONE (sientia_do), e.g. 2026-10-29 19:48:51-0300 +READING_TIMESTAMP = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}-0300$') + + +async def _read_slot(opc_servers, pods, put_slot, slot_file, variables): + server = await opc_servers('plant-a', variables) + slot = load_scenario_input(slot_file, {'plant-a': server}) + put_slot(1, slot) + pod = await pods('pod-a') + await pod.tick() + return server, slot, pod + + +# 2.1 Persistence ----------------------------------------------------------------------- + + +async def test_scenario_2_1_1_value_change_is_persisted(opc_servers, pods, put_slot, mongo_db): + server, _, _ = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=20.0)) # initial value + + await server.set('Temperature', 21.5) + + [doc] = await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=21.5)) + assert doc['tag'] == 'ns=2;s=Temperature' + assert doc['name'] == 'temperature' + assert doc['model_id'] == '10' + assert READING_TIMESTAMP.match(doc['timestamp']), doc['timestamp'] + assert doc['inserted_at'] is not None + assert set(doc) == {'tag', 'name', 'model_id', 'timestamp', 'value', 'inserted_at'} + + +async def test_scenario_2_1_2_one_address_fans_out_to_every_name( + opc_servers, pods, put_slot, mongo_db +): + server, _, _ = await _read_slot( + opc_servers, pods, put_slot, 'slot_multi_name.json', {'Pressure': 1.0} + ) + + await server.set('Pressure', 2.5) + + [furnace] = await wait_until(lambda: readings(mongo_db, 'raw_furnace', value=2.5)) + [boiler] = await wait_until(lambda: readings(mongo_db, 'raw_boiler', value=2.5)) + assert (furnace['name'], furnace['model_id']) == ('furnace_pressure', '10') + assert (boiler['name'], boiler['model_id']) == ('boiler_inlet_pressure', '20') + assert furnace['tag'] == boiler['tag'] == 'ns=2;s=Pressure' + assert furnace['timestamp'] == boiler['timestamp'] + + +async def test_scenario_2_1_3_two_servers_in_one_slot(opc_servers, pods, put_slot, mongo_db): + plant_a = await opc_servers('plant-a', {'Temperature': 20.0}) + plant_b = await opc_servers('plant-b', {'Flow': 3.0}) + put_slot( + 1, load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b}) + ) + pod = await pods('pod-a') + await pod.tick() + + await plant_a.set('Temperature', 22.0) + await plant_b.set('Flow', 3.5) + + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=22.0)) + await wait_until(lambda: readings(mongo_db, 'raw_plant_b', value=3.5)) + assert set(pod.manager.opc_managers) == {'plant-a', 'plant-b'} + + +async def test_scenario_2_1_4_values_keep_their_opc_type(opc_servers, pods, put_slot, mongo_db): + variables = { + 'Temperature': (20.0, 'double', True), + 'Counter': (1, 'int', True), + 'Running': (False, 'bool', True), + 'Batch': ('B-001', 'string', True), + } + server, _, _ = await _read_slot(opc_servers, pods, put_slot, 'slot_data_types.json', variables) + + await server.set('Temperature', 20.25) + await server.set('Counter', 7) + await server.set('Running', True) + await server.set('Batch', 'B-002') + + expected = {'temperature': 20.25, 'counter': 7, 'running': True, 'batch': 'B-002'} + for name, value in expected.items(): + [doc] = await wait_until( + lambda n=name, v=value: readings(mongo_db, 'raw_types', name=n, value=v) + ) + assert type(doc['value']) is type(value) + + +# 2.2 Slot changes ---------------------------------------------------------------------- + + +async def test_scenario_2_2_1_tag_added_to_the_slot_is_subscribed( + opc_servers, pods, put_slot, mongo_db +): + server, slot, pod = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0, 'Level': 1.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature')) + + slot['plant-a']['tags']['ns=2;s=Level'] = { + 'frequency': 1000, + 'names': [{'collection': 'raw_plant_a', 'tag_name': 'level', 'model_id': '10'}], + } + put_slot(1, slot) + await pod.tick() + + await server.set('Level', 1.75) + await server.set('Temperature', 23.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='level', value=1.75)) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature', value=23.0)) + + +async def test_scenario_2_2_2_tag_removed_from_the_slot_is_no_longer_read( + opc_servers, pods, put_slot, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0, 'Level': 1.0}) + slot = load_scenario_input('slot_single_server.json', {'plant-a': server}) + slot['plant-a']['tags']['ns=2;s=Level'] = { + 'frequency': 1000, + 'names': [{'collection': 'raw_plant_a', 'tag_name': 'level', 'model_id': '10'}], + } + put_slot(1, slot) + pod = await pods('pod-a') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='level')) + + del slot['plant-a']['tags']['ns=2;s=Level'] + put_slot(1, slot) + await pod.tick() + + await server.set('Level', 9.0) + await server.set('Temperature', 24.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature', value=24.0)) + assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', name='level', value=9.0)) + + +async def test_scenario_2_2_3_name_added_to_a_tag_receives_its_readings( + opc_servers, pods, put_slot, mongo_db +): + server, slot, pod = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + slot['plant-a']['tags']['ns=2;s=Temperature']['names'].append( + {'collection': 'raw_dryer', 'tag_name': 'dryer_temperature', 'model_id': '50'} + ) + put_slot(1, slot) + await pod.tick() + + await server.set('Temperature', 26.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=26.0)) + [doc] = await wait_until(lambda: readings(mongo_db, 'raw_dryer', value=26.0)) + assert (doc['name'], doc['model_id']) == ('dryer_temperature', '50') + + +async def test_scenario_2_2_4_server_url_change_reconnects_to_the_new_server( + opc_servers, pods, put_slot, mongo_db +): + old = await opc_servers('plant-a', {'Temperature': 20.0}) + new = await opc_servers('plant-a-new', {'Temperature': 40.0}) + slot = load_scenario_input('slot_single_server.json', {'plant-a': old}) + put_slot(1, slot) + pod = await pods('pod-a') + await pod.tick() + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=20.0)) + + slot['plant-a']['url'] = new.url + put_slot(1, slot) + await pod.tick() + + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=40.0)) + await new.set('Temperature', 41.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=41.0)) + await wait_until(lambda: old.session_count() == 0) + await old.set('Temperature', 21.0) + assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=21.0)) + + +# 2.3 Failures and resilience ----------------------------------------------------------- + + +async def test_scenario_2_3_1_unreachable_server_does_not_stop_the_others( + opc_servers, pods, put_slot, mongo_db +): + plant_a = await opc_servers('plant-a', {'Temperature': 20.0}) + plant_b = await opc_servers('plant-b', {'Flow': 3.0}) + slot = load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b}) + slot['plant-b']['url'] = f'opc.tcp://127.0.0.1:{free_port()}' # nothing listens there + put_slot(1, slot) + + pod = await pods('pod-a') + await pod.tick() + + assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-b') + assert set(pod.manager.opc_managers) == {'plant-a'} + await plant_a.set('Temperature', 27.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=27.0)) + + +async def test_scenario_2_3_2_tag_without_data_raises_a_warning( + opc_servers, pods, put_slot, mongo_db +): + _, _, pod = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + await pod.tick(5) # POLL_INTERVAL 1 s / frequency 1000 ms: one cycle per iteration + + [warning] = notifications(mongo_db, 'TAG_ns=2;s=Temperature:temperature_LISTENNING_STOPPED') + assert warning['level'] in ('WARNING', 'warning') + + +async def test_scenario_2_3_3_silent_server_is_reconnected_after_15_cycles( + opc_servers, pods, put_slot, mongo_db +): + server, _, pod = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + first_client = pod.manager.pool._connections['1'][0] + + await pod.tick(15) + assert notifications(mongo_db, 'OPC_LISTENNING_STOPPED__1') + assert notifications(mongo_db, 'OPC_CONNECTION_RETRY__1') + assert 'plant-a' not in pod.manager.opc_managers + await pod.tick() + + assert 'plant-a' in pod.manager.opc_managers + assert pod.manager.pool._connections['1'][0] is not first_client + await server.set('Temperature', 28.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=28.0)) + + +async def test_scenario_2_3_4_server_restart_resumes_the_reads( + opc_servers, pods, put_slot, mongo_db +): + server, _, pod = await _read_slot( + opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0} + ) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a')) + + await server.stop() + await pod.tick(15) # no data: the server is dropped + assert 'plant-a' not in pod.manager.opc_managers + await pod.tick() # still down: reconnection fails and is notified + assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-a') + + await server.start() + await pod.tick(2) + assert 'plant-a' in pod.manager.opc_managers + await server.set('Temperature', 29.0) + await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=29.0)) diff --git a/e2e/test_writes.py b/e2e/test_writes.py new file mode 100644 index 0000000..3895f95 --- /dev/null +++ b/e2e/test_writes.py @@ -0,0 +1,386 @@ +"""E2E scenarios β€” 3. Writes (NATS C2 β†’ OPC UA β†’ Postgres write_status C3). See ``e2e/scenarios.md``.""" + +import json +from datetime import datetime + +import nats +import pytest +import pytest_asyncio + +from e2e.conftest import WRITE_SUBJECT, nats_server_container +from e2e.helpers import ( + SOURCE_TIMESTAMP_BIT, + STATUS_CODE_BIT, + datavalue_mask, + insert_prediction, + load_scenario_input, + notifications, + prediction_count, + stays_false, + wait_until, + write_status, +) +from e2e.opc_server import free_port + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + +TIMESTAMP = '2026-10-13T12:00:00-03:00' +MODEL_ID = 2 +MARKER_MODEL_ID = 99 + + +@pytest_asyncio.fixture +async def target(opc_servers, mongo_db, postgres_engine): + """plant-a with a writable Setpoint and a read-only Locked, its connector in Mongo + and a pending prediction row (model 2).""" + server = await opc_servers( + 'plant-a', + { + 'Setpoint': 0.0, + 'Counter': (0, 'int', True), + 'Locked': (5.0, 'float', False), + 'Marker': 0.0, + }, + ) + mongo_db['connectors'].insert_one( + load_scenario_input('connector_plant_a.json', {'plant-a': server}) + ) + insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending') + return server + + +def command(**overrides) -> dict: + return load_scenario_input('write_command.json', **overrides) + + +async def processed(publish_command, postgres_engine) -> None: + """Wait until every command published so far was handled: NATS delivers the + messages of a subscription in order, so once a marker command (another row, the + ``Marker`` variable) is recorded, the previous ones are done.""" + insert_prediction(postgres_engine, MARKER_MODEL_ID, TIMESTAMP, 'pending') + await publish_command(command(model_id=str(MARKER_MODEL_ID), addr='ns=2;s=Marker')) + await wait_until(lambda: write_status(postgres_engine, MARKER_MODEL_ID, TIMESTAMP) == 'ok') + + +async def wait_status(postgres_engine, expected: str, model_id: int = MODEL_ID) -> None: + await wait_until(lambda: write_status(postgres_engine, model_id, TIMESTAMP) == expected) + + +# 3.1 DataValue parts: timestamp and quality (SIENTIAPDE-2202) ------------------------ + +DATAVALUE_CASES = [ + pytest.param({}, {}, True, True, id='3_1_1_defaults_on'), + pytest.param({'WRITE_TIMESTAMP': 'false'}, {}, False, True, id='3_1_2_timestamp_off'), + pytest.param({'WRITE_QUALITY': 'false'}, {}, True, False, id='3_1_3_quality_off'), + pytest.param( + {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, + {}, + False, + False, + id='3_1_4_both_off', + ), + pytest.param( + {'WRITE_TIMESTAMP': 'true'}, + {'write_timestamp': False}, + False, + True, + id='3_1_5_command_overrides_env', + ), + pytest.param( + {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, + {'write_timestamp': True, 'write_quality': True}, + True, + True, + id='3_1_6_command_turns_on_over_env', + ), +] + + +@pytest.mark.parametrize('env,flags,expect_timestamp,expect_quality', DATAVALUE_CASES) +async def test_scenario_3_1_datavalue_parts( + target, + pods, + publish_command, + postgres_engine, + opc_wire, + env, + flags, + expect_timestamp, + expect_quality, +): + await pods('pod-a', **env) + + await publish_command(command(**flags)) + + await wait_status(postgres_engine, 'ok') + stored = await target.read('Setpoint') + assert stored.Value.Value == 1.25 + if expect_timestamp: + assert stored.SourceTimestamp == datetime.fromisoformat(TIMESTAMP) + else: + assert stored.SourceTimestamp is None + [request] = opc_wire + mask = datavalue_mask(request, 'ns=2;s=Setpoint') + assert bool(mask & SOURCE_TIMESTAMP_BIT) is expect_timestamp + assert bool(mask & STATUS_CODE_BIT) is expect_quality + + +# 3.2 write_status (C3) ----------------------------------------------------------------- + + +async def test_scenario_3_2_1_pending_row_becomes_ok( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + + await publish_command(command(value=3.5)) + + await wait_status(postgres_engine, 'ok') + assert (await target.read('Setpoint')).Value.Value == 3.5 + + +async def test_scenario_3_2_2_null_write_status_becomes_ok( + opc_servers, mongo_db, pods, publish_command, postgres_engine +): + server = await opc_servers('plant-a', {'Setpoint': 0.0}) + mongo_db['connectors'].insert_one( + load_scenario_input('connector_plant_a.json', {'plant-a': server}) + ) + insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, None) # row from before the column + await pods('pod-a') + + await publish_command(command()) + + await wait_status(postgres_engine, 'ok') + + +@pytest.mark.parametrize('worse', ['error', 'session_bad']) +async def test_scenario_3_2_3_worse_outcome_is_kept( + opc_servers, mongo_db, pods, publish_command, postgres_engine, worse +): + server = await opc_servers('plant-a', {'Setpoint': 0.0, 'Marker': 0.0}) + mongo_db['connectors'].insert_one( + load_scenario_input('connector_plant_a.json', {'plant-a': server}) + ) + insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, worse) + await pods('pod-a') + + await publish_command(command(value=4.0)) + await processed(publish_command, postgres_engine) + + assert (await server.read('Setpoint')).Value.Value == 4.0 # the OPC write happened + assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == worse + + +async def test_scenario_3_2_4_failure_replaces_ok(target, pods, publish_command, postgres_engine): + await pods('pod-a') + await publish_command(command()) + await wait_status(postgres_engine, 'ok') + + await publish_command(command(value='not-a-number')) + + await wait_status(postgres_engine, 'error') + + +async def test_scenario_3_2_5_missing_row_is_not_created( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + + await publish_command(command(model_id='3', value=6.0)) + await processed(publish_command, postgres_engine) + + assert (await target.read('Setpoint')).Value.Value == 6.0 + assert write_status(postgres_engine, 3, TIMESTAMP) == '' + assert prediction_count(postgres_engine) == 2 # model 2 + marker + + +# 3.3 Failed commands ------------------------------------------------------------------- + + +async def test_scenario_3_3_1_unknown_server_is_an_error( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + + await publish_command(command(server_id='404')) + + await wait_status(postgres_engine, 'error') + assert target.writes == [] + + +@pytest.mark.parametrize( + 'connector_change', + [ + pytest.param({'status': 'Inactive'}, id='inactive'), + pytest.param({'type': 'modbus'}, id='not_opc_ua'), + pytest.param({'security': {'mode': 'None'}}, id='no_policy'), + pytest.param({'url': ''}, id='no_url'), + ], +) +async def test_scenario_3_3_2_unusable_connector_is_an_error( + target, mongo_db, pods, publish_command, postgres_engine, connector_change +): + mongo_db['connectors'].update_one({'connector_id': '1'}, {'$set': connector_change}) + await pods('pod-a') + + await publish_command(command()) + + await wait_status(postgres_engine, 'error') + assert target.writes == [] + + +@pytest.mark.parametrize( + 'overrides', + [ + pytest.param( + {'value': 'abc', 'data_type': 'int', 'addr': 'ns=2;s=Counter'}, id='conversion' + ), + pytest.param({'data_type': 'decimal128'}, id='unknown_data_type'), + pytest.param({'addr': None}, id='missing_addr'), + ], +) +async def test_scenario_3_3_3_invalid_command_is_an_error_without_opc_write( + target, pods, publish_command, postgres_engine, overrides +): + await pods('pod-a') + cmd = command(**overrides) + if overrides.get('addr', '') is None: + del cmd['addr'] + + await publish_command(cmd) + + await wait_status(postgres_engine, 'error') + assert target.writes == [] + + +async def test_scenario_3_3_4_bad_timestamp_skips_the_write_and_keeps_the_row( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + + await publish_command(command(timestamp='13/10/2026 12:00')) + await processed(publish_command, postgres_engine) + + assert [w.NodeId.Identifier for w in target.writes] == ['Marker'] + assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == 'pending' + + +@pytest.mark.parametrize( + 'addr', + [ + pytest.param('ns=2;s=Locked', id='not_writable'), + pytest.param('ns=2;s=Missing', id='unknown_node'), + ], +) +async def test_scenario_3_3_5_opc_rejection_is_an_error( + target, pods, publish_command, postgres_engine, addr +): + await pods('pod-a') + + await publish_command(command(addr=addr)) + + await wait_status(postgres_engine, 'error') + assert (await target.read('Locked')).Value.Value == 5.0 + + +async def test_scenario_3_3_6_malformed_json_is_dropped_and_the_consumer_keeps_going( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + + await publish_command(b'{"server_id": "1", "value": ') + await publish_command(command(value=7.0)) + + await wait_status(postgres_engine, 'ok') + assert (await target.read('Setpoint')).Value.Value == 7.0 + + +# 3.4 Connections ----------------------------------------------------------------------- + + +async def test_scenario_3_4_1_write_reuses_the_read_connection( + opc_servers, pods, put_slot, publish_command, postgres_engine, mongo_db +): + server = await opc_servers('plant-a', {'Temperature': 20.0, 'Setpoint': 0.0}) + put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) + insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending') + pod = await pods('pod-a') + await pod.tick() + assert mongo_db['connectors'].count_documents({}) == 0 # only the slot knows plant-a + + await publish_command(command()) + + await wait_status(postgres_engine, 'ok') + assert server.session_count() == 1 + + +async def test_scenario_3_4_2_write_only_server_gets_its_own_connection( + target, pods, publish_command, postgres_engine +): + pod = await pods('pod-a') + assert target.session_count() == 0 + + await publish_command(command()) + + await wait_status(postgres_engine, 'ok') + assert target.session_count() == 1 + assert pod.manager.pool.config_for('1')['url'] == target.url + + +# 3.5 Throughput and scaling ------------------------------------------------------------ + + +async def test_scenario_3_5_1_min_pause_between_writes( + target, pods, publish_command, postgres_engine +): + await pods('pod-a', WRITE_MIN_PAUSE_S='0.5') + + for value in (1.0, 2.0, 3.0): + await publish_command(command(value=value)) + + await wait_until(lambda: len(target.writes) == 3) + gaps = [b - a for a, b in zip(target.write_times, target.write_times[1:], strict=False)] + assert all(gap >= 0.45 for gap in gaps), gaps + + +async def test_scenario_3_5_2_queue_group_writes_each_command_once( + target, pods, publish_command, postgres_engine +): + await pods('pod-a') + await pods('pod-b') + + for value in range(6): + await publish_command(command(value=float(value))) + + await wait_until(lambda: len(target.writes) == 6) + assert await stays_false(lambda: len(target.writes) > 6) + + +# 3.6 NATS availability ----------------------------------------------------------------- + + +async def test_scenario_3_6_1_nats_down_at_startup_connects_later( + target, pods, postgres_engine, mongo_db +): + port = free_port() + url = f'nats://127.0.0.1:{port}' + pod = await pods('pod-a', NATS_URL=url) # nothing listens yet + await pod.tick() # retried by idle_sweep, notified once per outage + + assert len(notifications(mongo_db, 'GATEWAY_NATS_CONSUMER_START_FAILED')) == 1 + nats_server = nats_server_container(host_port=port) + nats_server.start() + try: + await pod.tick() + assert pod.runtime._nats_started + + client = await nats.connect(url) + cmd = load_scenario_input('write_command.json') + await client.publish(WRITE_SUBJECT, json.dumps(cmd).encode()) + await client.flush() + await wait_status(postgres_engine, 'ok') + await client.drain() + finally: + await pod.stop() + nats_server.stop() diff --git a/pyproject.toml b/pyproject.toml index a7cc499..8c6c19c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,11 @@ ignore = [ "S105", # hardcoded passwords ok in tests "S106", # hardcoded passwords ok in tests ] +"e2e/**/*.py" = [ + "S101", # assert allowed in tests + "S105", # hardcoded passwords ok in tests + "S106", # hardcoded passwords ok in tests +] [tool.ruff.lint.mccabe] max-complexity = 15 @@ -129,7 +134,7 @@ addopts = [ ] markers = [ "asyncio: marks tests as async", - "integration: marks tests as integration tests", + "integration: marks tests as integration tests (the e2e/ suite: Docker required, run with -o testpaths=e2e -m integration)", "unit: marks tests as unit tests", ] diff --git a/requirements-dev.txt b/requirements-dev.txt index 09381f5..377959a 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -13,6 +13,8 @@ types-requests>=2.31.0 # Type stubs for requests pytest>=7.4.0 # Testing framework pytest-cov>=4.1.0 # Coverage plugin for pytest pytest-asyncio>=0.21.0 # Async test support (already in main requirements) +testcontainers[postgres,minio,mongodb,redis]>=4.15.0 # E2E suite (e2e/): real MongoDB, Redis, PostgreSQL, + # MinIO and NATS containers; `testcontainers.community.*` namespace # Development Tools ipython>=8.12.0 # Enhanced Python shell diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py deleted file mode 100644 index 217938f..0000000 --- a/tests/functional/conftest.py +++ /dev/null @@ -1,31 +0,0 @@ -# import subprocess -# from time import sleep -# import pytest -# from redis import Redis - - -# @pytest.fixture(scope="session", autouse=True) -# def docker_compose(): -# """Sobe os containers antes dos testes e derruba depois.""" -# print("\nπŸš€ Subindo Docker Compose...") -# subprocess.run(["docker", "compose", "up", "-d"], check=True) - -# print("⏳ Aguardando containers ficarem prontos...") -# sleep(15) # ajuste conforme necessΓ‘rio - -# yield # os testes rodam aqui - -# print("\n🧹 Derrubando Docker Compose...") -# subprocess.run(["docker", "compose", "down"], check=True) - - -# @pytest.fixture() -# def redis_client(): -# redis = Redis(host="localhost", port=6379, decode_responses=True) -# redis.flushdb() - -# yield redis - -# # Limpa o banco de dados apΓ³s os testes -# redis.flushdb() -# redis.close() diff --git a/tests/functional/test_single_node.py b/tests/functional/test_single_node.py deleted file mode 100644 index 330a815..0000000 --- a/tests/functional/test_single_node.py +++ /dev/null @@ -1,93 +0,0 @@ -# import json -# import subprocess -# from time import sleep - - -# new_data = { -# "slot:opc_tags:1": { -# "server1": { -# "name": "server1", -# "url": "opc.tcp://simulator:4840", -# "server_uri": "http://opcua-server.simulator", -# "tags": { -# 'ns=2;i=2': { -# 'tag_name': 'Counter', -# 'frequency': 1000, -# 'topics': [], -# }, -# 'ns=2;i=3': { -# 'tag_name': 'Rollout', -# 'frequency': 1000, -# "topics": [], -# }, -# 'ns=2;i=4': { -# 'tag_name': 'Square', -# 'frequency': 1000, -# "topics": [], -# }, -# } -# } -# }, -# "slot:opc_tags:2": { -# "server2": { -# "name": "server2", -# "url": "opc.tcp://simulator:4840", -# "server_uri": "http://opcua-server.simulator", -# "tags": { -# 'ns=2;i=2': { -# 'tag_name': 'Counter', -# 'frequency': 1000, -# 'topics': [], -# }, -# 'ns=2;i=3': { -# 'tag_name': 'Rollout', -# 'frequency': 1000, -# "topics": [], -# }, -# 'ns=2;i=4': { -# 'tag_name': 'Square', -# 'frequency': 1000, -# "topics": [], -# }, -# } -# } -# } -# } - - -# def test_simple(redis_client): -# new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [ -# 'test_topic_1'] -# redis_client.set("slot:opc_tags:1", -# json.dumps(new_data['slot:opc_tags:1'])) - -# sleep(20) # Espera o Ingestor processar os dados - -# # Check if lease is in Redis -# assert redis_client.get("lease:opc_tags:1") == 'ingestor' -# assert redis_client.get("heartbeat:ingestor:ingestor") == '1' - - -# def test_simple_double_slot(redis_client): - -# new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [ -# 'test_topic_double_slot1'] -# redis_client.set("slot:opc_tags:1", -# json.dumps(new_data['slot:opc_tags:1'])) - -# sleep(20) # Espera o Ingestor processar os dados - -# assert redis_client.get("lease:opc_tags:1") == 'ingestor' -# assert redis_client.get("heartbeat:ingestor:ingestor") == '1' - -# new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [ -# 'test_topic_double_slot2'] -# redis_client.set("slot:opc_tags:2", -# json.dumps(new_data['slot:opc_tags:2'])) - -# sleep(20) # Espera o Ingestor processar os dados - -# # Check if lease is in Redis -# assert redis_client.get("lease:opc_tags:2") == 'ingestor' -# assert redis_client.get("lease:opc_tags:1") == 'ingestor' -# assert redis_client.get("heartbeat:ingestor:ingestor") == '1' diff --git a/tests/functional/test_write_e2e.py b/tests/functional/test_write_e2e.py deleted file mode 100644 index f57ad6f..0000000 --- a/tests/functional/test_write_e2e.py +++ /dev/null @@ -1,173 +0,0 @@ -"""End-to-end OPC write (SIENTIAPDE-2202): env β†’ GatewayConfig β†’ WriteEngine β†’ -OpcConnectionPool β†’ a real asyncua server, in process. - -The server keeps the SourceTimestamp it receives, so the timestamp is checked on the -server. An omitted StatusCode decodes as Good on the server (OPC UA default), so the -quality is checked on the wire: the DataValue encoding mask of the WriteRequest the -client sent. -""" - -import socket -from datetime import datetime -from unittest.mock import AsyncMock, MagicMock - -import pytest -import pytest_asyncio -from asyncua import Server, ua -from asyncua.client import ua_client -from asyncua.ua.ua_binary import Primitives, nodeid_to_binary - -from ingestor.config import GatewayConfig -from ingestor.connection.pool import OpcConnectionPool -from ingestor.engines.write import WriteEngine - -TIMESTAMP = '2026-07-13T12:00:00Z' - -# DataValue encoding mask bits (OPC UA Part 6, 5.2.2.17). -_STATUS_CODE_BIT = 0x02 -_SOURCE_TIMESTAMP_BIT = 0x04 - - -def _free_port() -> int: - with socket.socket() as s: - s.bind(('127.0.0.1', 0)) - return s.getsockname()[1] - - -@pytest_asyncio.fixture -async def opc_server(): - url = f'opc.tcp://127.0.0.1:{_free_port()}' - server = Server() - await server.init() - server.set_endpoint(url) - idx = await server.register_namespace('urn:sientia:e2e') - var = await server.nodes.objects.add_variable(idx, 'Pred', 0.0, ua.VariantType.Float) - await var.set_writable() - async with server: - yield url, var - - -@pytest.fixture -def wire(monkeypatch): - """Serialized WriteRequests, as sent by the client.""" - sent: list[bytes] = [] - original = ua_client.struct_to_binary - - def capture(obj): - data = original(obj) - if isinstance(obj, ua.WriteRequest): - sent.append(data) - return data - - monkeypatch.setattr(ua_client, 'struct_to_binary', capture) - return sent - - -def _datavalue_mask(request: bytes, addr: str) -> int: - # WriteValue = NodeId, AttributeId, IndexRange, DataValue: the byte right after - # the first three is the DataValue encoding mask. The NodeId is parsed from the - # address as the client does (its binary encoding depends on how it was built). - prefix = ( - nodeid_to_binary(ua.NodeId.from_string(addr)) - + Primitives.UInt32.pack(ua.AttributeIds.Value) - + Primitives.String.pack(None) - ) - return request[request.index(prefix) + len(prefix)] - - -async def _engine(url: str, pools: list) -> WriteEngine: - config = GatewayConfig.from_env() - logger, notification_handler = MagicMock(), MagicMock() - # emit() is awaited, start()/shutdown_sync() are not. - metrics_controller = MagicMock(emit=AsyncMock(), shutdown=AsyncMock()) - pool = OpcConnectionPool(MagicMock(), logger, notification_handler, metrics_controller) - pools.append(pool) - servers = MagicMock() - servers.resolve = AsyncMock( - return_value={ - 'server_id': 'e2e', - 'url': url, - 'security': {'mode': 'None', 'policy': 'None'}, - 'certificate': None, - } - ) - return WriteEngine( - pool, - AsyncMock(), - logger, - notification_handler, - metrics_controller, - write_timestamp=config.write_timestamp, - write_quality=config.write_quality, - servers=servers, - ) - - -@pytest_asyncio.fixture -async def pools(): - created: list[OpcConnectionPool] = [] - yield created - for pool in created: - for client, _ in pool._connections.values(): - await client.disconnect() - - -@pytest.mark.integration -@pytest.mark.asyncio -@pytest.mark.parametrize( - 'env,cmd_flags,expect_timestamp,expect_quality', - [ - pytest.param({}, {}, True, True, id='defaults-on'), - pytest.param({'WRITE_TIMESTAMP': 'false'}, {}, False, True, id='timestamp-off'), - pytest.param({'WRITE_QUALITY': 'false'}, {}, True, False, id='quality-off'), - pytest.param( - {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, {}, False, False, id='both-off' - ), - pytest.param( - {'WRITE_TIMESTAMP': 'true'}, - {'write_timestamp': False}, - False, - True, - id='command-overrides-env', - ), - pytest.param( - {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, - {'write_timestamp': True, 'write_quality': True}, - True, - True, - id='command-turns-on-over-env', - ), - ], -) -async def test_write_datavalue_parts( - monkeypatch, opc_server, wire, pools, env, cmd_flags, expect_timestamp, expect_quality -): - monkeypatch.delenv('WRITE_TIMESTAMP', raising=False) - monkeypatch.delenv('WRITE_QUALITY', raising=False) - for name, value in env.items(): - monkeypatch.setenv(name, value) - url, var = opc_server - engine = await _engine(url, pools) - cmd = { - 'server_id': 'e2e', - 'addr': var.nodeid.to_string(), - 'value': 1.5, - 'data_type': 'float', - 'model_id': 'm1', - 'timestamp': TIMESTAMP, - **cmd_flags, - } - - result = await engine.write(cmd) - - assert result.status == 'ok', result.message - stored = await var.read_data_value() - assert stored.Value.Value == 1.5 - if expect_timestamp: - assert stored.SourceTimestamp == datetime.fromisoformat(TIMESTAMP) - else: - assert stored.SourceTimestamp is None - assert len(wire) == 1 - mask = _datavalue_mask(wire[0], cmd['addr']) - assert bool(mask & _SOURCE_TIMESTAMP_BIT) is expect_timestamp - assert bool(mask & _STATUS_CODE_BIT) is expect_quality