docs(gateway): e2e validation guide + seed script for the KepServer session

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 '<json printed by the seed script>'
+```
+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`.

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())