diff --git a/docs/opc-communication.md b/docs/opc-communication.md index ce612b9..3084d05 100644 --- a/docs/opc-communication.md +++ b/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) diff --git a/e2e/conftest.py b/e2e/conftest.py index f29ec0d..cc49d9b 100644 --- a/e2e/conftest.py +++ b/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 diff --git a/e2e/opc_test_server.py b/e2e/opc_test_server.py new file mode 100644 index 0000000..69b02c8 --- /dev/null +++ b/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} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 03f58c2..8841424 100644 --- a/e2e/scenarios.md +++ b/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 diff --git a/e2e/test_opc_real_server.py b/e2e/test_opc_real_server.py new file mode 100644 index 0000000..200a8a4 --- /dev/null +++ b/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, + ) diff --git a/pyproject.toml b/pyproject.toml index 5123123..a878882 100644 --- a/pyproject.toml +++ b/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]