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>
267 lines
10 KiB
Python
267 lines
10 KiB
Python
"""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))
|