SIENTIAPDE-1811

Enhance OPC UA testing framework and documentation

- Added a new marker in pyproject.toml for tests using the in-process OPC UA server.
- Updated opc-communication.md to clarify E2E test scenarios involving the real OPC server and mock server.
- Introduced an in-process asyncua OPC UA server fixture in conftest.py for E2E tests.
- Created a new fixture for activities using the real OpcRepository connected to the in-process server.
- Updated scenarios.md to include instructions for running OPC real-server tests.

docs/opc-communication.md

@@ -157,4 +157,6 @@ Legacy activity metrics: `laborious_prediction_opc_writing_count`, `laborious_pr
 
 - Unit: [`tests/laborious/utils/repository/test_opc_repository.py`](../tests/laborious/utils/repository/test_opc_repository.py)
 - Unit: [`tests/laborious/activities/test_opc.py`](../tests/laborious/activities/test_opc.py)
-- E2E: [`e2e/test_predictions_batch_format_export.py`](../e2e/test_predictions_batch_format_export.py), scenarios in [`e2e/scenarios.md`](../e2e/scenarios.md)
+- E2E (mock OPC): [`e2e/test_predictions_batch_format_export.py`](../e2e/test_predictions_batch_format_export.py)
+- E2E (in-process asyncua server + real `OpcRepository`): [`e2e/test_opc_real_server.py`](../e2e/test_opc_real_server.py) — scenarios 3.1.2, 3.2.2, 3.2.4, 3.2.5
+- Scenarios: [`e2e/scenarios.md`](../e2e/scenarios.md)

e2e/conftest.py

@@ -21,6 +21,7 @@ from testcontainers.postgres import PostgresContainer
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
+from e2e.opc_test_server import OpcE2ETestServer
 from laborious.activities.activities import Activities
 from laborious.workflows.predictions_batch import PredictionsBatch
 from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
@@ -304,6 +305,19 @@ def mock_opc_repository():
     mock_repo.disconnect = AsyncMock()
     return mock_repo
 
+
+@pytest_asyncio.fixture
+async def opc_e2e_server():
+    """
+    In-process asyncua OPC UA server for E2E tests against OpcRepository.
+    """
+    server = OpcE2ETestServer()
+    await server.start()
+    try:
+        yield server
+    finally:
+        await server.stop()
+
 @pytest_asyncio.fixture
 def patch_create_engine(postgres_engine):
     """Patch create_engine to return test postgres_engine."""
@@ -584,3 +598,84 @@ async def temporal_worker_real_minio(temporal_test_env, test_activities_real_min
         activities=_worker_activity_list(test_activities_real_minio),
     ) as worker:
         yield worker
+
+
+@pytest_asyncio.fixture(scope='function')
+async def test_activities_real_opc(
+    postgres_engine,
+    postgres_container,
+    opc_e2e_server: OpcE2ETestServer,
+    mock_logger,
+    notification_handler,
+    metrics_controller,
+    patch_create_engine,
+    patch_minio_repository,
+    patch_mlflow,
+    patch_pi_web_api_repository,
+):
+    """
+    Activities with a real OpcRepository connected to the in-process OPC UA server.
+    """
+    activities = Activities(
+        postgres_config={
+            'host': 'localhost',
+            'port': postgres_container.get_exposed_port(5432),
+            'user': 'test',
+            'password': 'test',
+            'dbname': 'test',
+            'min_connections': 1,
+            'max_connections': 5,
+        },
+        mlflow_config={
+            'host': 'http://localhost',
+            'port': '5000',
+            'username': 'test',
+            'password': 'test',
+        },
+        minio_config={
+            'endpoint_url': 'localhost:9000',
+            'access_key': 'test',
+            'secret_key': 'test',
+            'default_bucket': 'test-bucket',
+            'retention_hours': 24,
+            'secure': False,
+        },
+        opc_config={
+            '1': {
+                'id': '1',
+                'server_name': 'e2e-opc',
+                'url': opc_e2e_server.url,
+                'server_uri': opc_e2e_server.url,
+                'cert_path': None,
+                'private_key_path': None,
+                'server_cert_path': None,
+                'reconnection_interval': 0,
+            }
+        },
+        pi_web_api_config={
+            'base_url': 'http://localhost:8080',
+            'auth_type': 'bearer',
+            'auth_token': 'test_token',
+        },
+        logger=mock_logger,
+        notification_handler=notification_handler,
+    )
+    await activities.init_opc()
+    repo = activities.opc_repository['1']
+    assert repo._session_ready.is_set(), 'OPC E2E server connection failed during init_opc'
+    try:
+        yield activities
+    finally:
+        await activities.shutdown()
+
+
+@pytest_asyncio.fixture(scope='function')
+async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc):
+    """Temporal worker backed by Activities using the in-process OPC UA server."""
+    async with Worker(
+        temporal_test_env.client,
+        task_queue='test-queue',
+        workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction],
+        activities=_worker_activity_list(test_activities_real_opc),
+    ) as worker:
+        yield worker

e2e/opc_test_server.py

@@ -0,0 +1,189 @@
+"""
+In-process OPC UA server for E2E tests (asyncua).
+
+Provides writable prediction/confidence nodes and optional write faults
+(Tier-1 BadSessionIdInvalid via PreWrite callback).
+"""
+
+from __future__ import annotations
+
+import socket
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from asyncua import Server, ua
+from asyncua.common.callback import CallbackType
+from asyncua.common.utils import ServiceError
+
+if TYPE_CHECKING:
+    from asyncua.common.node import Node
+
+
+UNKNOWN_NODE_ID = 'ns=99;i=9999'
+
+
+@dataclass(frozen=True)
+class OpcE2ENodeIds:
+    """NodeId strings used in opc_output_config for E2E workflows."""
+
+    prediction: str
+    confidence: str
+    unknown: str = UNKNOWN_NODE_ID
+
+
+class OpcE2ETestServer:
+    """
+    Ephemeral asyncua server with Laborious E2E variables and controllable faults.
+
+    Args:
+        host: Bind address (default 127.0.0.1).
+    """
+
+    def __init__(self, host: str = '127.0.0.1') -> None:
+        self._host = host
+        self._server: Server | None = None
+        self._prediction_node: Node | None = None
+        self._confidence_node: Node | None = None
+        self._session_bad_on_write = False
+        self._url: str | None = None
+        self._node_ids: OpcE2ENodeIds | None = None
+
+    @property
+    def url(self) -> str:
+        if self._url is None:
+            raise RuntimeError('OPC E2E server is not started')
+        return self._url
+
+    @property
+    def node_ids(self) -> OpcE2ENodeIds:
+        if self._node_ids is None:
+            raise RuntimeError('OPC E2E server is not started')
+        return self._node_ids
+
+    def set_session_bad_on_write(self, enabled: bool) -> None:
+        """
+        When enabled, every client Write is rejected with BadSessionIdInvalid.
+
+        Args:
+            enabled (bool): Turn Tier-1 session fault injection on or off.
+        """
+        self._session_bad_on_write = enabled
+
+    async def start(self) -> OpcE2ENodeIds:
+        """
+        Start the OPC UA server on a free TCP port.
+
+        Return:
+            OpcE2ENodeIds: NodeId strings for prediction and confidence tags.
+        """
+        port = _free_port(self._host)
+        self._url = f'opc.tcp://{self._host}:{port}/freeopcua/server/'
+
+        server = Server()
+        server.set_endpoint(self._url)
+        await server.init()
+        server.iserver.callback_service.addListener(
+            CallbackType.PreWrite,
+            self._pre_write_callback,
+        )
+
+        idx = await server.register_namespace('http://sientia.test/laborious-e2e')
+        e2e_object = await server.nodes.objects.add_object(idx, 'LaboriousE2E')
+        prediction = await e2e_object.add_variable(
+            idx,
+            'Prediction',
+            ua.Variant(0.0, ua.VariantType.Float),
+        )
+        confidence = await e2e_object.add_variable(
+            idx,
+            'Confidence',
+            ua.Variant(0.0, ua.VariantType.Float),
+        )
+        await prediction.set_writable()
+        await confidence.set_writable()
+
+        await server.start()
+        self._server = server
+        self._prediction_node = prediction
+        self._confidence_node = confidence
+        self._node_ids = OpcE2ENodeIds(
+            prediction=prediction.nodeid.to_string(),
+            confidence=confidence.nodeid.to_string(),
+        )
+        return self._node_ids
+
+    async def stop(self) -> None:
+        """Stop the OPC UA server and release the listening port."""
+        if self._server is not None:
+            await self._server.stop()
+            self._server = None
+        self._prediction_node = None
+        self._confidence_node = None
+        self._url = None
+        self._node_ids = None
+        self._session_bad_on_write = False
+
+    async def read_prediction(self) -> float:
+        """
+        Read the current prediction variable value from the address space.
+
+        Return:
+            float: Stored prediction value.
+        """
+        if self._prediction_node is None:
+            raise RuntimeError('OPC E2E server is not started')
+        value = await self._prediction_node.read_value()
+        return float(value)
+
+    async def read_confidence(self) -> float:
+        """
+        Read the current confidence variable value from the address space.
+
+        Return:
+            float: Stored confidence value.
+        """
+        if self._confidence_node is None:
+            raise RuntimeError('OPC E2E server is not started')
+        value = await self._confidence_node.read_value()
+        return float(value)
+
+    async def _pre_write_callback(self, _event, _service) -> None:
+        if self._session_bad_on_write:
+            raise ServiceError(ua.StatusCodes.BadSessionIdInvalid)
+
+
+def _free_port(host: str) -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind((host, 0))
+        return int(sock.getsockname()[1])
+
+
+def build_opc_output_config(
+    node_ids: OpcE2ENodeIds,
+    *,
+    prediction_tag: str | None = None,
+    confidence_tag: str | None = None,
+    prediction_only: bool = False,
+    server_key: str = '1',
+) -> dict[str, dict]:
+    """
+    Build opc_output_config for PredictionsBatch using real server NodeIds.
+
+    Args:
+        node_ids (OpcE2ENodeIds): Node ids from OpcE2ETestServer.
+        prediction_tag (str | None): Override prediction NodeId (default: node_ids.prediction).
+        confidence_tag (str | None): Override confidence NodeId (default: node_ids.confidence).
+        prediction_only (bool): When True, omit confidence_tags (single write per activity).
+        server_key (str): OPC server id key in opc_output_config.
+
+    Return:
+        dict: opc_output_config payload for workflow input.
+    """
+    pred = prediction_tag if prediction_tag is not None else node_ids.prediction
+    conf = confidence_tag if confidence_tag is not None else node_ids.confidence
+    server_config: dict = {
+        'prediction_tags': {pred: {'data_type': 'float'}},
+    }
+    if not prediction_only:
+        server_config['confidence_tags'] = {conf: {'data_type': 'float'}}
+    return {server_key: server_config}

e2e/scenarios.md

@@ -8,6 +8,8 @@ This document describes all possible test scenarios for the `predictions_batch`
 - **Dependencies**: install dev requirements (includes `testcontainers[postgres,minio]`).
 - **Invocation**: run only integration-marked tests, for example: `pytest e2e/ -m integration`.
 - **MinIO tests**: `e2e/test_minio_offload.py` exercises real S3 uploads; other E2E modules continue to mock MinIO on the worker used by most scenarios.
+- **OPC tests (real server)**: `e2e/test_opc_real_server.py` uses an in-process **asyncua** server and real `OpcRepository` (`test_activities_real_opc`). Scenarios 3.1.2, 3.2.2, 3.2.4, and 3.2.5 are covered there. Other E2E modules keep the OPC mock.
+- Run only OPC real-server tests: `pytest e2e/test_opc_real_server.py -m "integration and opc"`.
 
 ## Workflow Overview
 

e2e/test_opc_real_server.py

@@ -0,0 +1,194 @@
+"""
+E2E tests for OPC export using an in-process asyncua server and real OpcRepository.
+
+Covers scenarios 3.1.2, 3.2.2, 3.2.4, and 3.2.5 from e2e/scenarios.md.
+Mock-based OPC tests remain in test_predictions_batch_format_export.py.
+"""
+
+import asyncio
+
+import pytest
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from e2e.helpers import assert_prediction, insert_sample_data, make_workflow_id, start_and_await_workflow
+from e2e.opc_test_server import UNKNOWN_NODE_ID, OpcE2ETestServer, build_opc_output_config
+from e2e.test_predictions_batch_format_export import get_base_input_data
+from laborious.activities.activities import Activities
+from laborious.activities.opc import OPC_RECONNECT_IN_PROGRESS_COMMENT
+from laborious.utils.repository.opc_repository import OpcRepository
+from laborious.workflows.predictions_batch import PredictionsBatch
+
+
+async def _slow_reconnect_under_lock(repo: OpcRepository, hold_seconds: float = 0.75) -> None:
+    """
+    Hold the connection lock briefly so concurrent writes see reconnect_in_progress.
+
+    Args:
+        repo (OpcRepository): Connected repository.
+        hold_seconds (float): Time to keep the lock before reconnecting.
+    """
+    async with repo._connection_lock:
+        await asyncio.sleep(hold_seconds)
+        await repo._reconnect_locked()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_1_2_export_with_opc_only_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.1.2 (real OPC): connect, write prediction and confidence, verify server values.
+    """
+    client = temporal_test_env.client
+    model_id = 412
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids)
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client,
+        PredictionsBatch.run,
+        input_data,
+        make_workflow_id('test-opc-real-happy'),
+    )
+
+    test_activities_real_opc.pi_web_api_client.write_value.assert_not_called()
+    assert await opc_e2e_server.read_prediction() == pytest.approx(0.5)
+    assert await opc_e2e_server.read_confidence() == pytest.approx(0.0)
+    assert_prediction(postgres_engine, model_id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_2_opc_write_error_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.2 (real OPC): unknown NodeId yields generic write failure (confidence 12).
+    """
+    client = temporal_test_env.client
+    model_id = 422
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    node_ids = opc_e2e_server.node_ids
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(
+        node_ids,
+        prediction_tag=UNKNOWN_NODE_ID,
+        confidence_tag=UNKNOWN_NODE_ID,
+    )
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client,
+        PredictionsBatch.run,
+        input_data,
+        make_workflow_id('test-opc-real-bad-node'),
+    )
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=12,
+        comments='Some data could not be written to OPC servers',
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_4_opc_session_bad_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.4 (real OPC): server PreWrite fault injects BadSessionIdInvalid (confidence 14).
+    """
+    client = temporal_test_env.client
+    model_id = 424
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    opc_e2e_server.set_session_bad_on_write(True)
+    try:
+        input_data = get_base_input_data(model_id)
+        input_data['opc_output_config'] = build_opc_output_config(
+            opc_e2e_server.node_ids,
+            prediction_only=True,
+        )
+        input_data['pi_web_api_output_config'] = None
+
+        await start_and_await_workflow(
+            client,
+            PredictionsBatch.run,
+            input_data,
+            make_workflow_id('test-opc-real-session-bad'),
+        )
+    finally:
+        opc_e2e_server.set_session_bad_on_write(False)
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains='OPC UA session/channel error: BadSessionIdInvalid',
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.5 (real OPC): writes rejected while reconnect holds the connection lock.
+    """
+    client = temporal_test_env.client
+    model_id = 425
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    repo = test_activities_real_opc.opc_repository['1']
+    repo._session_ready.clear()
+    reconnect_task = asyncio.create_task(_slow_reconnect_under_lock(repo))
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids)
+    input_data['pi_web_api_output_config'] = None
+
+    try:
+        await start_and_await_workflow(
+            client,
+            PredictionsBatch.run,
+            input_data,
+            make_workflow_id('test-opc-real-reconnect-block'),
+        )
+    finally:
+        await reconnect_task
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains=OPC_RECONNECT_IN_PROGRESS_COMMENT,
+    )

pyproject.toml

@@ -121,6 +121,7 @@ markers = [
     "asyncio: marks tests as async",
     "integration: marks tests as integration tests",
     "unit: marks tests as unit tests",
+    "opc: marks tests that use the in-process OPC UA server (OpcRepository E2E)",
 ]
 
 [tool.coverage.run]