vitor-aignosi
4ee672a715
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 <noreply@anthropic.com>
85 lines
3.5 KiB
Python
85 lines
3.5 KiB
Python
"""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]
|