From c91971307567b947f5fad2801664c075ae45f895 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 May 2026 16:40:56 -0300 Subject: [PATCH] SIENTIAPDE-1646 SIENTIAPDE-1646 Enhance OPC integration and E2E testing framework - Updated .gitignore to exclude '.cursor/*' for better file management. - Added new marker in pyproject.toml for E2E tests using OPC. - Introduced async OPC server fixtures and improved connection handling in conftest.py. - Enhanced error handling in OPC activities and added metrics for session management in metrics.py. - Updated E2E tests to cover new OPC scenarios, including session errors and reconnect handling. - Refactored helper functions to improve prediction assertion logic in helpers.py. - Documented new OPC scenarios in scenarios.md for clarity on expected outcomes. --- e2e/conftest.py | 140 +++++++++++++++++++++++++++++++++++++++++++++++ e2e/scenarios.md | 45 +++++++++++++++ 2 files changed, 185 insertions(+) diff --git a/e2e/conftest.py b/e2e/conftest.py index 123cf4a..d54bb69 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -1,5 +1,7 @@ """Pytest configuration and fixtures for E2E tests.""" +import asyncio +import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path from unittest.mock import MagicMock @@ -14,6 +16,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.drift import Drift from laborious.workflows.minimal_retrain import MinimalRetrain @@ -461,3 +464,140 @@ async def temporal_worker_minimal_retrain(temporal_test_env, test_activities): yield worker +def _connect_activities_to_opc_server(activities: Activities, server_url: str) -> None: + """ + Initialize OPC repositories and block until the E2E server session is ready. + + Runs synchronously (typically via ``asyncio.to_thread``) so the asyncua test + server event loop is not blocked during ``Client.connect()``. + + Args: + activities (Activities): Worker activities under test. + server_url (str): ``opc.tcp://`` URL from ``OpcE2ETestServer``. + """ + activities.init_opc() + repo = activities.opc_repository['1'] + deadline = time.monotonic() + 30.0 + while time.monotonic() < deadline: + if repo._session_ready.is_set(): + return + connected, _ = repo.connect() + if connected: + return + time.sleep(0.5) + raise RuntimeError(f'Could not connect OpcRepository to OPC E2E server at {server_url}') + + +def _build_e2e_opc_config(server_url: str) -> dict[str, dict]: + """ + OPC server config for E2E Activities pointing at an in-process asyncua server. + + Args: + server_url (str): ``opc.tcp://`` endpoint from ``OpcE2ETestServer``. + + Return: + dict: ``opc_config`` payload for ``Activities`` (server id ``1``). + """ + return { + '1': { + 'id': '1', + 'server_name': 'e2e_opcua', + 'url': server_url, + 'server_uri': server_url, + 'cert_path': None, + 'private_key_path': None, + 'server_cert_path': None, + 'reconnection_interval': 0, + } + } + + +@pytest_asyncio.fixture +async def opc_e2e_server(): + """In-process asyncua server with writable prediction/confidence nodes.""" + server = OpcE2ETestServer() + await server.start() + await asyncio.sleep(0.5) + try: + yield server + finally: + await server.stop() + + +@pytest_asyncio.fixture(scope='function') +async def test_activities_real_opc( + postgres_container, + minio_container, + mock_logger, + notification_handler, + metrics_controller, + mlflow_repository_stub, + plugin_store_stub, + pi_web_api_client_stub, + opc_e2e_server: OpcE2ETestServer, +): + """Activities with real OpcRepository connected to the in-process OPC UA server.""" + minio_client = minio_container.get_client() + if not minio_client.bucket_exists('test-bucket'): + minio_client.make_bucket('test-bucket') + minio_port = minio_container.get_exposed_port(9000) + + activities = Activities( + postgres_config={ + 'host': 'localhost', + 'port': int(postgres_container.get_exposed_port(5432)), + 'user': postgres_container.username, + 'password': postgres_container.password, + 'dbname': postgres_container.dbname, + 'min_connections': 1, + 'max_connections': 5, + }, + plugin_store=plugin_store_stub, + minio_config={ + 'endpoint_url': f'localhost:{minio_port}', + 'access_key': 'minioadmin', + 'secret_key': 'minioadmin', + 'default_bucket': 'test-bucket', + 'retention_hours': 24, + 'secure': False, + }, + opc_config=_build_e2e_opc_config(opc_e2e_server.url), + pi_web_api_config={ + 'base_url': 'http://localhost:8080', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + }, + logger=mock_logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + mlflow_repository=mlflow_repository_stub, + ) + activities.pi_web_api_client = pi_web_api_client_stub + await asyncio.to_thread(_connect_activities_to_opc_server, activities, opc_e2e_server.url) + try: + yield activities + finally: + await asyncio.to_thread(_teardown_real_opc_activities, activities) + + +def _teardown_real_opc_activities(activities: Activities) -> None: + """Disconnect OPC sessions and shut down activities (sync, for asyncio.to_thread).""" + for opc_repo in activities.opc_repository.values(): + opc_repo.disconnect() + activities.shutdown() + + +@pytest_asyncio.fixture(scope='function') +async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc): + """Temporal worker using real OpcRepository against the in-process OPC UA server.""" + with ThreadPoolExecutor(max_workers=32) as activity_executor: + async with Worker( + temporal_test_env.client, + task_queue='test-queue', + workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], + activities=_worker_activity_list(test_activities_real_opc), + activity_executor=activity_executor, + ) as worker: + yield worker + + diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 5fd681b..b0fee2c 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -10,6 +10,27 @@ It is a functional reference of scenario behavior, inputs, and expected outcomes - Tests run under `e2e/` and are marked with `@pytest.mark.integration`. - PostgreSQL and MinIO are provisioned with testcontainers. - `test_minio_offload.py` uses real MinIO I/O; other scenario suites may use stubs/mocks for optional outputs. +- Real OPC UA scenarios use `@pytest.mark.opc` and an in-process asyncua server (`e2e/test_opc_real_server.py`). + +### Local validation + +Use the existing project virtualenv and the shared `validate` script for unit/quality gates; run E2E separately (Docker required). + +```bash +source ./venv/bin/activate + +# Auto-fix + static checks (no pytest) +validate --fix --project-name=laborious + +# Full unit + quality gate +validate --project-name=laborious + +# E2E (integration) +pytest e2e/ --override-ini testpaths=e2e -m integration + +# E2E (real OPC server only) +pytest e2e/test_opc_real_server.py --override-ini testpaths=e2e -m opc +``` --- @@ -187,6 +208,30 @@ Source: `e2e/test_predictions_batch_format_export.py` - Workflow completes. - Prediction persisted with PI error confidence and descriptive comment. +#### 3.2.4 OPC session / channel error (confidence 14) +**Summary**: Tier-1 `BadSessionIdInvalid` (or equivalent session error) degrades the prediction without failing the workflow. + +**Sources**: +- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_4_opc_session_bad_mock` +- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_4_opc_session_bad_real_server` (`@pytest.mark.opc`) + +**Expected Outcome**: +- Workflow completes. +- `prediction_confidence` is 14. +- Comments contain `OPC UA session/channel error: BadSessionIdInvalid`. + +#### 3.2.5 OPC write blocked during reconnect (confidence 14) +**Summary**: While reconnect holds the repository connection lock, writes fail fast with `reconnect_in_progress`. + +**Sources**: +- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_5_opc_reconnect_in_progress_mock` +- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server` (`@pytest.mark.opc`) + +**Expected Outcome**: +- Workflow completes. +- `prediction_confidence` is 14. +- Comments contain `OPC UA reconnect in progress`. + ### 3.3.1 Combined Optional Outputs (PI + OPC) **Summary**: Both external output channels are enabled together.