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

@@ -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
+
+