"""E2E scenarios — 3. Writes (NATS C2 → OPC UA → Postgres write_status C3). See ``e2e/scenarios.md``.""" import json from datetime import datetime import nats import pytest import pytest_asyncio from e2e.conftest import WRITE_SUBJECT, nats_server_container from e2e.helpers import ( SOURCE_TIMESTAMP_BIT, STATUS_CODE_BIT, datavalue_mask, insert_prediction, load_scenario_input, notifications, prediction_count, stays_false, wait_until, write_status, ) from e2e.opc_server import free_port pytestmark = [pytest.mark.integration, pytest.mark.asyncio] TIMESTAMP = '2026-10-13T12:00:00-03:00' MODEL_ID = 2 MARKER_MODEL_ID = 99 @pytest_asyncio.fixture async def target(opc_servers, mongo_db, postgres_engine): """plant-a with a writable Setpoint and a read-only Locked, its connector in Mongo and a pending prediction row (model 2).""" server = await opc_servers( 'plant-a', { 'Setpoint': 0.0, 'Counter': (0, 'int', True), 'Locked': (5.0, 'float', False), 'Marker': 0.0, }, ) mongo_db['connectors'].insert_one( load_scenario_input('connector_plant_a.json', {'plant-a': server}) ) insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending') return server def command(**overrides) -> dict: return load_scenario_input('write_command.json', **overrides) async def processed(publish_command, postgres_engine) -> None: """Wait until every command published so far was handled: NATS delivers the messages of a subscription in order, so once a marker command (another row, the ``Marker`` variable) is recorded, the previous ones are done.""" insert_prediction(postgres_engine, MARKER_MODEL_ID, TIMESTAMP, 'pending') await publish_command(command(model_id=str(MARKER_MODEL_ID), addr='ns=2;s=Marker')) await wait_until(lambda: write_status(postgres_engine, MARKER_MODEL_ID, TIMESTAMP) == 'ok') async def wait_status(postgres_engine, expected: str, model_id: int = MODEL_ID) -> None: await wait_until(lambda: write_status(postgres_engine, model_id, TIMESTAMP) == expected) # 3.1 DataValue parts: timestamp and quality (SIENTIAPDE-2202) ------------------------ DATAVALUE_CASES = [ pytest.param({}, {}, True, True, id='3_1_1_defaults_on'), pytest.param({'WRITE_TIMESTAMP': 'false'}, {}, False, True, id='3_1_2_timestamp_off'), pytest.param({'WRITE_QUALITY': 'false'}, {}, True, False, id='3_1_3_quality_off'), pytest.param( {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, {}, False, False, id='3_1_4_both_off', ), pytest.param( {'WRITE_TIMESTAMP': 'true'}, {'write_timestamp': False}, False, True, id='3_1_5_command_overrides_env', ), pytest.param( {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'}, {'write_timestamp': True, 'write_quality': True}, True, True, id='3_1_6_command_turns_on_over_env', ), ] @pytest.mark.parametrize('env,flags,expect_timestamp,expect_quality', DATAVALUE_CASES) async def test_scenario_3_1_datavalue_parts( target, pods, publish_command, postgres_engine, opc_wire, env, flags, expect_timestamp, expect_quality, ): await pods('pod-a', **env) await publish_command(command(**flags)) await wait_status(postgres_engine, 'ok') stored = await target.read('Setpoint') assert stored.Value.Value == 1.25 if expect_timestamp: assert stored.SourceTimestamp == datetime.fromisoformat(TIMESTAMP) else: assert stored.SourceTimestamp is None [request] = opc_wire mask = datavalue_mask(request, 'ns=2;s=Setpoint') assert bool(mask & SOURCE_TIMESTAMP_BIT) is expect_timestamp assert bool(mask & STATUS_CODE_BIT) is expect_quality # 3.2 write_status (C3) ----------------------------------------------------------------- async def test_scenario_3_2_1_pending_row_becomes_ok( target, pods, publish_command, postgres_engine ): await pods('pod-a') await publish_command(command(value=3.5)) await wait_status(postgres_engine, 'ok') assert (await target.read('Setpoint')).Value.Value == 3.5 async def test_scenario_3_2_2_null_write_status_becomes_ok( opc_servers, mongo_db, pods, publish_command, postgres_engine ): server = await opc_servers('plant-a', {'Setpoint': 0.0}) mongo_db['connectors'].insert_one( load_scenario_input('connector_plant_a.json', {'plant-a': server}) ) insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, None) # row from before the column await pods('pod-a') await publish_command(command()) await wait_status(postgres_engine, 'ok') @pytest.mark.parametrize('worse', ['error', 'session_bad']) async def test_scenario_3_2_3_worse_outcome_is_kept( opc_servers, mongo_db, pods, publish_command, postgres_engine, worse ): server = await opc_servers('plant-a', {'Setpoint': 0.0, 'Marker': 0.0}) mongo_db['connectors'].insert_one( load_scenario_input('connector_plant_a.json', {'plant-a': server}) ) insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, worse) await pods('pod-a') await publish_command(command(value=4.0)) await processed(publish_command, postgres_engine) assert (await server.read('Setpoint')).Value.Value == 4.0 # the OPC write happened assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == worse async def test_scenario_3_2_4_failure_replaces_ok(target, pods, publish_command, postgres_engine): await pods('pod-a') await publish_command(command()) await wait_status(postgres_engine, 'ok') await publish_command(command(value='not-a-number')) await wait_status(postgres_engine, 'error') async def test_scenario_3_2_5_missing_row_is_not_created( target, pods, publish_command, postgres_engine ): await pods('pod-a') await publish_command(command(model_id='3', value=6.0)) await processed(publish_command, postgres_engine) assert (await target.read('Setpoint')).Value.Value == 6.0 assert write_status(postgres_engine, 3, TIMESTAMP) == '' assert prediction_count(postgres_engine) == 2 # model 2 + marker # 3.3 Failed commands ------------------------------------------------------------------- async def test_scenario_3_3_1_unknown_server_is_an_error( target, pods, publish_command, postgres_engine ): await pods('pod-a') await publish_command(command(server_id='404')) await wait_status(postgres_engine, 'error') assert target.writes == [] @pytest.mark.parametrize( 'connector_change', [ pytest.param({'status': 'Inactive'}, id='inactive'), pytest.param({'type': 'modbus'}, id='not_opc_ua'), pytest.param({'security': {'mode': 'None'}}, id='no_policy'), pytest.param({'url': ''}, id='no_url'), ], ) async def test_scenario_3_3_2_unusable_connector_is_an_error( target, mongo_db, pods, publish_command, postgres_engine, connector_change ): mongo_db['connectors'].update_one({'connector_id': '1'}, {'$set': connector_change}) await pods('pod-a') await publish_command(command()) await wait_status(postgres_engine, 'error') assert target.writes == [] @pytest.mark.parametrize( 'overrides', [ pytest.param( {'value': 'abc', 'data_type': 'int', 'addr': 'ns=2;s=Counter'}, id='conversion' ), pytest.param({'data_type': 'decimal128'}, id='unknown_data_type'), pytest.param({'addr': None}, id='missing_addr'), ], ) async def test_scenario_3_3_3_invalid_command_is_an_error_without_opc_write( target, pods, publish_command, postgres_engine, overrides ): await pods('pod-a') cmd = command(**overrides) if overrides.get('addr', '') is None: del cmd['addr'] await publish_command(cmd) await wait_status(postgres_engine, 'error') assert target.writes == [] async def test_scenario_3_3_4_bad_timestamp_skips_the_write_and_keeps_the_row( target, pods, publish_command, postgres_engine ): await pods('pod-a') await publish_command(command(timestamp='13/10/2026 12:00')) await processed(publish_command, postgres_engine) assert [w.NodeId.Identifier for w in target.writes] == ['Marker'] assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == 'pending' @pytest.mark.parametrize( 'addr', [ pytest.param('ns=2;s=Locked', id='not_writable'), pytest.param('ns=2;s=Missing', id='unknown_node'), ], ) async def test_scenario_3_3_5_opc_rejection_is_an_error( target, pods, publish_command, postgres_engine, addr ): await pods('pod-a') await publish_command(command(addr=addr)) await wait_status(postgres_engine, 'error') assert (await target.read('Locked')).Value.Value == 5.0 async def test_scenario_3_3_6_malformed_json_is_dropped_and_the_consumer_keeps_going( target, pods, publish_command, postgres_engine ): await pods('pod-a') await publish_command(b'{"server_id": "1", "value": ') await publish_command(command(value=7.0)) await wait_status(postgres_engine, 'ok') assert (await target.read('Setpoint')).Value.Value == 7.0 # 3.4 Connections ----------------------------------------------------------------------- async def test_scenario_3_4_1_write_reuses_the_read_connection( opc_servers, pods, put_slot, publish_command, postgres_engine, mongo_db ): server = await opc_servers('plant-a', {'Temperature': 20.0, 'Setpoint': 0.0}) put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server})) insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending') pod = await pods('pod-a') await pod.tick() assert mongo_db['connectors'].count_documents({}) == 0 # only the slot knows plant-a await publish_command(command()) await wait_status(postgres_engine, 'ok') assert server.session_count() == 1 async def test_scenario_3_4_2_write_only_server_gets_its_own_connection( target, pods, publish_command, postgres_engine ): pod = await pods('pod-a') assert target.session_count() == 0 await publish_command(command()) await wait_status(postgres_engine, 'ok') assert target.session_count() == 1 assert pod.manager.pool.config_for('1')['url'] == target.url # 3.5 Throughput and scaling ------------------------------------------------------------ async def test_scenario_3_5_1_min_pause_between_writes( target, pods, publish_command, postgres_engine ): await pods('pod-a', WRITE_MIN_PAUSE_S='0.5') for value in (1.0, 2.0, 3.0): await publish_command(command(value=value)) await wait_until(lambda: len(target.writes) == 3) gaps = [b - a for a, b in zip(target.write_times, target.write_times[1:], strict=False)] assert all(gap >= 0.45 for gap in gaps), gaps async def test_scenario_3_5_2_queue_group_writes_each_command_once( target, pods, publish_command, postgres_engine ): await pods('pod-a') await pods('pod-b') for value in range(6): await publish_command(command(value=float(value))) await wait_until(lambda: len(target.writes) == 6) assert await stays_false(lambda: len(target.writes) > 6) # 3.6 NATS availability ----------------------------------------------------------------- async def test_scenario_3_6_1_nats_down_at_startup_connects_later( target, pods, postgres_engine, mongo_db ): port = free_port() url = f'nats://127.0.0.1:{port}' pod = await pods('pod-a', NATS_URL=url) # nothing listens yet await pod.tick() # retried by idle_sweep, notified once per outage assert len(notifications(mongo_db, 'GATEWAY_NATS_CONSUMER_START_FAILED')) == 1 nats_server = nats_server_container(host_port=port) nats_server.start() try: await pod.tick() assert pod.runtime._nats_started client = await nats.connect(url) cmd = load_scenario_input('write_command.json') await client.publish(WRITE_SUBJECT, json.dumps(cmd).encode()) await client.flush() await wait_status(postgres_engine, 'ok') await client.drain() finally: await pod.stop() nats_server.stop()