diff --git a/docs/e2e-test-guide.md b/docs/e2e-test-guide.md new file mode 100644 index 0000000..93e848f --- /dev/null +++ b/docs/e2e-test-guide.md @@ -0,0 +1,94 @@ +# OPC Gateway — E2E Validation Guide (SIENTIAPDE-1972) + +Session script for validating the gateway against a real KepServer + cluster (with the feature owner). Each step names the command and the expected result. Run order matters. + +**In scope:** cert management, subscription (legacy slot v1), NATS write → OPC → `write_status`, gRPC ReadTag, 1879 write flags, resilience, metrics. +**Out of scope (integration-gated):** multi-name subscription / scheduled active reads / gRPC persist (need slot v2 → SIENTIAPDE-1974); the real Laborious publish (SIENTIAPDE-1975) — here we hand-publish the C2 message, which *is* the contract. + +--- + +## 0. Prerequisites + +| What | How | +|---|---| +| KepServer reachable | `OPC_SERVER_URL` (default `opc.tcp://localhost:49320`); Simulation Examples tags enabled | +| NATS (test instance) | `docker run -d --name nats -p 4222:4222 nats:2` — no JetStream needed (Core NATS). Cluster provisioning is SIENTIAPDE-1985; this container is only for the session | +| `nats` CLI | [natscli](https://github.com/nats-io/natscli) or `docker exec` a `nats-box` | +| `grpcurl` | for the ReadTag call | +| Mongo/Redis/Postgres | same instances the app already uses (`.env`) | +| **MinIO policy (pre-req if testing security)** | frontend user needs write on bucket `sientia-dataops`; gateway reads it. The 1569 CRUD now defaults to `sientia-dataops` (`MINIO_BUCKET` env overrides) | + +## 1. Seed test data + +```bash +python scripts/seed_e2e.py --print-only # inspect first +python scripts/seed_e2e.py # seed Mongo + Redis slot v1 + Postgres table +``` +Creates: connector `e2e-kepserver` (9001), 2 tags **with `metadata: {}`**, slot `slot:opc_tags:1` (v1), and `sientia_data.e2e_predictions` with the `write_status` column + 1 row (`model_id='9001'`, `timestamp='2026-07-15T12:00:00Z'`). + +> Adjust the two tag `address` values to real KepServer node ids if the Simulation Examples channel differs. + +## 2. Start the gateway + +```bash +# .env: MONGODB_*/REDIS_* as today, plus NATS_URL=nats://localhost:4222, GRPC_ENABLED=true +python -m ingestor.app +``` +Expect in logs: `NatsWriteConsumer started: subject=opc.write queue=gateway-writers`, `GrpcServer started on port 50051`, `GatewayRuntime started: nats_started=True grpc_enabled=True`. + +Health probe (what k8s will run): `curl -sf http://localhost:9090/metrics | grep '^app_up'` → `app_up{...} 1.0`. + +## 3. Subscription (legacy v1 path) + +The slot from step 1 subscribes `e2e_ramp`. Expect within ~1 cycle: +- Mongo collection `raw_e2e-gateway-test` receiving documents `{tag, name, timestamp, value}`. +- Metrics: `opc_connection_status{...} 1`, `opc_tags_subscribed_current` ≥ 1. + +## 4. Write via NATS (C2 → OPC → C3) + +The seed prints the exact command; it targets the seeded predictions row: +```bash +nats pub opc.write '' +``` +Expect: +1. KepServer: `e2e_write_target` (User1) becomes `42.5`. +2. Postgres: `SELECT write_status FROM sientia_data.e2e_predictions WHERE model_id='9001';` → `ok`. +3. Metrics: `opc_write_total{result="success"}` +1, `opc_write_latency_seconds` observed, `opc_write_status_by_model{status="ok"} 1`. + +**1879 flag checks** (repeat the publish varying the payload): +- `"write_timestamp": false` / `"write_quality": false` → KepServer must still accept (or reject — that's the point of testing against the real server; compare behaviors). +- Bad value `"value": "abc"` → no crash; `write_status='error'`, `opc_error_by_type_total{error_type="conversion"}` +1. +- Malformed JSON (`nats pub opc.write 'not-json'`) → dropped + logged, app alive. +- Missing field (remove `"addr"`) → `write_status='error'` (`bad_command`), no crash. +- Two rapid publishes with `WRITE_MIN_PAUSE_S=1` set → second write delayed ≥1s (rate-limit). + +## 5. gRPC on-demand read + +```bash +grpcurl -plaintext -d '{"server_id":"9001","addr":"ns=2;s=Simulation Examples.Functions.Ramp1"}' \ + localhost:50051 opcgateway.OpcGateway/ReadTag +``` +Expect: `value`, `opc_status_code: "Good"`, timestamps, **`persisted: false`** (correct until 1974). Metrics: `opc_active_read_total{result="success"}` +1, `opc_read_lag_seconds` set. +Error case: bad addr → gRPC `INTERNAL`, `opc_active_read_total{result="error"}` +1, app alive. + +## 6. Certificates (round 2 — security on) + +1. Upload platform cert via the connectors CRUD (writes to `sientia-dataops` now) — or place objects under `platform/certificates/` + the single `platform_certificates` Mongo doc manually. +2. Flip the seeded connector/slot to `use_security: true`, restart the gateway. +3. Expect Basic256Sha256 session on KepServer (check its connection log); reads/writes as in steps 3–5. + +## 7. Resilience + +- **NATS down:** `docker stop nats` → app keeps reading (subscription + gRPC fine); ONE notification `GATEWAY_NATS_CONSUMER_START_FAILED`; `docker start nats` → consumer reconnects on next loop (~5s); publish works again. +- **MinIO down at boot** (security round): gateway logs `GatewayRuntime failed to initialize, continuing without gateway` and the read path still runs. +- **KepServer restart:** subscription reconnect kicks in after ~15 idle cycles (`opc_reconnections_tried_total`). + +## 8. Wrap-up checks + +- `curl -s localhost:9090/metrics | grep -E 'opc_(write|active_read|error|tag_value|read_lag)'` — all families populated. +- No `kafka` anywhere: `grep -ri kafka ingestor/` → empty. +- Postgres row still has exactly the statuses produced by step 4 (no stray writes). + +--- + +**Ledger:** phases/commits in `docs/superpowers/plans/2026-07-13-opc-gateway.md`; deploy details in `docs/deploy-notes.md`. diff --git a/scripts/seed_e2e.py b/scripts/seed_e2e.py new file mode 100644 index 0000000..0659399 --- /dev/null +++ b/scripts/seed_e2e.py @@ -0,0 +1,154 @@ +"""Seed test data for the OPC Gateway e2e validation session (SIENTIAPDE-1972). + +Populates: + 1. Mongo `connectors` + `connector_tags` (each tag with `metadata: {}` — required + for documentation, per the feature owner). + 2. Redis slot v1 (`slot:opc_tags:1`) so the legacy subscription path collects them. + 3. Postgres test predictions table (with the `write_status` column) + one row the + NATS write test can UPDATE (correlated by timestamp + model_id). + +Idempotent: connectors/tags are upserted, the slot is overwritten, the table uses +CREATE IF NOT EXISTS + ON CONFLICT DO NOTHING. + +Usage (from the repo root, sientia-dataops conda env): + python scripts/seed_e2e.py --print-only # dry-run: show docs/SQL, touch nothing + python scripts/seed_e2e.py # seed Mongo + Redis + Postgres + +Env (same names the gateway/app use): + MONGODB_URL/MONGODB_USERNAME/MONGODB_PASSWORD/MONGODB_DATABASE + REDIS_HOST/REDIS_PORT POSTGRES_HOST/PORT/USER/PASSWORD/DBNAME + OPC_SERVER_URL (default opc.tcp://localhost:49320 — KepServer default port) + +# ponytail: plain sync pymongo/redis/sqlalchemy, no sientia_do async plumbing — +# this is a test fixture, not production code. +""" + +import argparse +import json +import os +import sys + +OPC_URL = os.getenv('OPC_SERVER_URL', 'opc.tcp://localhost:49320') + +CONNECTOR = { + 'connector_id': 9001, + 'type': 'opc_ua', + 'name': 'e2e-kepserver', + 'url': OPC_URL, + 'server_uri': 'urn:e2e:gateway:test', + 'subscription_period_ms': 1000, + 'use_security': False, # round 1; flip after certs are in sientia-dataops + 'status': 'Active', + 'description': 'OPC Gateway e2e validation connector (SIENTIAPDE-1972)', +} + +# metadata: {} on every tag — feature owner asked test tags to carry the field. +TAGS = [ + {'tag_id': 9101, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.Ramp1', + 'tag_name': 'e2e_ramp', 'category': 'Read', 'data_type': 'float', 'unit': '-', + 'status': 'Active', 'description': 'e2e read tag', 'metadata': {}}, + {'tag_id': 9102, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.User1', + 'tag_name': 'e2e_write_target', 'category': 'Write', 'data_type': 'float', 'unit': '-', + 'status': 'Active', 'description': 'e2e write tag', 'metadata': {}}, +] + +# Slot v1 (legacy path the gateway still runs until SIENTIAPDE-1974 delivers slot v2). +SLOT_V1 = { + 'e2e-kepserver': { + 'name': 'e2e-kepserver', + 'server_id': '9001', + 'url': OPC_URL, + 'server_uri': 'urn:e2e:gateway:test', + 'subscription_period_ms': 1000, + 'tags': { + TAGS[0]['address']: { + 'tag_name': TAGS[0]['tag_name'], + 'frequency': '5000', + 'topics': ['raw_e2e-gateway-test'], + } + }, + } +} + +PREDICTIONS_SQL = """ +CREATE SCHEMA IF NOT EXISTS sientia_data; +CREATE TABLE IF NOT EXISTS sientia_data.e2e_predictions ( + model_id text NOT NULL, + prediction double precision, + prediction_confidence double precision, + timestamp text NOT NULL, + write_status text, + PRIMARY KEY (model_id, timestamp) +); +INSERT INTO sientia_data.e2e_predictions (model_id, prediction, prediction_confidence, timestamp) +VALUES ('9001', 42.5, 0.97, '2026-07-15T12:00:00Z') +ON CONFLICT DO NOTHING; +""" + +# The matching NATS write command for the row above (paste into `nats pub opc.write '...'`). +NATS_CMD = { + 'msg_id': 'e2e-0001', 'server_id': '9001', 'addr': TAGS[1]['address'], + 'value': 42.5, 'data_type': 'float', 'tag_type': 'prediction', + 'model_id': '9001', 'model_name': 'e2e-model', + 'schema': 'sientia_data', 'table_name': 'e2e_predictions', + 'timestamp': '2026-07-15T12:00:00Z', + 'url': OPC_URL, 'use_security': False, 'server_uri': 'urn:e2e:gateway:test', +} + + +def seed_mongo(): + from pymongo import MongoClient + + url = os.getenv('MONGODB_URL', 'localhost:27017') + user, pwd = os.getenv('MONGODB_USERNAME', 'sientia'), os.getenv('MONGODB_PASSWORD', 'sientia') + db = MongoClient(f'mongodb://{user}:{pwd}@{url}')[os.getenv('MONGODB_DATABASE', 'sientia')] + db['connectors'].replace_one({'connector_id': CONNECTOR['connector_id']}, CONNECTOR, upsert=True) + for tag in TAGS: + db['connector_tags'].replace_one({'tag_id': tag['tag_id']}, tag, upsert=True) + print(f'mongo: upserted 1 connector + {len(TAGS)} tags (metadata: {{}})') + + +def seed_redis(): + import redis + + r = redis.Redis(host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', '6379'))) + r.set('slot:opc_tags:1', json.dumps(SLOT_V1)) + print('redis: wrote slot:opc_tags:1 (v1)') + + +def seed_postgres(): + from sqlalchemy import create_engine, text + + host, port = os.getenv('POSTGRES_HOST', 'localhost'), os.getenv('POSTGRES_PORT', '5432') + user, pwd = os.getenv('POSTGRES_USER', 'sientia'), os.getenv('POSTGRES_PASSWORD', 'sientia') + dbname = os.getenv('POSTGRES_DBNAME', 'sientia') + engine = create_engine(f'postgresql://{user}:{pwd}@{host}:{port}/{dbname}') + with engine.begin() as conn: + for stmt in filter(str.strip, PREDICTIONS_SQL.split(';')): + conn.execute(text(stmt)) + print('postgres: sientia_data.e2e_predictions ready (write_status column + 1 row)') + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--print-only', action='store_true', help='show docs/SQL, touch nothing') + args = parser.parse_args() + + if args.print_only: + print('connector:', json.dumps(CONNECTOR, indent=2)) + print('tags:', json.dumps(TAGS, indent=2)) + print('slot v1:', json.dumps(SLOT_V1, indent=2)) + print('sql:', PREDICTIONS_SQL) + print('nats write cmd:', json.dumps(NATS_CMD, indent=2)) + return 0 + + seed_mongo() + seed_redis() + seed_postgres() + print('\nNATS write command for the session:') + print(f"nats pub opc.write '{json.dumps(NATS_CMD)}'") + return 0 + + +if __name__ == '__main__': + sys.exit(main())