SIENTIAPDE-1974-connectors-pi-web-api: Consolidate OPC server and tag data sourcing to the tag database. This change removes the legacy opc-servers collection and enforces tag_id as the sole reference for all read and write tags via connectors and connector_tags. The Orchestrator workflow now exclusively loads OPC server and tag information from the tag database, simplifying data flow and removing conditional logic for legacy data sources. Improved error reporting for pipeline timestamp updates and updated documentation, examples, and tests are also included.
7.0 KiB
E2E Scenario Documentation — Orchestrator
Functional reference for orchestrator E2E scenarios. Tests live under e2e/, use @pytest.mark.e2e, and run with:
pytest e2e/ --override-ini testpaths=e2e -m e2e
Every section below maps 1:1 to an actual test function (named in each section, or as a bullet
under a Source: heading) — this file describes what the suite verifies today, not planned or
aspirational coverage. 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
- MongoDB, Redis, PostgreSQL: testcontainers (session-scoped).
- SMTP: in-process
aiosmtpd(e2e/smtp_test_server.py). - Temporal:
WorkflowEnvironment.start_local()with stub workers onscouter/laborious. - Production code under
orchestrator/**is not mocked; onlyLoggermay be aMagicMock.
1. Orchestrator workflow
Source: e2e/test_orchestrator_main_workflow.py
1.1.1 Happy path
Pipelines in Mongo → both Temporal schedules created in the laborious namespace, one
orchestrated_schedules entry per pipeline, and the predictions_batch write tag (tag_id)
resolved into opc_output_config through the tag database.
1.3.1 Create-only
New pipeline only → Temporal schedule created for it.
1.3.3 Delete-only
Pipeline removed from Mongo → previously-created Temporal schedule deleted.
1.5.1 Empty pipelines
No active pipelines → no orchestrated_schedules documents written, even with a pre-existing
unrelated Temporal schedule present.
1.6.1 TTL index bootstrap
First run creates TTL index on notification collection used by scouter pipelines.
1.7.1 Simple metrics parallel windows
Two simple_metrics pipelines for the same model_id (60 min and 1440 min windows) → one
Temporal schedule per window, each with its own frequency as interval.
1.7.2 Adaptive frequency escalation/recovery round trip
Laborious-style write (adaptive_state.escalated=true + BSON-datetime updated_at) on one
window → next run switches only that schedule to frequency_alert; the recovery write switches it
back to frequency.
1.8.1 Drift window and chunk_period
Two drift pipelines on frequency='1m': one without interval_minutes and with an invalid
chunk_period='hour' → workflow input interval=1 (frequency, no overlap),
chunk_period='min' and exactly one DRIFT_CHUNK_PERIOD_INVALID notification; one with
interval_minutes=60 and chunk_period='s' → both forwarded unchanged, no warning
(SIENTIAPDE-2212).
1.9 OPC tag database: read and write tags by tag_id
Scenario input: orchestrator_tag_database.json (one opc_ua connector, its connector_tags of
every category, one inactive, an OPC scouter reading by tag_id and a predictions_batch writing by
tag_id, with a Read tag, an inactive tag, an unknown tag_id and an entry without tag_id among its
write tags).
1.9.1 Write tags through the tag database
write_tags: [{tag_id}] → opc_output_config keyed by connector_id, address and data_type
from connector_tags (float when absent), Write → prediction_tags, Prediction Confidence →
confidence_tags. The Read, inactive and unknown tags are left out and notified once
(ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION); the entry without tag_id is notified
(ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION).
1.9.2 A tag change redeploys its pipelines
A second run with nothing changed redeploys nothing (orchestrated_schedules is not stamped
again). Then renaming the read tag, moving the write tag and deleting the confidence tag in
connector_tags by hand (no updated_at bumped anywhere, pipelines untouched) → the next run
redeploys both through their config_hash: the scouter model_tags and the slot names take
the new name, the opc_output_config the new address without the confidence tag, and
orchestrated_schedules records the new config_hash.
1.9.3 OPC slot from the tag database
The slot v2 written to Redis holds only the active Read tag, with the connector's server fields.
2. Alerts workflow
Source: e2e/test_alerts_main_workflow.py
A.1.1 Happy path
ERROR notification → one SMTP message, one log_report row, Redis cache key.
A.1.2 TTL duplicate suppression
Second run with same data and cache seeded → no extra email or log row.
A.2.1 Group filtering
Receiver group levels / ignore_models_id honored, both halves now covered:
the query-time union/intersection across active groups (SIENTIAPDE-2011, see
LoadNotificationPackage below) and the per-group precise slice
(SIENTIAPDE-2012) — a group only receives notifications matching its own
levels/ignore_models_id, even when the query-time filter widened the fetch
to satisfy another active group's broader configuration.
A.2.2 Model_id allow-list
Receiver group model_ids (SIENTIAPDE-2149) honored: a group configuring it only
receives notifications from those models, even when another active group with
no such restriction widens the query-time fetch to include every model.
A.3.1 Empty queue
No notifications → no SMTP, no Postgres row.
3. Reports workflow
Source: e2e/test_reports_main_workflow.py
R.1.1 Happy path
Mixed ERROR/WARNING/INFO → one HTML email with all section headings.
R.1.2 Per-level rendering
Single-level notifications → only matching section in HTML body.
R.2.1 Empty queue
No notifications → no SMTP, no Postgres row.
4. Subworkflows
LoadNotificationPackage — e2e/test_subworkflow_load_notification_package.py
- No prior Redis timestamp → all matching notifications returned, max timestamp stored.
- Prior timestamp → only newer notifications returned.
- Query-time filter (SIENTIAPDE-2011) → union of active groups'
levels, intersection of theirignore_models_id; a model ignored by only one of several active groups is not excluded from the fetch. - Model_id allow-list union at query time (SIENTIAPDE-2149) → when every active group sets
model_ids, the fetch is restricted to the union of every group's allow-list; a model outside that union is never fetched at all, even for a group that would otherwise want it via a different axis. - Empty Mongo → no Redis timestamp write.
- Batch size cap (SIENTIAPDE-2013) → with 5 pending notifications and a batch size of 2, each run returns exactly the next-oldest 2 (sorted ascending by timestamp), and the Redis cursor advances only to what was actually returned, so three successive runs drain the backlog in order instead of skipping anything.
- Batch size 0 means unlimited (SIENTIAPDE-2013) → the configured default (
0) does not cap the query; 12 pending notifications all come back in one run, verified against a real MongoDB instance rather than assumed from pymongo's documented convention.
ProcessNotifications — e2e/test_subworkflow_process_notifications.py
- Full round trip: HTML → SMTP →
log_reportin Postgres. - Empty receiver groups →
{}.