diff --git a/.gitignore b/.gitignore index b5a95e7..c27efb2 100644 --- a/.gitignore +++ b/.gitignore @@ -54,4 +54,5 @@ catboost_info/ mlruns/ relatorio* -openspec/* \ No newline at end of file +openspec/* +.cursor/* \ No newline at end of file diff --git a/docs/opc-communication.md b/docs/opc-communication.md new file mode 100644 index 0000000..2c51c75 --- /dev/null +++ b/docs/opc-communication.md @@ -0,0 +1,162 @@ +# OPC UA communication (Laborious) + +Laborious exports predictions to OPC UA servers through `OpcRepository` ([`laborious/utils/repository/opc_repository.py`](../laborious/utils/repository/opc_repository.py)) and the synchronous Temporal activity layer in [`laborious/activities/opc.py`](../laborious/activities/opc.py). The repository uses `asyncua.sync.Client` (asyncio on a background thread) so activities remain blocking without `async def`. + +Implementation plan for session/channel recovery on Tier-1 `Bad*` errors: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](../.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). + +## Architecture + +```text +Worker (long-lived) + └── OpcRepository per OPC server id (from OPC_CONFIG / env) + ├── connect / disconnect / validate_connection (read-only) + ├── _connect_locked / _reconnect_locked (under _connection_lock) + ├── write_data (single attempt per call) + └── background reconnect on Tier-1 Bad* + +Temporal activity write_opc_data + └── OPC.manage_output_tags → write_data per tag (sequential per activity) +``` + +One worker process holds one `OpcRepository` instance per configured server. Multiple Temporal activities can call `write_data` concurrently on the same repository. + +## Connection lifecycle + +| Phase | Behavior | +|-------|----------| +| Startup | `init_opc()` creates repositories and calls `connect()` → `_connect_locked()` | +| Steady state | `validate_connection()` is read-only (`protocol.state` only); `_session_ready` is checked in `write_data` | +| Tier-1 Bad* | `_start_reconnect_on_bad` → `_run_reconnect_on_bad` → `_reconnect_locked()` (respects `reconnection_interval`) | +| Write | `write_data()` checks `_session_ready`, validates, then one `get_node` + `write_value` | +| Shutdown | `close()` disconnects all repositories | + +### Session and channel timeouts + +Requested session and secure-channel lifetime: **10 minutes** (`OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS` in `opc_repository.py`). The server may revise these values; negotiated values are logged after connect and exposed as `opc_session_revised_timeout_milliseconds`. + +### Reconnection interval + +`OPC_RECONNECTION_INTERVAL` is in **seconds** (default `120`). It gates **background** reconnect after Tier-1 `Bad*` (`last_reconnection_time` is updated only in `_reconnect_locked()`). It limits load on the OPC server when many workflows fail at once. + +## Concurrency: connection lock and session readiness + +To allow **multiple concurrent writes** when the session is healthy, but **block all writes** while the connection is being torn down or re-established: + +| Primitive | Role | +|-----------|------| +| `_connection_lock` (`asyncio.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. | +| `_session_ready` (`asyncio.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. | + +**Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):** + +| Method | Role | +|--------|------| +| `_create_client()` | Create asyncua `Client` + optional `set_security`; raises if `client` already exists | +| `_open_session()` | `client.connect()` + metrics; raises if session already open or client missing | +| `_connect_locked()` | `_create_client()` (when needed) + `_open_session()`; raises if already connected | +| `_disconnect_locked()` | Teardown session and clear `client` | +| `_reconnect_locked()` | `_disconnect_locked()` + `_connect_locked()`; sets `last_reconnection_time` | + +Public `connect()` / `disconnect()` acquire the lock and call `_connect_locked()` / `_disconnect_locked()`. + +**Write path (`write_data`):** + +1. If `_session_ready` is cleared → **fail immediately** (`opc_error_kind=reconnect_in_progress`). +2. `validate_connection()` checks `protocol.state` only (read-only). +3. Single `get_node` + `write_value` (no retry). + +**Reconnect path (`_run_reconnect_on_bad`):** + +1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows. +2. `async with _connection_lock:` → `_reconnect_locked()`. +3. `_session_ready` is set on successful `_open_session()`. + +A second `_connect_locked()` while a session is already open raises `OpcSessionAlreadyConnectedError` (disconnect first). + +**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes with an optional `asyncio.Semaphore(1)` while keeping the connection lock semantics above. + +**Future threads:** replace `asyncio.Lock` / `Event` with `threading` primitives or route all OPC I/O through one dedicated loop. + +## Tier-1 `Bad*` errors and reconnect + +When the server invalidates the session (e.g. `BadSessionIdInvalid`) but the client still sees transport as open, `write_data` fails once, records the OPC status in metrics, and **schedules** reconnect if: + +- The exception is a `UaStatusCodeError` whose name is in `RECONNECTABLE_OPC_BAD_NAMES` (see plan), and +- `reconnection_interval` has elapsed since `last_reconnection_time`, and +- No reconnect task is already running. + +There is **no write retry**: the failed export is not sent again in the same activity. + +## Prediction confidence and PostgreSQL comments + +| `prediction_confidence` | Meaning | +|-------------------------|---------| +| (unchanged) | Successful OPC export | +| **12** | Generic OPC write failure (`OPC_WRITTING_ERROR_CONFIDENCE`) | +| **14** | Tier-1 session/channel `Bad*` on export (`OPC_SESSION_BAD_CONFIDENCE`) | +| **14** | Write while reconnect in progress (`OPC_SESSION_BAD_CONFIDENCE`, comment `OPC UA reconnect in progress`) | +| **13** | PI Web API write failure (separate path) | + +Session/channel errors use a stable comment for counting: + +```text +OPC UA session/channel error: BadSessionIdInvalid +``` + +Reconnect-in-progress exports use: + +```text +OPC UA reconnect in progress +``` + +Example SQL: + +```sql +SELECT count(*) FROM predictions WHERE prediction_confidence = 14; +SELECT count(*) FROM predictions WHERE comments LIKE 'OPC UA session/channel error:%'; +``` + +## Prometheus metrics (`opc_*`) + +Defined in [`laborious/metrics.py`](../laborious/metrics.py). Do not rename in production without a dashboard migration. + +| Metric | Purpose | +|--------|---------| +| `opc_connections_initiated_total` | Connection attempts | +| `opc_connections_failed_total` | Failed connects | +| `opc_connection_status` | Gauge 1=connected, 0=disconnected | +| `opc_session_created_total` | Session established after connect | +| `opc_session_closed_total` | Disconnect initiated | +| `opc_session_revised_timeout_milliseconds` | Negotiated session timeout (ms) | +| `opc_write_attempts_total` | Per write; label `result` = `OK` or exception name | +| `opc_write_inter_arrival_over_session_timeout_total` | Successful writes spaced longer than revised session timeout | + +Legacy activity metrics: `laborious_prediction_opc_writing_count`, `laborious_prediction_opc_writing_response_time_monitor`. + +## Environment variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `OPC_CONFIG` | — | JSON map of server configs (overrides single-server env) | +| `OPC_ID` | `1` | Server id | +| `OPC_URL` | `opc.tcp://localhost:4840` | Endpoint | +| `OPC_SERVER_NAME` | `default_server` | Label for metrics/logs | +| `OPC_SERVER_URI` | same as URL | Application URI / cert SAN | +| `OPC_CERT_PATH` | — | Client certificate (secure mode) | +| `OPC_PRIVATE_KEY_PATH` | — | Client private key | +| `OPC_SERVER_CERT_PATH` | — | Server certificate | +| `OPC_RECONNECTION_INTERVAL` | `120` | Minimum seconds between reconnects | + +## Operations checklist + +- Correlate `BadSessionIdInvalid` in `opc_write_attempts_total` with `opc_session_closed_total` / `opc_session_created_total` (reconnect may finish after the row is stored with confidence 14). +- Use confidence **14** and comment prefix for session invalidation rates; use **12** for other OPC failures. +- Respect `OPC_RECONNECTION_INTERVAL` under parallel load; bursts of confidence 14 are expected until the next successful cycle. + +## Related tests + +- 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 (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 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/helpers.py b/e2e/helpers.py index 3ffd1b0..7add050 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -205,6 +205,7 @@ def assert_prediction( prediction_confidence: int | Decimal = 0, prediction_status: str = 'Good', comments: str = '', + comments_contains: str | None = None, ) -> None: """ Assert exactly one prediction row exists for model_id with expected columns. @@ -215,10 +216,9 @@ def assert_prediction( prediction: Expected prediction value. prediction_confidence: Expected confidence (int or Decimal for numeric column). prediction_status: Expected status string. - comments: Expected comments string. + comments: Expected exact comments string (ignored when ``comments_contains`` is set). + comments_contains: When set, assert this substring appears in comments. """ - import pytest - with postgres_engine.connect() as conn: result_query = conn.execute( text( @@ -238,7 +238,13 @@ def assert_prediction( str(prediction_confidence) ), f'Expected prediction_confidence={prediction_confidence}, got {row[2]}' assert row[3] == prediction_status, f"Expected prediction_status='{prediction_status}', got {row[3]}" - assert row[4] == comments, f"Expected comments='{comments}', got {row[4]}" + actual_comments = row[4] or '' + if comments_contains is not None: + assert comments_contains in actual_comments, ( + f"Expected comments to contain '{comments_contains}', got '{actual_comments}'" + ) + else: + assert actual_comments == comments, f"Expected comments='{comments}', got '{actual_comments}'" def assert_continue( 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 1d02303..d4b96ba 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 +``` --- @@ -195,6 +216,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. diff --git a/e2e/test_opc_real_server.py b/e2e/test_opc_real_server.py new file mode 100644 index 0000000..ab0e59e --- /dev/null +++ b/e2e/test_opc_real_server.py @@ -0,0 +1,201 @@ +""" +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 threading +import time + +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 + + +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. + """ + with repo._connection_lock: + time.sleep(hold_seconds) + 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_thread = threading.Thread( + target=_slow_reconnect_under_lock, + args=(repo,), + daemon=True, + ) + reconnect_thread.start() + + 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: + reconnect_thread.join(timeout=5.0) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments_contains=OPC_RECONNECT_IN_PROGRESS_COMMENT, + ) diff --git a/e2e/test_predictions_batch_format_export.py b/e2e/test_predictions_batch_format_export.py index 822efee..d903756 100644 --- a/e2e/test_predictions_batch_format_export.py +++ b/e2e/test_predictions_batch_format_export.py @@ -4,7 +4,7 @@ End-to-end tests for PredictionsBatch workflow - Format and Export scenarios. from decimal import Decimal from typing import Any, cast -from unittest.mock import ANY, call +from unittest.mock import call import pytest from sientia_do.notifications.models import NotificationLevel @@ -123,7 +123,6 @@ async def test_scenario_3_1_1_default_prediction_export( 'addr_1', 0, 'float', - ANY, { 'model_id': 311, 'model_name': 'test_model', @@ -135,7 +134,6 @@ async def test_scenario_3_1_1_default_prediction_export( 'addr_2', 2, 'float', - ANY, { 'model_id': 311, 'model_name': 'test_model', @@ -222,20 +220,28 @@ async def test_scenario_3_1_2_export_with_opc_only( opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) opc_write_data.assert_has_calls( [ - call('addr_1', 0.5, 'float', ANY, - { - 'model_id': 312, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), - call('addr_2', 0, 'float', ANY, - { - 'model_id': 312, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), + call( + 'addr_1', + 0.5, + 'float', + { + 'model_id': 312, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), + call( + 'addr_2', + 0, + 'float', + { + 'model_id': 312, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), ] ) @@ -470,20 +476,28 @@ async def test_scenario_3_1_5_export_without_transformed_data( opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) opc_write_data.assert_has_calls( [ - call('addr_1', 0.5, 'float', ANY, - { - 'model_id': 315, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), - call('addr_2', 0, 'float', ANY, - { - 'model_id': 315, - 'model_name': 'test_model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'predictions_batch', - }), + call( + 'addr_1', + 0.5, + 'float', + { + 'model_id': 315, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), + call( + 'addr_2', + 0, + 'float', + { + 'model_id': 315, + 'model_name': 'test_model', + 'schedule_name': 'test-schedule', + 'workflow_name': 'predictions_batch', + }, + ), ] ) @@ -616,7 +630,110 @@ async def test_scenario_3_2_2_opc_write_error( prediction_confidence=12, comments='Some data could not be written to OPC servers', ) - + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_scenario_3_2_4_opc_session_bad_mock( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, +): + """ + Scenario 3.2.4 (mock): Tier-1 session error maps to confidence 14. + """ + client = temporal_test_env.client + model_id = 324 + + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) + opc_write_data.return_value = ( + False, + { + 'opc_error_kind': 'session_bad', + 'opc_status': 'BadSessionIdInvalid', + 'notification_id': 'OPC_WRITE_DATA_ERROR_1', + 'message': 'OPC session invalid', + 'block': 'opc_repository', + 'level': NotificationLevel.ERROR, + 'attachment_content': 'BadSessionIdInvalid', + }, + ) + + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = { + '1': { + 'prediction_tags': {'addr_1': {'data_type': 'float'}}, + 'confidence_tags': {'addr_2': {'data_type': 'float'}}, + } + } + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-opc-session-bad-mock') + ) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments_contains='OPC UA session/channel error: BadSessionIdInvalid', + ) + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_scenario_3_2_5_opc_reconnect_in_progress_mock( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, +): + """ + Scenario 3.2.5 (mock): reconnect_in_progress maps to confidence 14. + """ + client = temporal_test_env.client + model_id = 325 + + insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) + + opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data) + opc_write_data.return_value = ( + False, + { + 'opc_error_kind': 'reconnect_in_progress', + 'notification_id': 'OPC_WRITE_DATA_ERROR_1', + 'message': 'OPC reconnect in progress', + 'block': 'opc_repository', + 'level': NotificationLevel.ERROR, + 'attachment_content': 'reconnect', + }, + ) + + input_data = get_base_input_data(model_id) + input_data['opc_output_config'] = { + '1': { + 'prediction_tags': {'addr_1': {'data_type': 'float'}}, + 'confidence_tags': {'addr_2': {'data_type': 'float'}}, + } + } + input_data['pi_web_api_output_config'] = None + + await start_and_await_workflow( + client, + PredictionsBatch.run, + input_data, + make_workflow_id('test-opc-reconnect-mock'), + ) + + assert_prediction( + postgres_engine, + model_id, + prediction_confidence=14, + comments_contains='OPC UA reconnect in progress', + ) @pytest.mark.asyncio diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index eb8b2af..733ecb9 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -15,6 +15,45 @@ with workflow.unsafe.imports_passed_through(): from laborious.utils.repository.opc_repository import OpcRepository OPC_WRITTING_ERROR_CONFIDENCE = 12 +OPC_SESSION_BAD_CONFIDENCE = 14 +OPC_SESSION_BAD_COMMENT_PREFIX = 'OPC UA session/channel error:' +OPC_WRITTING_ERROR_MESSAGE = 'Some data could not be written to OPC servers' +OPC_RECONNECT_IN_PROGRESS_COMMENT = 'OPC UA reconnect in progress' +OPC_COMMENT_SEPARATOR = ' | ' + + +def _opc_session_bad_comment(opc_status: str | None) -> str: + status = opc_status or 'Unknown' + return f'{OPC_SESSION_BAD_COMMENT_PREFIX} {status}' + + +def _apply_opc_write_error( + error_info: dict[str, Any] | None, + session_bad_seen: bool, + session_bad_status: str | None, + reconnect_in_progress_seen: bool, +) -> tuple[bool, str | None, bool]: + """ + Update session/reconnect flags from an OPC write error payload. + + Args: + error_info: Repository error details, or None when the write succeeded. + session_bad_seen: Whether a session_bad error was seen so far. + session_bad_status: Last known OPC status for session errors. + reconnect_in_progress_seen: Whether reconnect_in_progress was seen so far. + + Return: + Updated (session_bad_seen, session_bad_status, reconnect_in_progress_seen). + """ + if not error_info: + return session_bad_seen, session_bad_status, reconnect_in_progress_seen + + kind = error_info.get('opc_error_kind') + if kind == 'session_bad': + return True, error_info.get('opc_status', session_bad_status), reconnect_in_progress_seen + if kind == 'reconnect_in_progress': + return session_bad_seen, session_bad_status, True + return session_bad_seen, session_bad_status, reconnect_in_progress_seen class OPC(SientiaMonitoring): @@ -67,13 +106,14 @@ class OPC(SientiaMonitoring): logger=self.logger, notification_handler=self.notification_handler, metrics_controller=self.metrics_controller, + reconnection_interval=server.get('reconnection_interval', 60), server_uri=server['server_uri'], cert_path=server['cert_path'], private_key_path=server['private_key_path'], server_cert_path=server['server_cert_path'], ) - ok, err = self.opc_repository[opc_id].connect() - if not ok: + is_connected, error_data = self.opc_repository[opc_id].connect() + if not is_connected: self.send_notification( metadata={ 'model_id': '-', @@ -81,11 +121,11 @@ class OPC(SientiaMonitoring): 'workflow_name': '-', 'schedule_name': 'INITIALIZATION', }, - notification_id=f'OPC_CONNECTION_ERROR_{server.get("id", opc_id)}', - message=err.get('message', 'Failed to connect to OPC server'), - block='opc_repository', - level=NotificationLevel.ERROR, - attachment_content=err.get('attachment_content', traceback.format_exc()), + notification_id=error_data['notification_id'], + message=error_data['message'], + block=error_data['block'], + level=error_data.get('level', NotificationLevel.ERROR), + attachment_content=error_data.get('attachment_content', None), ) else: self.info( @@ -101,19 +141,13 @@ class OPC(SientiaMonitoring): data_type: str, tag_type: str, metadata: dict[str, Any], - ) -> float | None: + ) -> tuple[float | None, dict[str, Any] | None]: """ Write data to a specific OPC server tag with comprehensive error handling. - Args: - - server_id (str): The id of the OPC server. - - tag (str): The tag to write to. - - data (Any): The data to write. - - data_type (str): The data type. - - tag_type (str): The tag type. - - Returns: - - float | None: Response time in seconds if successful, None otherwise. + Return: + tuple[float | None, dict[str, Any] | None]: Response time on success, or + (None, error info_data) on repository failure. """ try: @@ -129,8 +163,8 @@ class OPC(SientiaMonitoring): level=info_data.get('level', NotificationLevel.ERROR), attachment_content=info_data.get('attachment_content', None), ) - return None - return info_data['response_time'] + return None, info_data + return info_data['response_time'], None except Exception as e: trace = traceback.format_exc() self.send_notification( @@ -147,10 +181,6 @@ class OPC(SientiaMonitoring): """ Validate that an OPC repository exists for the requested server identifier. - This guard prevents write attempts against unknown/uninitialized servers. - When the server is missing, it emits an error notification with the list - of available repositories to help operators diagnose configuration drift. - Args: - server_id (str): OPC server identifier from workflow output config. - metadata (dict[str, Any]): Workflow metadata used for logs/alerts. @@ -172,72 +202,116 @@ class OPC(SientiaMonitoring): return False return True + def _write_tags_from_config( + self, + server_id: str, + tags_config: dict[str, dict[str, Any]], + data: DataFrame, + data_column: str, + tag_type: str, + log_label: str, + metadata: dict[str, Any], + ) -> tuple[dict[str, float | None], bool, str | None, bool]: + """ + Write a group of OPC tags and collect response times and error flags. + + Args: + server_id: Target OPC server identifier. + tags_config: Tag name to configuration mapping. + data: DataFrame with prediction/confidence columns. + data_column: Column name whose first row value is written. + tag_type: Tag category passed to write_data ('prediction' or 'confidence'). + log_label: Human-readable label for success logs. + metadata: Context metadata for logging and notifications. + + Return: + (response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen) + """ + response_times: dict[str, float | None] = {} + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False + + for tag, tag_config in tags_config.items(): + response_time, error_info = self.write_data( + server_id=server_id, + tag=tag, + data=data.head(1)[data_column].values[0], + data_type=tag_config['data_type'], + tag_type=tag_type, + metadata=metadata, + ) + session_bad_seen, session_bad_status, reconnect_in_progress_seen = ( + _apply_opc_write_error( + error_info, + session_bad_seen, + session_bad_status, + reconnect_in_progress_seen, + ) + ) + if response_time is not None: + self.info( + f'{log_label} written to OPC server {server_id} for tag {tag}.', + metadata, + ) + response_times[tag] = response_time + + return response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen + def manage_output_tags( self, server_id: str, config: dict[str, Any], data: DataFrame, metadata: dict[str, Any], - ) -> tuple[bool, dict[str, float | None]]: + ) -> tuple[bool, dict[str, float | None], bool, str | None, bool]: """ Write prediction and confidence values for one OPC server configuration. - The method iterates through optional ``prediction_tags`` and - ``confidence_tags``, performs synchronous writes for each tag, collects - per-tag response times, and returns an aggregate success flag - (all tags successful) with a metrics-friendly response map. - - Args: - - server_id (str): Target OPC server id. - - config (dict[str, Any]): Server output configuration containing optional - ``prediction_tags`` and ``confidence_tags`` sections. - - data (DataFrame): Prediction dataframe used as source values. - - metadata (dict[str, Any]): Workflow metadata for logging/notifications. - Return: - tuple[bool, dict[str, float | None]]: Global success flag and response-time - map per tag (``None`` for failed writes). + tuple: success flag, per-tag response times, session_bad flags. """ - response_times: dict[str, float | None] = {} + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False - if 'prediction_tags' in config: - for tag, tag_config in config['prediction_tags'].items(): - response_time = self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - metadata=metadata, - ) - if response_time is not None: - self.info( - f'Prediction data written to OPC server {server_id} for tag {tag}.', - metadata, - ) - response_times[tag] = response_time - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - response_time = self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction_confidence'].values[0], - data_type=tag_config['data_type'], - tag_type='confidence', - metadata=metadata, - ) - if response_time is not None: - self.info( - f'Confidence data written to OPC server {server_id} for tag {tag}.', - metadata, - ) - response_times[tag] = response_time + tag_groups = ( + ('prediction_tags', 'prediction', 'prediction', 'Prediction data'), + ('confidence_tags', 'prediction_confidence', 'confidence', 'Confidence data'), + ) + for config_key, data_column, tag_type, log_label in tag_groups: + if config_key not in config: + continue + ( + group_times, + group_session_bad, + group_status, + group_reconnect, + ) = self._write_tags_from_config( + server_id=server_id, + tags_config=config[config_key], + data=data, + data_column=data_column, + tag_type=tag_type, + log_label=log_label, + metadata=metadata, + ) + response_times.update(group_times) + if group_session_bad: + session_bad_seen = True + session_bad_status = group_status or session_bad_status + if group_reconnect: + reconnect_in_progress_seen = True success = None not in response_times.values() - - return success, response_times + return ( + success, + response_times, + session_bad_seen, + session_bad_status, + reconnect_in_progress_seen, + ) @activity.defn(name='write_opc_data') def write_opc_data( @@ -246,19 +320,13 @@ class OPC(SientiaMonitoring): """ Execute OPC writes across all configured servers and collect per-tag metrics. - For each server in ``opc_output_config``, this activity validates server - availability, writes enabled prediction/confidence tags, accumulates - response-time metrics, and then normalizes confidence/comments in the - returned prediction payload when at least one write fails. - Args: - input_data (dict[str, Any]): Payload containing workflow metadata, data to write, and ``opc_output_config`` server/tag definitions. Return: tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]: Updated - prediction payload dict and nested metrics - ``{server_id: {tag_name: response_time_or_none}}``. + prediction payload dict and nested metrics per server/tag. """ metadata = input_data['metadata'] self.info('Writing data to OPC servers...', metadata) @@ -267,20 +335,32 @@ class OPC(SientiaMonitoring): self.info(f'Data to write: {data.size} rows', metadata) success = True + session_bad_seen = False + session_bad_status: str | None = None + reconnect_in_progress_seen = False - metrics: dict[str, dict[str, float | None]] = {} + opc_metrics: dict[str, dict[str, float | None]] = {} for server_id, config in opc_output_config.items(): if not self.validate_server(server_id, metadata): success = False continue - local_success, local_response_times = self.manage_output_tags( - server_id, config, data, metadata - ) - metrics[server_id] = local_response_times + ( + local_success, + local_response_times, + local_session_bad, + local_status, + local_reconnect_in_progress, + ) = self.manage_output_tags(server_id, config, data, metadata) + opc_metrics[server_id] = local_response_times local_count = len(local_response_times) success = success and local_success + if local_session_bad: + session_bad_seen = True + session_bad_status = local_status or session_bad_status + if local_reconnect_in_progress: + reconnect_in_progress_seen = True n_pred = len(config.get('prediction_tags') or {}) n_conf = len(config.get('confidence_tags') or {}) @@ -289,34 +369,55 @@ class OPC(SientiaMonitoring): metadata, ) - return self.process_confidence(data, success, metadata), metrics + return ( + self.process_confidence( + data, + success, + metadata, + session_bad=session_bad_seen, + opc_status=session_bad_status, + reconnect_in_progress=reconnect_in_progress_seen, + ), + opc_metrics, + ) def process_confidence( - self, data: DataFrame, success: bool, metadata: dict[str, Any] + self, + data: DataFrame, + success: bool, + metadata: dict[str, Any], + *, + session_bad: bool = False, + opc_status: str | None = None, + reconnect_in_progress: bool = False, ) -> dict[Hashable, Any]: """ Apply fallback confidence/comment values when OPC writes are not fully successful. - Args: - - data (DataFrame): Prediction dataframe to be returned to downstream steps. - - success (bool): Aggregate write status across all attempted OPC tags. - - metadata (dict[str, Any]): Workflow metadata used for debug logs. - Return: - dict[Hashable, Any]: Serialized dataframe dict with original values on success, - or downgraded confidence/comment fields on failure. + dict[Hashable, Any]: Serialized dataframe dict with updated confidence/comments on failure. """ - message = 'Some data could not be written to OPC servers' - if not success: - data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - data['comments'] = message + comment_parts: list[str] = [] + confidence = OPC_WRITTING_ERROR_CONFIDENCE + + if session_bad: + comment_parts.append(_opc_session_bad_comment(opc_status)) + confidence = OPC_SESSION_BAD_CONFIDENCE + if reconnect_in_progress: + comment_parts.append(OPC_RECONNECT_IN_PROGRESS_COMMENT) + confidence = OPC_SESSION_BAD_CONFIDENCE + if not comment_parts: + comment_parts.append(OPC_WRITTING_ERROR_MESSAGE) + + comments = OPC_COMMENT_SEPARATOR.join(comment_parts) + data['prediction_confidence'] = confidence + data['comments'] = comments self.debug( - f'{message}, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.', + f'OPC write issues, confidence={confidence}, comments={comments}', metadata, ) - else: self.debug('Data written to OPC servers successfully.', metadata) @@ -325,9 +426,6 @@ class OPC(SientiaMonitoring): def close(self) -> None: """ Disconnect all tracked OPC repositories and clear in-memory references. - - This method should be called during worker shutdown to ensure every - synchronous OPC session is explicitly closed before process exit. """ for opc in self.opc_repository.values(): diff --git a/laborious/metrics.py b/laborious/metrics.py index 776b826..74a9189 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -89,6 +89,40 @@ OPC_CONNECTION_STATUS = Gauge( ['pod_id', 'server_name', 'server_url'], ) +_OPC_SESSION_DEBUG_LABELS = ['pod_id', 'server_name', 'runtime', 'opc_server_id', 'session_id'] + +OPC_SESSION_CREATED_TOTAL = Counter( + 'opc_session_created_total', + 'OPC UA sessions established (after successful connect)', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_SESSION_CLOSED_TOTAL = Counter( + 'opc_session_closed_total', + 'OPC UA client disconnects completed (session tear-down initiated)', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_SESSION_REVISED_TIMEOUT_MS = Gauge( + 'opc_session_revised_timeout_milliseconds', + 'Server-revised OPC UA session timeout (RevisedSessionTimeout) in ms after connect', + _OPC_SESSION_DEBUG_LABELS, +) + +OPC_WRITE_ATTEMPT_LABELS = [*_OPC_SESSION_DEBUG_LABELS, 'model_id', 'model_name', 'result'] + +OPC_WRITE_ATTEMPTS_TOTAL = Counter( + 'opc_write_attempts_total', + 'OPC UA write attempts with session and outcome (result=OK or exception class name)', + OPC_WRITE_ATTEMPT_LABELS, +) + +OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL = Counter( + 'opc_write_inter_arrival_over_session_timeout_total', + 'Successful writes where seconds since the previous successful write exceeded RevisedSessionTimeout (ms)', + _OPC_SESSION_DEBUG_LABELS, +) + # ================== Model metrics ================== MODEL_READ_LAG = Histogram( diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 029f427..1e9b0e1 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -3,8 +3,6 @@ from os import getenv from typing import Any - - def build_mlflow_config() -> dict[str, Any]: """ Read MLflow tracking and registry credentials from the environment. diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 15faf79..64854ff 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,12 +1,14 @@ """ -Synchronous OPC UA client repository using asyncua (opcua-asyncio) ``sync`` API. +Synchronous OPC UA client repository using asyncua ``sync`` API. -``asyncua.sync.Client`` runs the asyncio client on a background thread so callers -stay synchronous. Connect/disconnect, optional Basic256 security, session checks, -and typed writes mirror the previous python-opcua integration. +``asyncua.sync.Client`` runs the asyncio stack on a background thread so Temporal +activities and other callers stay blocking while preserving the same session +lifecycle, security policy, reconnect semantics, and write error classification +as the async ``origin/main`` implementation. """ import json +import threading import time import traceback from datetime import datetime @@ -16,6 +18,7 @@ from typing import Any from asyncua import ua from asyncua.crypto import security_policies from asyncua.sync import Client +from asyncua.ua.uaerrors import UaStatusCodeError from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -24,6 +27,119 @@ from sientia_do.observability.sientia_monitoring import SientiaMonitoring from laborious import metrics +# Requested session and secure channel lifetime (ms) before server revision; 10 minutes. +OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS = 10 * 60 * 1000 + + +class OpcClientAlreadyExistsError(RuntimeError): + """Raised when _create_client is called while self.client is already set.""" + + +class OpcSessionAlreadyConnectedError(RuntimeError): + """Raised when _open_session is called while a UA session is already open.""" + + +class OpcClientNotInitializedError(RuntimeError): + """Raised when _open_session is called before _create_client.""" + + +RECONNECTABLE_OPC_BAD_NAMES: frozenset[str] = frozenset( + { + 'BadSessionIdInvalid', + 'BadSessionClosed', + 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', + 'BadSecureChannelClosed', + 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', + 'BadServerNotConnected', + 'BadConnectionClosed', + 'BadDisconnect', + 'BadConnectionRejected', + 'BadCommunicationError', + 'BadRequestInterrupted', + 'BadUnknownResponse', + 'BadTimeout', + 'BadRequestTimeout', + 'BadSequenceNumberInvalid', + 'BadSequenceNumberUnknown', + 'BadSecurityModeInsufficient', + 'BadRequestHeaderInvalid', + 'BadInvalidState', + } +) + + +def _opc_authentication_token_str(client: Client | None) -> str: + """ + Serialize the current OPC UA authentication token (session handle) for logging and metrics. + + Return: + str: Token string, or "unknown" if unavailable. + """ + if client is None: + return 'unknown' + try: + proto = client.aio_obj.uaclient.protocol + if proto is None: + return 'unknown' + tok = getattr(proto, 'authentication_token', None) + if tok is None: + return 'unknown' + return str(tok) + except Exception: + return 'unknown' + + +def _opc_status_from_exception(exc: BaseException) -> str: + """ + Resolve OPC UA status name from an exception, including chained UaStatusCodeError causes. + + Args: + exc (BaseException): Raised error from asyncua. + + Return: + str: Status class name or generic Python exception name. + """ + current: BaseException | None = exc + while current is not None: + if isinstance(current, UaStatusCodeError): + return type(current).__name__ + current = current.__cause__ + return type(exc).__name__ + + +def is_reconnectable_opcua_bad(exc: BaseException) -> bool: + """ + Return whether the exception is a Tier-1 OPC UA Bad* that should trigger reconnect. + + Args: + exc (BaseException): Raised error from get_node or write_value. + + Return: + bool: True if reconnect should be scheduled. + """ + return _opc_status_from_exception(exc) in RECONNECTABLE_OPC_BAD_NAMES + + +def _model_labels_from_write_metadata(metadata: dict[str, Any] | None) -> dict[str, str]: + """ + Extract model_id and model_name from write metadata for Prometheus labels. + + Args: + metadata (dict[str, Any] | None): Context passed into write_data; may omit keys. + + Return: + dict[str, str]: Labels model_id and model_name, defaulting to "unknown". + """ + if not metadata: + return {'model_id': 'unknown', 'model_name': 'unknown'} + return { + 'model_id': str(metadata.get('model_id', 'unknown')), + 'model_name': str(metadata.get('model_name', 'unknown')), + } + + data_type_map = { 'float': { 'converter': float, @@ -52,11 +168,8 @@ class OpcRepository(SientiaMonitoring): """ Synchronous OPC UA repository for connect/disconnect and typed writes. - Attributes: - url: OPC UA endpoint URL. - id: Server identifier used in metrics and notifications. - server_name: Human-readable server name for labels. - client: Active ``asyncua.sync.Client`` while connected. + Uses ``asyncua.sync.Client`` with the same session metrics, Tier-1 Bad* reconnect, + and structured write error payloads as the async repository on ``origin/main``. """ def __init__( @@ -80,10 +193,10 @@ class OpcRepository(SientiaMonitoring): self.cert_path = cert_path self.private_key_path = private_key_path self.server_cert_path = server_cert_path - self.error_count = 0 self.reconnection_interval = reconnection_interval self.last_reconnection_time: datetime | None = None self.disconnection_interval = 10.0 + self.notification_handler = notification_handler self.client: Client | None = None SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) @@ -94,15 +207,63 @@ class OpcRepository(SientiaMonitoring): 'workflow_name': 'opc_repository', 'schedule_name': '-', } + self._last_write_mono: float | None = None + self._connection_lock = threading.Lock() + self._session_ready = threading.Event() + self._reconnect_thread: threading.Thread | None = None + + def _opc_debug_tags(self, session_id: str) -> dict[str, str]: + return { + 'pod_id': str(getattr(self, 'pod_id', 'unknown')), + 'server_name': self.server_name, + 'runtime': str(getattr(self, 'runtime', 'unknown')), + 'opc_server_id': self.id, + 'session_id': session_id, + } + + def _is_session_open(self) -> bool: + """ + Return whether the asyncua client has an open transport session. + + Return: + bool: True when protocol exists and is not closed. + """ + if self.client is None: + return False + try: + proto = self.client.aio_obj.uaclient.protocol + return proto is not None and proto.state != 'closed' + except Exception: + return False + + def _reconnection_window_elapsed(self) -> bool: + """ + Return whether enough time has passed since the last reconnect attempt. + + Return: + bool: True if a new reconnect is allowed. + """ + if self.last_reconnection_time is None: + return True + return ( + datetime.now() - self.last_reconnection_time + ).total_seconds() > self.reconnection_interval + + def _not_connected_error(self) -> dict[str, Any]: + return { + 'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}', + 'message': f'OPC server {self.id} is not connected', + 'block': 'opc_repository', + 'level': NotificationLevel.WARNING, + } def set_security(self) -> None: """ - Configure Basic256 security policy, certificates, and long channel/session timeouts. + Configure certificates and timeouts on the sync asyncua client. Raises: - ValueError: If certificate paths are missing or client is not initialized. + ValueError: If cert paths or client are missing. """ - if self.cert_path is None or self.private_key_path is None: raise ValueError( 'Certificate and private key paths must be provided for secure connection.' @@ -115,7 +276,7 @@ class OpcRepository(SientiaMonitoring): if self.client is None: raise ValueError('Client must be initialized before setting security') - self.client.application_uri = self.server_uri or self.client.application_uri + self.client.application_uri = self.server_uri self.info('Setting security...', self.metadata) self.client.set_security( security_policies.SecurityPolicyBasic256, @@ -124,75 +285,105 @@ class OpcRepository(SientiaMonitoring): None, str(server_cert) if server_cert else None, ) - self.client.aio_obj.secure_channel_timeout = 10000000 - self.client.aio_obj.session_timeout = 10000000 + aio = self.client.aio_obj + aio.secure_channel_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS + aio.session_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS - def connect(self) -> tuple[bool, dict[str, Any]]: + def _create_client(self) -> None: """ - Create the synchronous client, optionally apply security, and connect to the server. + Instantiate the sync Client and apply security when configured. - Return: - tuple[bool, dict[str, Any]]: Success flag and error payload when False. + Caller must hold _connection_lock. Does not open a UA session. + + Raises: + OpcClientAlreadyExistsError: If self.client is already set. """ + if self.client is not None: + raise OpcClientAlreadyExistsError( + f'OPC client already exists for server {self.id}; ' + 'call disconnect() before creating a new client' + ) self.client = Client(self.url, timeout=10) - - self.client.aio_obj.name = self.pod_id - self.client.aio_obj.description = self.pod_id + aio = self.client.aio_obj + if hasattr(aio, 'watchdog_intervall'): + aio.watchdog_intervall = 50 + aio.name = self.pod_id + aio.description = self.pod_id pod_uri = self.pod_id.replace('-', ':') self.client.application_uri = pod_uri - self.client.aio_obj.product_uri = pod_uri - + aio.product_uri = pod_uri if self.cert_path: self.set_security() - self.info( - f'Starting connection to OPC server {self.id}:{self.server_name}...', self.metadata - ) - return self.try_connect() - def try_connect(self) -> tuple[bool, dict[str, Any]]: + def _open_session(self) -> tuple[bool, dict[str, Any]]: """ - Perform the TCP/session handshake and emit connection metrics. + Open the OPC UA session on the existing client. + + Caller must hold _connection_lock. + + Raises: + OpcClientNotInitializedError: If self.client is None. + OpcSessionAlreadyConnectedError: If a session is already open. Return: - tuple[bool, dict[str, Any]]: Success flag and structured error when False. + tuple[bool, dict[str, Any]]: Success flag and error payload on connect failure. """ + if self.client is None: + raise OpcClientNotInitializedError( + f'OPC client is not initialized for server {self.id}; ' + 'call _create_client() before opening a session' + ) + if self._is_session_open(): + raise OpcSessionAlreadyConnectedError( + f'OPC session already connected for server {self.id}; ' + 'call disconnect() before connecting again' + ) tags = { 'pod_id': self.pod_id, 'server_name': self.server_name, } self.emit_metric_sync(metrics.OPC_CONNECTIONS_TOTAL, tags) + try: - self.last_reconnection_time = datetime.now() - if self.client is None: - return False, { - 'notification_id': f'OPC_CONNECTION_ERROR_{self.id}', - 'message': 'Client is not initialized', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - } self.client.connect() + aio = self.client.aio_obj + session_id = _opc_authentication_token_str(self.client) + revised_session_timeout_ms = int(aio.session_timeout) + revised_secure_channel_timeout_ms = int(aio.secure_channel_timeout) + self.info( + f'OPC new session connected opc_server_id={self.id} session_id={session_id} ' + f'revised_session_timeout_ms={revised_session_timeout_ms} ' + f'revised_secure_channel_timeout_ms={revised_secure_channel_timeout_ms}', + self.metadata, + ) + self.emit_metric_sync( + metrics.OPC_SESSION_CREATED_TOTAL, self._opc_debug_tags(session_id) + ) + self.emit_metric_sync( + metric_object=metrics.OPC_SESSION_REVISED_TIMEOUT_MS, + method='set', + tags=self._opc_debug_tags(session_id), + value=revised_session_timeout_ms, + ) self.emit_metric_sync( metric_object=metrics.OPC_CONNECTION_STATUS, method='set', - tags={ - **tags, - 'server_url': self.url, - }, + tags={**tags, 'server_url': self.url}, value=1, ) + self._last_write_mono = None + self._session_ready.set() return True, {} - except Exception as e: - self.disconnect() + except Exception as e: + self._disconnect_locked() trace = traceback.format_exc() self.error(trace, self.metadata) - self.emit_metric_sync(metrics.OPC_CONNECTIONS_FAILED, tags) - return False, { 'notification_id': f'OPC_CONNECTION_ERROR_{self.id}', 'message': f'Failed to connect to OPC server: {e}', @@ -201,20 +392,38 @@ class OpcRepository(SientiaMonitoring): 'attachment_content': trace, } - def disconnection_fallback(self) -> list[dict[str, Any]]: + def _connect_locked(self) -> tuple[bool, dict[str, Any]]: """ - Retry disconnect up to five times with linear backoff. + Create the client when absent, then open a UA session. + + Caller must hold _connection_lock. + + Raises: + OpcSessionAlreadyConnectedError: If a session is already open. Return: - list[dict[str, Any]]: Empty on success, otherwise error records per attempt. + tuple[bool, dict[str, Any]]: Result from _open_session on connect failure. """ + if self._is_session_open(): + raise OpcSessionAlreadyConnectedError( + f'OPC session already connected for server {self.id}; ' + 'call disconnect() before connecting again' + ) + if self.client is None: + self._create_client() + return self._open_session() + def _disconnection_fallback(self) -> list[dict[str, Any]]: + """ + Try up to five times to disconnect from the OPC UA server. + """ assert self.client is not None - error_stack = [] + error_stack: list[dict[str, Any]] = [] for i in range(5): try: self.info( - f'Disconnecting from OPC UA server, attempt {i + 1} of 5', self.metadata + f'Disconnecting from OPC UA server, attempt {i + 1} of 5', + self.metadata, ) self.client.disconnect() return [] @@ -233,15 +442,26 @@ class OpcRepository(SientiaMonitoring): time.sleep(self.disconnection_interval * i) return error_stack - def disconnect(self) -> None: + def _disconnect_locked(self) -> None: """ - Tear down the UA session and reset connection metrics. + Tear down the current session and client. + + Caller must hold _connection_lock. """ + self._last_write_mono = None + self._session_ready.clear() if self.client is None: return - errors = self.disconnection_fallback() + session_id = _opc_authentication_token_str(self.client) + self.info( + f'OPC disconnecting opc_server_id={self.id} session_id={session_id}', + self.metadata, + ) + self.emit_metric_sync(metrics.OPC_SESSION_CLOSED_TOTAL, self._opc_debug_tags(session_id)) + + errors = self._disconnection_fallback() if errors: self.send_notification( metadata=self.metadata, @@ -252,9 +472,8 @@ class OpcRepository(SientiaMonitoring): attachment_content=json.dumps(errors, indent=4), ) else: - self.warning( - f'Disconnected from OPC server {self.id} successfully', self.metadata - ) + self.warning(f'Disconnected from OPC server {self.id} successfully', self.metadata) + self.emit_metric_sync( metric_object=metrics.OPC_CONNECTION_STATUS, method='set', @@ -265,139 +484,313 @@ class OpcRepository(SientiaMonitoring): }, value=0, ) - self.client = None - def _session_alive(self) -> bool: + def _reconnect_locked(self) -> tuple[bool, dict[str, Any]]: """ - Best-effort check that the synchronous client still has a working session. + Close the current session and open a new one. + + Caller must hold _connection_lock. Records last_reconnection_time for interval gating. Return: - bool: True if a root browse succeeds, False otherwise. + tuple[bool, dict[str, Any]]: Result from _connect_locked after teardown. """ + self.last_reconnection_time = datetime.now() + self._disconnect_locked() + return self._connect_locked() - if self.client is None: - return False - try: - self.client.get_root_node() - return True - except Exception: - return False + def connect(self) -> tuple[bool, dict[str, Any]]: + """ + Open an OPC UA session under the connection lock (worker initialization). + """ + with self._connection_lock: + self.info( + f'Starting connection to OPC server {self.id}:{self.server_name}...', + self.metadata, + ) + return self._connect_locked() + + def disconnect(self) -> None: + """ + Gracefully disconnect from the OPC server under the connection lock. + """ + with self._connection_lock: + self._disconnect_locked() def validate_connection(self) -> tuple[bool, dict[str, Any]]: """ - Ensure the UA session is usable; reconnect when outside the backoff window. + Read-only check that the asyncua protocol is open. + + Caller must ensure _session_ready before writing. Does not connect or reconnect. Return: - tuple[bool, dict[str, Any]]: Whether the session is ready and optional error payload. + tuple[bool, dict[str, Any]]: (True, {}) when open, otherwise (False, error). """ - - if self.client is None: - return self.connect() - - try: - if not self._session_alive(): - self.error(f'OPC server {self.id} is not connected', self.metadata) - if ( - self.last_reconnection_time is None - or (datetime.now() - self.last_reconnection_time).total_seconds() - > self.reconnection_interval - ): - self.disconnect() - self.info( - f'Trying to reconnect to OPC server {self.id}...', self.metadata - ) - return self.connect() - - return False, { - 'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}', - 'message': ( - f'OPC server {self.id} is not connected, waiting for next reconnection window...' - ), - 'block': 'opc_repository', - 'level': NotificationLevel.WARNING, - } + if self._is_session_open(): return True, {} - except Exception as e: - trace = traceback.format_exc() - message = f'Failed to validate connection to OPC server: {e}' - self.error(message, self.metadata) - return False, { - 'notification_id': f'OPC_CONNECTION_CHECK_ERROR_{self.id}', - 'message': message, - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } + self.error(f'OPC server {self.id} is not connected', self.metadata) + return False, self._not_connected_error() + + def _start_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + """ + Schedule a background reconnect if interval and thread state allow it. + + Args: + opc_status (str): OPC UA status name that triggered reconnect. + session_id (str): Session token before failure. + """ + if not self._reconnection_window_elapsed(): + self.warning( + f'OPC reconnect skipped reason=reconnection_window opc_server_id={self.id} ' + f'opc_status={opc_status}', + self.metadata, + ) + return + if self._reconnect_thread is not None and self._reconnect_thread.is_alive(): + self.warning( + f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} ' + f'opc_status={opc_status}', + self.metadata, + ) + return + + self._session_ready.clear() + self.info( + f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} ' + f'old_session_id={session_id}', + self.metadata, + ) + self._reconnect_thread = threading.Thread( + target=self._run_reconnect_on_bad, + args=(opc_status, session_id), + daemon=True, + ) + self._reconnect_thread.start() + + def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + """ + Tear down and re-establish the OPC UA session under the connection lock. + + Args: + opc_status (str): OPC UA status that triggered reconnect. + session_id (str): Previous session token string. + """ + try: + with self._connection_lock: + self.info( + f'OPC reconnect started opc_status={opc_status} opc_server_id={self.id} ' + f'old_session_id={session_id}', + self.metadata, + ) + self._reconnect_locked() + except Exception: + self.error( + f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}', + self.metadata, + ) + self.error(traceback.format_exc(), self.metadata) + + def _log_write_inter_arrival(self, session_id: str, node: str) -> None: + """ + Log elapsed wall time since the previous successful OPC write on this repository. + + Args: + session_id (str): Current OPC UA session token string. + node (str): Node id written in this operation. + """ + now = time.monotonic() + if self._last_write_mono is not None: + delta_s = now - self._last_write_mono + self.info( + f'OPC write inter-arrival_s={delta_s:.6f} opc_server_id={self.id} ' + f'session_id={session_id} node={node}', + self.metadata, + ) + if self.client is not None: + session_timeout_ms = float(self.client.aio_obj.session_timeout) + if session_timeout_ms > 0 and delta_s > (session_timeout_ms / 1000.0): + self.emit_metric_sync( + metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL, + self._opc_debug_tags(session_id), + ) + self._last_write_mono = now + + def _emit_opc_write_metric( + self, session_id: str, result: str, metadata: dict[str, Any] | None + ) -> None: + self.emit_metric_sync( + metrics.OPC_WRITE_ATTEMPTS_TOTAL, + { + **self._opc_debug_tags(session_id), + **_model_labels_from_write_metadata(metadata), + 'result': result, + }, + ) + + def _write_failure_payload( + self, + notification_id: str, + message: str, + level: NotificationLevel = NotificationLevel.ERROR, + attachment_content: str | None = None, + opc_error_kind: str | None = None, + opc_status: str | None = None, + ) -> dict[str, Any]: + payload: dict[str, Any] = { + 'notification_id': notification_id, + 'message': message, + 'block': 'opc_repository', + 'level': level, + } + if attachment_content is not None: + payload['attachment_content'] = attachment_content + if opc_error_kind is not None: + payload['opc_error_kind'] = opc_error_kind + if opc_status is not None: + payload['opc_status'] = opc_status + return payload + + def _handle_tier1_bad( + self, + exc: BaseException, + session_id: str, + node: str, + metadata: dict[str, Any], + phase: str, + ) -> tuple[bool, dict[str, Any]]: + """ + Record metrics/logs and schedule reconnect after a Tier-1 Bad* error. + + Args: + exc (BaseException): Tier-1 OPC UA error. + session_id (str): Session token at failure time. + node (str): Node id being written. + metadata (dict[str, Any]): Write context. + phase (str): get_node or write_value. + + Return: + tuple[bool, dict[str, Any]]: Always (False, error payload). + """ + opc_status = _opc_status_from_exception(exc) + trace = traceback.format_exc() + self.error(trace, metadata) + self._emit_opc_write_metric(session_id, opc_status, metadata) + self.error( + f'OPC write failed opc_status={opc_status} opc_server_id={self.id} ' + f'session_id={session_id} model_id={metadata.get("model_id", "unknown")} ' + f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}', + metadata, + ) + self._start_reconnect_on_bad(opc_status, session_id) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}', + message=f'Failed to {phase} on OPC server: {exc} | metadata: {metadata}', + attachment_content=trace, + opc_error_kind='session_bad', + opc_status=opc_status, + ) + + @staticmethod + def _write_node_value( + node_obj: Any, + ua_data: ua.DataValue, + data: Any, + variant_type: ua.VariantType, + ) -> None: + """ + Write a DataValue to a node, falling back to set_value when write_value is unavailable. + + Args: + node_obj: Sync or async node wrapper from asyncua. + ua_data (ua.DataValue): Encoded value for write_value. + data: Scalar converted value for set_value fallback. + variant_type (ua.VariantType): OPC UA type for set_value fallback. + """ + if hasattr(node_obj, 'write_value'): + try: + node_obj.write_value(ua_data) + return + except (AttributeError, TypeError): + pass + node_obj.set_value(data, variant_type) def write_data( self, node: str, value: Any, data_type: str, metadata: dict[str, Any] ) -> tuple[bool, dict[str, Any]]: """ - Write a typed value to an OPC UA node after validating connectivity. - - Args: - node: Node id string accepted by ``Client.get_node``. - value: Scalar value to encode. - data_type: Key into ``data_type_map`` (e.g. float, str). - metadata: Workflow metadata for error context. - - Return: - tuple[bool, dict[str, Any]]: Success flag and either ``response_time`` or error fields. + Write data to OPC server with a single attempt and Tier-1 Bad* reconnect scheduling. """ + if not self._session_ready.is_set(): + self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata) + self.warning( + f'OPC write rejected reconnect_in_progress opc_server_id={self.id} ' + f'model_id={metadata.get("model_id", "unknown")} ' + f'model_name={metadata.get("model_name", "unknown")}', + metadata, + ) + return False, { + 'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}', + 'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}', + 'block': 'opc_repository', + 'level': NotificationLevel.WARNING, + 'opc_error_kind': 'reconnect_in_progress', + } is_connected, error = self.validate_connection() - if not is_connected: + self._emit_opc_write_metric('unknown', 'NotConnected', metadata) return False, error start_time = time.time() + session_id = _opc_authentication_token_str(self.client) try: assert self.client is not None node_obj = self.client.get_node(node) except Exception as e: + if is_reconnectable_opcua_bad(e): + return self._handle_tier1_bad(e, session_id, node, metadata, 'get_node') trace = traceback.format_exc() self.error(trace, metadata) - self.error_count += 1 - return False, { - 'notification_id': f'OPC_WRITE_GET_NODE_ERROR_{self.id}', - 'message': f'Failed to get node from OPC server: {e} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } + self._emit_opc_write_metric(session_id, f'GetNodeError:{type(e).__name__}', metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_GET_NODE_ERROR_{self.id}', + message=f'Failed to get node from OPC server: {e} | metadata: {metadata}', + attachment_content=trace, + ) if data_type not in data_type_map: - return False, { - 'notification_id': f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}', - 'message': f'Unsupported data type: {data_type} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - } + self._emit_opc_write_metric(session_id, 'UnsupportedDataType', metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}', + message=f'Unsupported data type: {data_type} | metadata: {metadata}', + ) data = data_type_map[data_type]['converter'](value) self.info(f'Writing {data} - {type(data)} to {node}', metadata) variant_type = data_type_map[data_type]['opc_type'] + ua_data = ua.DataValue( + ua.Variant(data, variant_type), + ) try: - node_obj.set_value(data, variant_type) - + self._write_node_value(node_obj, ua_data, data, variant_type) end_time = time.time() response_time = end_time - start_time - except Exception as e: + if is_reconnectable_opcua_bad(e): + return self._handle_tier1_bad(e, session_id, node, metadata, 'write_value') trace = traceback.format_exc() self.error(trace, metadata) - self.error_count += 1 - return False, { - 'notification_id': f'OPC_WRITE_DATA_ERROR_{self.id}', - 'message': f'Failed to write data to OPC server: {e} | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': trace, - } - self.error_count = 0 + self._emit_opc_write_metric(session_id, type(e).__name__, metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}', + message=f'Failed to write data to OPC server: {e} | metadata: {metadata}', + attachment_content=trace, + ) + + self._emit_opc_write_metric(session_id, 'OK', metadata) + self._log_write_inter_arrival(session_id, node) return True, { 'response_time': response_time, diff --git a/pyproject.toml b/pyproject.toml index 5123123..aaf522d 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 E2E tests that use in-process asyncua + real OpcRepository", ] [tool.coverage.run] diff --git a/tests/laborious/activities/test_mlflow.py b/tests/laborious/activities/test_mlflow.py index 49e51c8..3b98c83 100644 --- a/tests/laborious/activities/test_mlflow.py +++ b/tests/laborious/activities/test_mlflow.py @@ -377,6 +377,7 @@ def test_retrain_model_success_data_success_retrain( mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow ): mock_mkdtemp.return_value = 'tmp' + mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx) mv_alias = MagicMock() mv_alias.run_id = 'source-run' @@ -470,6 +471,7 @@ def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag( mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow ): mock_mkdtemp.return_value = 'tmp' + mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx) mv_alias = MagicMock(run_id='src') mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias wrapper = MagicMock() diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index c82e4ba..4c41109 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -229,7 +229,6 @@ def test_calculate_drift_empty_drift_df(model_metrics_activity): def test_calculate_drift_empty_after_timestamp_filter(model_metrics_activity): """Rows dropped by target-window alignment yield an empty export list, not an insufficient-data error.""" - ts_target = Timestamp('2023-05-26 11:12:27') drift_df = _sample_drift_metrics_df(Timestamp('2020-01-01')) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) @@ -271,7 +270,7 @@ def test_calculate_drift_drift_insufficient_data_error_from_lib( lib_msg = ( '[MODEL_METRICS_DRIFT_INSUFFICIENT_DATA] Drift analysis produced no time chunks ' - '(chunk_period=\'min\', analysis_rows=1).' + "(chunk_period='min', analysis_rows=1)." ) model_metrics_activity.get_drift_metrics = MagicMock( side_effect=DriftInsufficientDataError(lib_msg, analysis_rows=1) diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index 0f4d3d9..f4e3078 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -5,7 +5,16 @@ from pandas import DataFrame from pytest import mark from sientia_do.notifications.models import NotificationLevel -from laborious.activities.opc import OPC +from laborious.activities.opc import ( + OPC, + OPC_COMMENT_SEPARATOR, + OPC_RECONNECT_IN_PROGRESS_COMMENT, + OPC_SESSION_BAD_COMMENT_PREFIX, + OPC_SESSION_BAD_CONFIDENCE, + OPC_WRITTING_ERROR_CONFIDENCE, + OPC_WRITTING_ERROR_MESSAGE, + _apply_opc_write_error, +) metadata = { 'metadata': { @@ -35,20 +44,26 @@ def test__init__(): def test_init_opc(mock_send_notification, mock_opc_repository): mock_logger = MagicMock() mock_metrics_controller = MagicMock() - server1 = MagicMock() - server1.connect.return_value = (True, {}) - server2 = MagicMock() - server2.connect.return_value = (True, {}) - server3 = MagicMock() - server3.connect.return_value = ( - False, - { - 'notification_id': 'OPC_CONNECTION_ERROR_server3', - 'message': 'Failed to connect to OPC server: Test error', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': 'Test error', - }, + server1 = MagicMock( + connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {})) + ) + server2 = MagicMock( + connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {})) + ) + server3 = MagicMock( + connect=MagicMock( + return_value=( + False, + { + 'notification_id': 'OPC_CONNECTION_ERROR_server3', + 'message': 'Failed to connect to OPC server: Test error', + 'block': 'opc_repository', + 'level': NotificationLevel.ERROR, + 'attachment_content': 'Test error', + }, + ) + ), + write_data=MagicMock(return_value=(True, {})), ) mock_opc_repository.side_effect = [server1, server2, server3] mock_notification_handler = MagicMock() @@ -105,12 +120,13 @@ def test_init_opc(mock_send_notification, mock_opc_repository): server_name='server1', url='http://localhost:8080', logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, server_uri='opc.tcp://localhost:4840', cert_path='', private_key_path='', server_cert_path='', + notification_handler=mock_notification_handler, + reconnection_interval=60, + metrics_controller=mock_metrics_controller, ), ] ) @@ -121,12 +137,13 @@ def test_init_opc(mock_send_notification, mock_opc_repository): server_name='server2', url='http://localhost:8080', logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, server_uri='opc.tcp://localhost:4840', cert_path='', private_key_path='', server_cert_path='', + notification_handler=mock_notification_handler, + reconnection_interval=60, + metrics_controller=mock_metrics_controller, ) ] ) @@ -154,40 +171,40 @@ def test_init_opc(mock_send_notification, mock_opc_repository): @pytest.fixture -def opc(): - with patch('laborious.activities.opc.OpcRepository') as mock_opc_repository: - mock_opc_repository.return_value.write_data = MagicMock(return_value=(True, {})) - mock_opc_repository.return_value.connect = MagicMock(return_value=(True, {})) - servers = { - 'server1': { - 'id': 'server1', - 'server_name': 'server1', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - } +@patch('laborious.activities.opc.OpcRepository') +def opc(mock_opc_repository): + servers = { + 'server1': { + 'id': 'server1', + 'server_name': 'server1', + 'url': 'http://localhost:8080', + 'server_uri': 'opc.tcp://localhost:4840', + 'cert_path': '', + 'private_key_path': '', + 'server_cert_path': '', + 'reconnection_interval': 60, } + } - opc_instance = OPC( - opc_servers=servers, - logger=MagicMock(), - notification_handler=MagicMock(), - metrics_controller=MagicMock(), - ) - opc_instance.init_opc() - opc_instance.send_notification = MagicMock() - opc_instance.emit_metric_sync = MagicMock() - yield opc_instance + mock_opc_repository.return_value.write_data = MagicMock(return_value=(True, {})) + mock_opc_repository.return_value.connect = MagicMock(return_value=(True, {})) + opc = OPC( + opc_servers=servers, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(), + ) + opc.init_opc() + opc.send_notification = MagicMock() + opc.emit_metric_sync = MagicMock() + return opc WRITE_DATA_CASES = [ ('tag1', 'int', 50), ('tag2', 'float', 50.5), ('tag3', 'bool', True), - ('tag4', 'str', 'test'), + ('tag4', 'string', 'test'), ] @@ -195,7 +212,7 @@ WRITE_DATA_CASES = [ def test_write_data_success(opc, tag, data_type, data): opc.opc_repository['server1'].write_data.return_value = (True, {'response_time': 0.1}) - result = opc.write_data( + response_time, error_info = opc.write_data( server_id='server1', tag=tag, data=data, @@ -203,10 +220,9 @@ def test_write_data_success(opc, tag, data_type, data): tag_type='prediction', metadata=metadata, ) - assert result == 0.1 - opc.opc_repository['server1'].write_data.assert_called_once_with( - tag, data, data_type, metadata - ) + assert response_time == 0.1 + assert error_info is None + opc.opc_repository['server1'].write_data.assert_called_once_with(tag, data, data_type, metadata) def test_write_data_failed(opc): @@ -221,7 +237,7 @@ def test_write_data_failed(opc): }, ) - result = opc.write_data( + response_time, error_info = opc.write_data( server_id='server1', tag='tag1', data=50, @@ -229,7 +245,8 @@ def test_write_data_failed(opc): tag_type='prediction', metadata=metadata, ) - assert result is None + assert response_time is None + assert error_info is not None opc.send_notification.assert_called_once_with( metadata=metadata, @@ -268,16 +285,205 @@ def test_write_data_exception(opc): raise AssertionError('Expected an exception to be raised') -def test_manage_output_tags_success(opc): - opc.write_data = MagicMock(return_value=0.1) +@mark.parametrize( + 'error_info,initial_seen,initial_status,initial_reconnect,expected', + [ + (None, False, None, False, (False, None, False)), + ({}, False, None, False, (False, None, False)), + ( + {'opc_error_kind': 'session_bad', 'opc_status': 'BadSessionIdInvalid'}, + False, + None, + False, + (True, 'BadSessionIdInvalid', False), + ), + ( + {'opc_error_kind': 'session_bad', 'opc_status': 'NewStatus'}, + True, + 'OldStatus', + False, + (True, 'NewStatus', False), + ), + ( + {'opc_error_kind': 'session_bad'}, + True, + 'KeptStatus', + False, + (True, 'KeptStatus', False), + ), + ( + {'opc_error_kind': 'reconnect_in_progress'}, + False, + None, + False, + (False, None, True), + ), + ( + {'opc_error_kind': 'other'}, + True, + 'Status', + True, + (True, 'Status', True), + ), + ], +) +def test_apply_opc_write_error( + error_info, initial_seen, initial_status, initial_reconnect, expected +): + result = _apply_opc_write_error( + error_info, + initial_seen, + initial_status, + initial_reconnect, + ) + assert result == expected + +def test_write_tags_from_config_prediction_success(opc): + opc.write_data = MagicMock(return_value=(0.1, None)) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + tags_config = {'tag1': {'data_type': 'float'}} + + response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config( + server_id='server1', + tags_config=tags_config, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': 0.1} + assert session_bad is False + assert opc_status is None + assert reconnect is False + opc.write_data.assert_called_once_with( + server_id='server1', + tag='tag1', + data=0.75, + data_type='float', + tag_type='prediction', + metadata=metadata['metadata'], + ) + + +def test_write_tags_from_config_confidence_success(opc): + opc.write_data = MagicMock(return_value=(0.2, None)) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + tags_config = {'tag2': {'data_type': 'float'}} + + response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config( + server_id='server1', + tags_config=tags_config, + data=data, + data_column='prediction_confidence', + tag_type='confidence', + log_label='Confidence data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag2': 0.2} + assert session_bad is False + assert opc_status is None + assert reconnect is False + opc.write_data.assert_called_once_with( + server_id='server1', + tag='tag2', + data=0.95, + data_type='float', + tag_type='confidence', + metadata=metadata['metadata'], + ) + + +def test_write_tags_from_config_write_failure(opc): + opc.write_data = MagicMock(return_value=(None, {})) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is False + assert opc_status is None + assert reconnect is False + + +def test_write_tags_from_config_session_bad(opc): + opc.write_data = MagicMock( + return_value=( + None, + { + 'opc_error_kind': 'session_bad', + 'opc_status': 'BadSessionIdInvalid', + }, + ) + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is True + assert opc_status == 'BadSessionIdInvalid' + assert reconnect is False + + +def test_write_tags_from_config_reconnect_in_progress(opc): + opc.write_data = MagicMock( + return_value=( + None, + {'opc_error_kind': 'reconnect_in_progress'}, + ) + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + + response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config( + server_id='server1', + tags_config={'tag1': {'data_type': 'float'}}, + data=data, + data_column='prediction', + tag_type='prediction', + log_label='Prediction data', + metadata=metadata['metadata'], + ) + + assert response_times == {'tag1': None} + assert session_bad is False + assert opc_status is None + assert reconnect is True + + +def test_manage_output_tags_success(opc): + opc._write_tags_from_config = MagicMock( + side_effect=[ + ({'tag1': 0.1}, False, None, False), + ({'tag2': 0.1}, False, None, False), + ] + ) data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) config = { 'prediction_tags': {'tag1': {'data_type': 'float'}}, 'confidence_tags': {'tag2': {'data_type': 'float'}}, } - output_data, opc_metrics = opc.manage_output_tags( + output_data, opc_metrics, session_bad, opc_status, reconnect = opc.manage_output_tags( server_id='server1', config=config, data=data, @@ -286,86 +492,57 @@ def test_manage_output_tags_success(opc): assert output_data is True assert opc_metrics == {'tag1': 0.1, 'tag2': 0.1} - opc.write_data.assert_has_calls( - [ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'], - ), - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'], - ), + assert session_bad is False + assert opc_status is None + assert reconnect is False + assert opc._write_tags_from_config.call_count == 2 + + +def test_manage_output_tags_failed(opc): + opc._write_tags_from_config = MagicMock( + side_effect=[ + ({'tag1': 0.1}, False, None, False), + ({'tag2': None}, False, None, False), ] ) - - -@mark.parametrize('side_effect', [[0.1, None], [None, 0.2]]) -def test_manage_output_tags_failed(opc, side_effect): - opc.write_data = MagicMock(side_effect=side_effect) data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) config = { 'prediction_tags': {'tag1': {'data_type': 'float'}}, 'confidence_tags': {'tag2': {'data_type': 'float'}}, } - output_data, opc_metrics = opc.manage_output_tags( + + output_data, opc_metrics, _, _, _ = opc.manage_output_tags( server_id='server1', config=config, data=data, metadata=metadata['metadata'], ) + assert output_data is False - assert opc_metrics == {'tag1': side_effect[0], 'tag2': side_effect[1]} - opc.write_data.assert_has_calls( - [ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'], - ), - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'], - ), - ] - ) + assert opc_metrics == {'tag1': 0.1, 'tag2': None} def test_manage_output_tags_do_nothing(opc): - opc.write_data = MagicMock(return_value=0.1) + opc._write_tags_from_config = MagicMock() data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) - config = { - '_invalid_key': {'tag1': {'data_type': 'float'}}, - } - output_data, opc_metrics = opc.manage_output_tags( + config = {'_invalid_key': {'tag1': {'data_type': 'float'}}} + + output_data, opc_metrics, _, _, _ = opc.manage_output_tags( server_id='server1', config=config, data=data, metadata=metadata['metadata'], ) + assert output_data is True assert opc_metrics == {} - opc.write_data.assert_not_called() + opc._write_tags_from_config.assert_not_called() @patch('laborious.activities.opc.DataFrame') def test_write_opc_data_success(mock_dataframe, opc): - input_data: dict[str, object] = { + # Arrange + input_data = { **metadata, 'data': {'prediction': [0.75], 'prediction_confidence': [0.95]}, 'opc_output_config': { @@ -375,19 +552,21 @@ def test_write_opc_data_success(mock_dataframe, opc): } }, } - opc_output_config = input_data['opc_output_config'] - assert isinstance(opc_output_config, dict) - opc.manage_output_tags = MagicMock(return_value=(True, {'tag1': 0.1, 'tag2': 0.2})) + # Act + opc.manage_output_tags = MagicMock( + return_value=(True, {'tag1': 0.1, 'tag2': 0.2}, False, None, False) + ) opc.process_confidence = MagicMock(return_value={'data': 'data'}) output_data, opc_metrics = opc.write_opc_data(input_data) + # Assert assert output_data == {'data': 'data'} assert opc_metrics == {'server1': {'tag1': 0.1, 'tag2': 0.2}} opc.manage_output_tags.assert_called_once_with( 'server1', - opc_output_config['server1'], + input_data['opc_output_config']['server1'], mock_dataframe.return_value, metadata['metadata'], ) @@ -395,10 +574,14 @@ def test_write_opc_data_success(mock_dataframe, opc): mock_dataframe.return_value, True, metadata['metadata'], + session_bad=False, + opc_status=None, + reconnect_in_progress=False, ) def test_write_opc_data_empty_config(opc): + # Arrange input_data = { **metadata, 'data': {'prediction': [0.75], 'prediction_confidence': [0.95]}, @@ -406,8 +589,10 @@ def test_write_opc_data_empty_config(opc): 'opc_output_config': {'server1': {'prediction_tags': {}, 'confidence_tags': {}}}, } + # Act opc.write_opc_data(input_data) + # Assert opc.opc_repository['server1'].write_data.assert_not_called() @@ -424,8 +609,10 @@ def test_write_opc_data_no_validate_server(opc): }, } + # Act opc.write_opc_data(input_data) + # Assert opc.opc_repository['server1'].write_data.assert_not_called() @@ -437,18 +624,97 @@ def test_write_opc_data_no_validate_server(opc): ], ) def test_process_confidence(opc, data, success, expected): - result = opc.process_confidence(data, success, metadata) - + result = opc.process_confidence(data, success, metadata['metadata']) assert result['prediction_confidence'][0] == expected +def test_process_confidence_session_bad(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence( + data, + False, + metadata['metadata'], + session_bad=True, + opc_status='BadSessionIdInvalid', + ) + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0].startswith(OPC_SESSION_BAD_COMMENT_PREFIX) + assert 'BadSessionIdInvalid' in result['comments'][0] + + +def test_process_confidence_generic_failure(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence(data, False, metadata['metadata']) + assert result['prediction_confidence'][0] == OPC_WRITTING_ERROR_CONFIDENCE + assert result['comments'][0] == OPC_WRITTING_ERROR_MESSAGE + + +def test_manage_output_tags_merges_error_flags(opc): + opc._write_tags_from_config = MagicMock( + side_effect=[ + ({'tag1': None}, True, 'BadSessionIdInvalid', False), + ({'tag2': 0.2}, False, None, True), + ] + ) + data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]}) + config = { + 'prediction_tags': {'tag1': {'data_type': 'float'}}, + 'confidence_tags': {'tag2': {'data_type': 'float'}}, + } + + ( + success, + metrics, + session_bad_seen, + opc_status, + reconnect_in_progress, + ) = opc.manage_output_tags('server1', config, data, metadata['metadata']) + + assert success is False + assert session_bad_seen is True + assert reconnect_in_progress is True + assert opc_status == 'BadSessionIdInvalid' + assert metrics == {'tag1': None, 'tag2': 0.2} + + +def test_process_confidence_reconnect_in_progress(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + result = opc.process_confidence( + data, + False, + metadata['metadata'], + reconnect_in_progress=True, + ) + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0] == OPC_RECONNECT_IN_PROGRESS_COMMENT + + +def test_process_confidence_concatenates_multiple_comments(opc): + data = DataFrame({'prediction_confidence': [0.9]}) + session_comment = f'{OPC_SESSION_BAD_COMMENT_PREFIX} BadSessionIdInvalid' + + result = opc.process_confidence( + data, + False, + metadata['metadata'], + session_bad=True, + opc_status='BadSessionIdInvalid', + reconnect_in_progress=True, + ) + + assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE + assert result['comments'][0] == OPC_COMMENT_SEPARATOR.join( + [session_comment, OPC_RECONNECT_IN_PROGRESS_COMMENT] + ) + + def test_validate_server(opc): assert opc.validate_server('server1', metadata) is True assert opc.validate_server('server2', metadata) is False def test_close(opc): - disconnect_mock = MagicMock(return_value=None) - opc.opc_repository['server1'].disconnect = disconnect_mock + repo = opc.opc_repository['server1'] + repo.disconnect = MagicMock(return_value=True) opc.close() - disconnect_mock.assert_called_once() + repo.disconnect.assert_called_once() diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 6c88c99..6b8b57a 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -1,12 +1,20 @@ +import concurrent.futures import json from datetime import datetime from unittest.mock import ANY, MagicMock, Mock, patch import pytest from asyncua.crypto import security_policies +from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid from sientia_do.notifications.models import NotificationLevel -from laborious.utils.repository.opc_repository import OpcRepository +from laborious.utils.repository.opc_repository import ( + OpcClientAlreadyExistsError, + OpcClientNotInitializedError, + OpcRepository, + OpcSessionAlreadyConnectedError, + is_reconnectable_opcua_bad, +) @pytest.fixture @@ -31,7 +39,13 @@ def opc_repository(mock_logger): ) repository.disconnection_interval = 0.1 repository.send_notification = MagicMock() + repository.send_notification = MagicMock() repository.emit_metric_sync = MagicMock() + repository.info = MagicMock() + repository.error = MagicMock() + repository.warning = MagicMock() + repository.debug = MagicMock() + repository._session_ready.set() return repository @@ -39,7 +53,12 @@ def opc_repository(mock_logger): def mock_client(): with patch('laborious.utils.repository.opc_repository.Client') as mock: client_instance = MagicMock() - client_instance.aio_obj = MagicMock() + aio = MagicMock() + client_instance.aio_obj = aio + aio.uaclient = MagicMock() + aio.uaclient.protocol = MagicMock(state='closed') + aio.session_timeout = 600_000 + aio.secure_channel_timeout = 600_000 mock.return_value = client_instance yield client_instance @@ -65,13 +84,13 @@ def test_init(opc_repository): assert opc_repository.reconnection_interval == 60 assert opc_repository.client is None assert opc_repository.last_reconnection_time is None - assert opc_repository.error_count == 0 def test_set_security(opc_repository, mock_client): opc_repository.client = mock_client opc_repository.set_security() + mock_client.application_uri = 'urn:test:server' mock_client.set_security.assert_called_once_with( security_policies.SecurityPolicyBasic256, '/path/to/cert.pem', @@ -79,64 +98,115 @@ def test_set_security(opc_repository, mock_client): None, '/path/to/server_cert.pem', ) - assert mock_client.aio_obj.secure_channel_timeout == 10000000 - assert mock_client.aio_obj.session_timeout == 10000000 + assert mock_client.aio_obj.secure_channel_timeout == 600_000 + assert mock_client.aio_obj.session_timeout == 600_000 def test_set_security_missing_certificates(opc_repository): opc_repository.cert_path = None opc_repository.private_key_path = None - with pytest.raises(ValueError, match='Certificate and private key paths'): + try: opc_repository.set_security() + except ValueError as e: + assert str(e) == 'Certificate and private key paths must be provided for secure connection.' def test_set_security_missing_client(opc_repository): opc_repository.client = None - with pytest.raises(ValueError, match='Client must be initialized'): + try: opc_repository.set_security() + except ValueError as e: + assert str(e) == 'Client must be initialized before setting security' def test_connect_with_security(opc_repository, mock_client): - opc_repository.try_connect = MagicMock(return_value=(True, {})) + opc_repository._create_client = MagicMock() + opc_repository._open_session = MagicMock(return_value=(True, {})) result = opc_repository.connect() - opc_repository.try_connect.assert_called_once() - assert opc_repository.client == mock_client + opc_repository._create_client.assert_called_once() + opc_repository._open_session.assert_called_once() assert result == (True, {}) def test_connect_without_security(opc_repository, mock_client): opc_repository.cert_path = None - opc_repository.try_connect = MagicMock(return_value=(True, {})) + opc_repository._create_client = MagicMock() + opc_repository._open_session = MagicMock(return_value=(True, {})) opc_repository.set_security = MagicMock() result = opc_repository.connect() - opc_repository.try_connect.assert_called_once() + opc_repository._create_client.assert_called_once() + opc_repository._open_session.assert_called_once() opc_repository.set_security.assert_not_called() - assert opc_repository.client == mock_client assert result == (True, {}) -def test_try_connect_success(opc_repository): - opc_repository.last_reconnection_time = None +def test_connect_raises_when_session_already_open(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'open' + mock_client.aio_obj.uaclient.protocol = proto + + with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'): + opc_repository.connect() + + +def test_create_client_raises_when_client_exists(opc_repository, mock_client): + opc_repository.client = mock_client + + with pytest.raises(OpcClientAlreadyExistsError, match='already exists'): + opc_repository._create_client() + + +def test_open_session_success(opc_repository): + closed_proto = MagicMock() + closed_proto.state = 'closed' opc_repository.client = MagicMock() - result = opc_repository.try_connect() + aio = MagicMock() + opc_repository.client.aio_obj = aio + aio.uaclient = MagicMock(protocol=closed_proto) + aio.session_timeout = 600_000 + aio.secure_channel_timeout = 600_000 + + open_proto = MagicMock() + open_proto.state = 'open' + open_proto.authentication_token = 'tok' + + def connect_side_effect(): + aio.uaclient.protocol = open_proto + + opc_repository.client.connect = MagicMock(side_effect=connect_side_effect) + + result = opc_repository._open_session() opc_repository.client.connect.assert_called_once() - assert opc_repository.last_reconnection_time is not None assert result == (True, {}) + assert opc_repository._session_ready.is_set() -def test_try_connect_fail(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.disconnect = MagicMock() +def test_open_session_raises_when_already_connected(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'open' + mock_client.aio_obj.uaclient.protocol = proto + + with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'): + opc_repository._open_session() + + +def test_open_session_fail(opc_repository): + opc_repository._disconnect_locked = MagicMock() opc_repository.client = MagicMock() - opc_repository.client.connect.side_effect = Exception('Test error') + aio = MagicMock() + opc_repository.client.aio_obj = aio + aio.uaclient = MagicMock(protocol=MagicMock(state='closed')) + opc_repository.client.connect = MagicMock(side_effect=Exception('Test error')) - is_connected, error_data = opc_repository.try_connect() + is_connected, error_data = opc_repository._open_session() - opc_repository.disconnect.assert_called_once() + opc_repository._disconnect_locked.assert_called_once() opc_repository.client.connect.assert_called_once() assert is_connected is False assert error_data['notification_id'] == f'OPC_CONNECTION_ERROR_{opc_repository.id}' @@ -146,29 +216,17 @@ def test_try_connect_fail(opc_repository): assert error_data['attachment_content'] is not None -def test_try_connect_no_client(opc_repository): +def test_open_session_raises_when_no_client(opc_repository): opc_repository.client = None - result = opc_repository.try_connect() - assert result == ( - False, - { - 'notification_id': f'OPC_CONNECTION_ERROR_{opc_repository.id}', - 'message': 'Client is not initialized', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - }, - ) - -def test_session_alive_returns_false_when_client_none(opc_repository): - opc_repository.client = None - assert opc_repository._session_alive() is False + with pytest.raises(OpcClientNotInitializedError, match='not initialized'): + opc_repository._open_session() def test_disconnection_fallback_success(opc_repository, mock_client): opc_repository.client = mock_client - mock_client.disconnect.return_value = None - result = opc_repository.disconnection_fallback() + mock_client.disconnect.return_value = True + result = opc_repository._disconnection_fallback() mock_client.disconnect.assert_called_once() assert result == [] @@ -177,7 +235,7 @@ def test_disconnection_fallback_success(opc_repository, mock_client): def test_disconnection_fallback_fail(opc_repository, mock_client): opc_repository.client = mock_client mock_client.disconnect.side_effect = Exception('Test error') - result = opc_repository.disconnection_fallback() + result = opc_repository._disconnection_fallback() assert result == [ {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, @@ -190,10 +248,10 @@ def test_disconnection_fallback_fail(opc_repository, mock_client): def test_disconnect(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.disconnection_fallback = MagicMock(return_value=[]) + opc_repository._disconnection_fallback = MagicMock(return_value=[]) opc_repository.disconnect() - opc_repository.disconnection_fallback.assert_called_once() + opc_repository._disconnection_fallback.assert_called_once() assert opc_repository.client is None @@ -204,12 +262,12 @@ def test_disconnect_no_client(opc_repository): def test_disconnect_error(opc_repository, mock_client): opc_repository.client = mock_client - opc_repository.disconnection_fallback = MagicMock( + opc_repository._disconnection_fallback = MagicMock( return_value=[{'attempt': 1, 'error': 'Test error', 'traceback': 'text'}] ) opc_repository.disconnect() - opc_repository.disconnection_fallback.assert_called_once() + opc_repository._disconnection_fallback.assert_called_once() opc_repository.send_notification.assert_called_once_with( metadata=opc_repository.metadata, notification_id=f'OPC_DISCONNECTION_ERROR_{opc_repository.id}', @@ -225,67 +283,25 @@ def test_disconnect_error(opc_repository, mock_client): def test_validate_connection_none_client(opc_repository): opc_repository.client = None - opc_repository.connect = MagicMock(return_value=(True, {})) response = opc_repository.validate_connection() - assert response == (True, {}) - opc_repository.connect.assert_called_once() + assert response == (False, opc_repository._not_connected_error()) -def test_validate_connection_disconnect_raises(opc_repository): - """Outer except path when reconnect cleanup fails mid-validation.""" - +def test_validate_connection_session_not_open(opc_repository): opc_repository.client = MagicMock() - opc_repository._session_alive = MagicMock(return_value=False) - opc_repository.last_reconnection_time = datetime(2020, 1, 1, 0, 0, 0) - opc_repository.disconnect = MagicMock(side_effect=RuntimeError('disconnect failed')) + opc_repository.client.aio_obj.uaclient.protocol = None response = opc_repository.validate_connection() - assert response[0] is False - assert response[1]['notification_id'] == f'OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}' - assert 'disconnect failed' in response[1]['message'] - - -@patch('laborious.utils.repository.opc_repository.datetime') -def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository): - _mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 0, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = MagicMock() - opc_repository.client.get_root_node.side_effect = RuntimeError('down') - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock(return_value=(True, {})) - - response = opc_repository.validate_connection() - opc_repository.connect.assert_not_called() - assert response == ( - False, - { - 'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}', - 'message': f'OPC server {opc_repository.id} is not connected, waiting for next reconnection window...', - 'block': 'opc_repository', - 'level': NotificationLevel.WARNING, - }, - ) - - -@patch('laborious.utils.repository.opc_repository.datetime') -def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository): - mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 1, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = MagicMock() - opc_repository.client.get_root_node.side_effect = RuntimeError('down') - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock(return_value=(True, {})) - - response = opc_repository.validate_connection() - opc_repository.connect.assert_called_once() - assert response == opc_repository.connect.return_value + assert response == (False, opc_repository._not_connected_error()) + opc_repository.error.assert_called_once() def test_validate_connection_success(opc_repository): opc_repository.client = MagicMock() - opc_repository.error_count = 0 - opc_repository.client.get_root_node.return_value = MagicMock() + proto = MagicMock() + proto.state = 'open' + opc_repository.client.aio_obj.uaclient.protocol = proto output = opc_repository.validate_connection() assert output == (True, {}) @@ -293,28 +309,22 @@ def test_validate_connection_success(opc_repository): def test_write_data_validate_connection_do_nothing(opc_repository): opc_repository.validate_connection = MagicMock(return_value=(True, {})) - opc_repository.client = MagicMock() + opc_repository.client = MagicMock(get_node=MagicMock()) mock_node = MagicMock() opc_repository.client.get_node.return_value = mock_node - result = opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] - ) + result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with('ns=2;s=TestNode') - mock_node.set_value.assert_called_once() assert result == (True, {'response_time': ANY}) def test_write_data_validate_connection_failed(opc_repository): opc_repository.validate_connection = MagicMock(return_value=(False, {})) opc_repository.client = MagicMock() - opc_repository.error_count = 0 - result = opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] - ) + result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_not_called() @@ -324,7 +334,6 @@ def test_write_data_validate_connection_failed(opc_repository): def test_write_data_get_node_failed(opc_repository): opc_repository.validate_connection = MagicMock(return_value=(True, {})) opc_repository.client = MagicMock() - opc_repository.error_count = 0 opc_repository.client.get_node = MagicMock(side_effect=Exception('Test error')) is_success, error_data = opc_repository.write_data( @@ -335,6 +344,13 @@ def test_write_data_get_node_failed(opc_repository): opc_repository.client.get_node.assert_called_once_with('ns=2;s=TestNode') assert is_success is False assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}' + assert ( + error_data['message'] + == "Failed to get node from OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" + ) + assert error_data['block'] == 'opc_repository' + assert error_data['level'] == NotificationLevel.ERROR + assert error_data['attachment_content'] is not None def test_write_data_invalid_data_type(opc_repository, mock_client): @@ -352,6 +368,13 @@ def test_write_data_invalid_data_type(opc_repository, mock_client): assert is_success is False assert error_data['notification_id'] == f'OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}' + assert ( + error_data['message'] + == "Unsupported data type: invalid_type | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" + ) + assert error_data['block'] == 'opc_repository' + assert error_data['level'] == NotificationLevel.ERROR + assert error_data.get('attachment_content') is None def test_write_data(opc_repository, mock_client): @@ -360,12 +383,10 @@ def test_write_data(opc_repository, mock_client): mock_node = MagicMock() mock_client.get_node = MagicMock(return_value=mock_node) - result = opc_repository.write_data( - 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] - ) + result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) mock_client.get_node.assert_called_once_with('ns=2;s=TestNode') - mock_node.set_value.assert_called_once() + mock_node.write_value.assert_called_once() assert result == (True, {'response_time': ANY}) @@ -373,9 +394,8 @@ def test_write_data_write_value_failed(opc_repository, mock_client): opc_repository.validate_connection = MagicMock(return_value=(True, {})) opc_repository.client = mock_client mock_node = MagicMock() - opc_repository.error_count = 0 mock_client.get_node = MagicMock(return_value=mock_node) - mock_node.set_value.side_effect = Exception('Test error') + mock_node.write_value.side_effect = Exception('Test error') is_success, error_data = opc_repository.write_data( 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] @@ -383,6 +403,106 @@ def test_write_data_write_value_failed(opc_repository, mock_client): opc_repository.validate_connection.assert_called_once() mock_client.get_node.assert_called_once_with('ns=2;s=TestNode') - mock_node.set_value.assert_called_once() + mock_node.write_value.assert_called_once() assert is_success is False assert error_data['notification_id'] == f'OPC_WRITE_DATA_ERROR_{opc_repository.id}' + assert ( + error_data['message'] + == "Failed to write data to OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" + ) + assert error_data['block'] == 'opc_repository' + assert error_data['level'] == NotificationLevel.ERROR + assert error_data['attachment_content'] is not None + + +def test_is_reconnectable_opcua_bad(): + assert is_reconnectable_opcua_bad(BadSessionIdInvalid()) is True + assert is_reconnectable_opcua_bad(BadNodeIdUnknown()) is False + assert is_reconnectable_opcua_bad(Exception('other')) is False + + +def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, mock_client): + opc_repository.validate_connection = MagicMock(return_value=(True, {})) + opc_repository.client = mock_client + opc_repository._start_reconnect_on_bad = MagicMock() + mock_node = MagicMock() + mock_client.get_node = MagicMock(return_value=mock_node) + mock_node.write_value.side_effect = BadSessionIdInvalid() + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + mock_node.write_value.assert_called_once() + opc_repository._start_reconnect_on_bad.assert_called_once() + assert is_success is False + assert error_data['opc_error_kind'] == 'session_bad' + assert error_data['opc_status'] == 'BadSessionIdInvalid' + + +def test_write_data_reconnect_in_progress_immediate(opc_repository): + opc_repository._session_ready.clear() + opc_repository._reconnect_thread = MagicMock() + opc_repository._reconnect_thread.is_alive.return_value = True + opc_repository.validate_connection = MagicMock() + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + opc_repository.validate_connection.assert_not_called() + assert is_success is False + assert error_data['opc_error_kind'] == 'reconnect_in_progress' + + +def test_start_reconnect_on_bad_skips_within_interval(opc_repository): + opc_repository.last_reconnection_time = datetime.now() + opc_repository.reconnection_interval = 3600 + + opc_repository._start_reconnect_on_bad('BadSessionIdInvalid', 'tok') + + assert opc_repository._reconnect_thread is None + + +def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client): + opc_repository.validate_connection = MagicMock(return_value=(True, {})) + opc_repository.client = mock_client + opc_repository.reconnection_interval = 0 + opc_repository.last_reconnection_time = None + mock_node = MagicMock() + mock_client.get_node = MagicMock(return_value=mock_node) + mock_node.write_value.side_effect = BadSessionIdInvalid() + opc_repository._start_reconnect_on_bad = MagicMock() + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit( + opc_repository.write_data, + node, + value, + 'float', + metadata['metadata'], + ) + for node, value in (('ns=2;s=TestNode', 1.0), ('ns=2;s=TestNode2', 2.0)) + ] + results = [future.result() for future in futures] + + assert 1 <= opc_repository._start_reconnect_on_bad.call_count <= 2 + assert mock_node.write_value.call_count == 2 + error_kinds = [r[1].get('opc_error_kind') for r in results] + assert error_kinds.count('session_bad') >= 1 + assert all(k in ('session_bad', 'reconnect_in_progress') for k in error_kinds) + + +@patch('laborious.utils.repository.opc_repository.datetime') +def test_reconnect_locked_sets_last_reconnection_time(mock_datetime, opc_repository): + mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 12, 0, 0)) + opc_repository._disconnect_locked = MagicMock() + opc_repository._connect_locked = MagicMock(return_value=(True, {})) + + result = opc_repository._reconnect_locked() + + opc_repository._disconnect_locked.assert_called_once() + opc_repository._connect_locked.assert_called_once() + assert result == (True, {}) + assert opc_repository.last_reconnection_time == datetime(2025, 1, 1, 12, 0, 0) diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index 9583048..dc09958 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -10,8 +10,7 @@ from laborious.utils.connectors_config import ( def test_build_mlflow_config_with_env_vars(): # Arrange - environ['MLFLOW_HOST'] = 'http://test-host' - environ['MLFLOW_PORT'] = '8080' + environ['MLFLOW_URL'] = 'http://test-host:8080' environ['MLFLOW_USERNAME'] = 'test-user' environ['MLFLOW_PASSWORD'] = 'test-pass' @@ -25,8 +24,7 @@ def test_build_mlflow_config_with_env_vars(): def test_build_mlflow_config_host_already_has_port(): - environ['MLFLOW_HOST'] = 'http://tracker.example.com:443' - environ['MLFLOW_PORT'] = '8080' + environ['MLFLOW_URL'] = 'http://tracker.example.com:443' environ['MLFLOW_USERNAME'] = 'u' environ['MLFLOW_PASSWORD'] = 'p' @@ -38,8 +36,7 @@ def test_build_mlflow_config_host_already_has_port(): def test_build_mlflow_config_with_defaults(): # Arrange # Clear any existing env vars - environ.pop('MLFLOW_HOST', None) - environ.pop('MLFLOW_PORT', None) + environ.pop('MLFLOW_URL', None) environ.pop('MLFLOW_USERNAME', None) environ.pop('MLFLOW_PASSWORD', None)