diff --git a/.gitignore b/.gitignore index c27efb2..525d66e 100644 --- a/.gitignore +++ b/.gitignore @@ -55,4 +55,4 @@ mlruns/ relatorio* openspec/* -.cursor/* \ No newline at end of file +.cursor/* diff --git a/README.md b/README.md index c864f7c..d20fcd2 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc - [Prediction Operation Metrics](#prediction-operation-metrics) - [OPC Export Metrics](#opc-export-metrics) - [Data Quality Metrics](#data-quality-metrics) +- [OPC UA Communication](#opc-ua-communication) - [Configuration](#configuration-1) - [Environment Variables](#environment-variables) - [OPC Configuration](#opc-configuration) @@ -164,7 +165,7 @@ Laborious uses a Temporal-based architecture with strong separation of concerns - `connectors_config.py`: Env-driven configuration builders - `models/minio_dataframe_payload.py`: MinIO-offloaded DataFrame payload model - ML models are loaded via `SientiaMLflowRepository` (wrapper-based, `@production` alias) constructed in `Activities` from `build_mlflow_config()`. -- `repository/opc_repository.py`: OPC communication and writes +- `repository/opc_repository.py`: OPC UA client, writes, session recovery (see [OPC UA Communication](#opc-ua-communication)) - `repository/minio_manager.py`: MinIO object storage operations - `filters/conditional_filters.py` and `filters/mlflow_filters.py` @@ -778,6 +779,15 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi - Labels: `pod_id`, `model_name`, `workflow_name`, `opc_server_id` - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] +OPC UA session and write diagnostics (Prometheus, `laborious/metrics.py`): + +- `opc_connections_initiated_total`, `opc_connections_failed_total`, `opc_connection_status` +- `opc_session_created_total`, `opc_session_closed_total`, `opc_session_revised_timeout_milliseconds` +- `opc_write_attempts_total` (label `result`: `OK` or exception name, e.g. `BadSessionIdInvalid`) +- `opc_write_inter_arrival_over_session_timeout_total` + +See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, and confidence codes **12** / **14**. + ### Data Quality Metrics - Filter pass/fail rates through notification system - MLFlow API response validation metrics @@ -822,7 +832,7 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi | `OPC_CERT_PATH` | OPC client certificate path | `None` | No | | `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | | `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | -| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | +| `OPC_RECONNECTION_INTERVAL` | Minimum seconds between OPC reconnects | `120` | No | | `PI_WEB_API_BASE_URL` | PI Web API server base URL | `None` | No | | `PI_WEB_API_AUTH_TYPE` | PI Web API authentication type (basic/bearer) | `None` | No | | `PI_WEB_API_AUTH_TOKEN` | PI Web API authentication token | `None` | No | @@ -915,6 +925,12 @@ Legacy MinIO object layout (relative key): `training_datasets/{model_name}/{object_prefix}_{timestamp}.parquet` where `object_prefix` is sanitized (slashes replaced by underscores) to keep a stable model-level directory. +## OPC UA Communication + +Full reference: **[docs/opc-communication.md](docs/opc-communication.md)** (connection lifecycle, Tier-1 `Bad*` reconnect, connection lock / session readiness, metrics, PostgreSQL confidence **12** vs **14**, tests). + +Implementation plan: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). + ### OPC Configuration For multiple OPC servers, use the `OPC_CONFIG` environment variable: @@ -1180,8 +1196,10 @@ laborious/ - Ensure proper connection pool configuration 4. **OPC Connection Failures** - - Verify OPC server is accessible + - See [docs/opc-communication.md](docs/opc-communication.md) + - Verify OPC server is accessible and `OPC_RECONNECTION_INTERVAL` is appropriate - Check certificate and key file paths + - Correlate `opc_write_attempts_total` with `opc_session_*` metrics; count session errors via `prediction_confidence = 14` - Review OPC server logs for connection issues 5. **PI Web API Connection Failures** diff --git a/docs/opc-communication.md b/docs/opc-communication.md index 2c51c75..d272c61 100644 --- a/docs/opc-communication.md +++ b/docs/opc-communication.md @@ -12,7 +12,7 @@ Worker (long-lived) ├── 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* + └── background reconnect on Tier-1 Bad*, closed protocol, or stale session Temporal activity write_opc_data └── OPC.manage_output_tags → write_data per tag (sequential per activity) @@ -26,9 +26,9 @@ One worker process holds one `OpcRepository` instance per configured server. Mul |-------|----------| | 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 | +| Tier-1 Bad* / protocol closed / session not ready | `_start_reconnect(reason)` → `_run_reconnect` (thread) → `_reconnect_locked()` (respects `reconnection_interval`) | +| Write | `write_data()` checks in-flight reconnect thread, `_session_ready`, validates, then one `get_node` + `write_value` | +| Shutdown | `disconnect()` sets `_allow_reconnect = False`, then tears down session | ### Session and channel timeouts @@ -44,8 +44,9 @@ To allow **multiple concurrent writes** when the session is healthy, but **block | 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_lock` (`threading.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. | +| `_session_ready` (`threading.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. | +| `_allow_reconnect` | Cleared in `disconnect()` so shutdown does not spawn reconnect threads | **Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):** @@ -61,21 +62,20 @@ Public `connect()` / `disconnect()` acquire the lock and call `_connect_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). +1. If a reconnect **thread** is alive → `reconnect_in_progress`. +2. If `_session_ready` is cleared → schedule `SessionNotReady` reconnect; return `reconnect_in_progress` or `connection_lost`. +3. If `validate_connection()` fails (protocol closed) → schedule `ProtocolClosed` reconnect; return `connection_lost`. +4. Single `get_node` + `write_value` (no retry in the same call). -**Reconnect path (`_run_reconnect_on_bad`):** +**Reconnect path (`_run_reconnect`):** -1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows. -2. `async with _connection_lock:` → `_reconnect_locked()`. +1. `_start_reconnect` clears `_session_ready` and starts a daemon thread when the interval allows and `_allow_reconnect` is true. +2. `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. +**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes while keeping the connection lock semantics above. ## Tier-1 `Bad*` errors and reconnect diff --git a/inter_arrival.py b/inter_arrival.py new file mode 100644 index 0000000..bce9cee --- /dev/null +++ b/inter_arrival.py @@ -0,0 +1,25 @@ +# %% + +# Load logs.txt +with open('logs.txt', 'r') as file: + lines = file.readlines() + +# %% +import re +# Grep "inter-arrival_s=number" with regex +intervals = [] +for line in lines: + match = re.search(r'inter-arrival_s=([0-9.]+)', line) + if match: + intervals.append(float(match.group(1))) +# %% + +print(intervals) +# %% +import matplotlib.pyplot as plt +plt.plot(intervals) +plt.ylabel('Inter-arrival time (s)') +plt.xlabel('Sample') +plt.title('Inter-arrival time distribution') +plt.show() +# %% diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 64854ff..6a0b172 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -211,6 +211,7 @@ class OpcRepository(SientiaMonitoring): self._connection_lock = threading.Lock() self._session_ready = threading.Event() self._reconnect_thread: threading.Thread | None = None + self._allow_reconnect = True def _opc_debug_tags(self, session_id: str) -> dict[str, str]: return { @@ -513,8 +514,12 @@ class OpcRepository(SientiaMonitoring): def disconnect(self) -> None: """ Gracefully disconnect from the OPC server under the connection lock. + + Disables background reconnect so late writes during worker shutdown do not + respawn sessions. """ with self._connection_lock: + self._allow_reconnect = False self._disconnect_locked() def validate_connection(self) -> tuple[bool, dict[str, Any]]: @@ -531,61 +536,82 @@ class OpcRepository(SientiaMonitoring): 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: + def _reconnect_thread_in_progress(self) -> bool: """ - Schedule a background reconnect if interval and thread state allow it. + Return whether a background reconnect thread is currently running. + + Return: + bool: True when a reconnect thread exists and is alive. + """ + return self._reconnect_thread is not None and self._reconnect_thread.is_alive() + + def _start_reconnect(self, reason: str, session_id: str) -> None: + """ + Schedule a background reconnect when allowed by interval and thread state. + + Clears _session_ready before starting the thread. No-op when _allow_reconnect is + False, the reconnection window has not elapsed, or a reconnect is already running. Args: - opc_status (str): OPC UA status name that triggered reconnect. + reason (str): Trigger for reconnect (OPC status name or synthetic reason). session_id (str): Session token before failure. """ + if not self._allow_reconnect: + return 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}', + f'reconnect_reason={reason}', self.metadata, ) return - if self._reconnect_thread is not None and self._reconnect_thread.is_alive(): + if self._reconnect_thread_in_progress(): self.warning( f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} ' - f'opc_status={opc_status}', + f'reconnect_reason={reason}', self.metadata, ) return self._session_ready.clear() self.info( - f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} ' + f'OPC reconnect scheduled reconnect_reason={reason} 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), + target=self._run_reconnect, + args=(reason, session_id), daemon=True, ) self._reconnect_thread.start() - def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + def _run_reconnect(self, reason: 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. + reason (str): Trigger for reconnect (OPC status or ProtocolClosed). 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'OPC reconnect started reconnect_reason={reason} opc_server_id={self.id} ' f'old_session_id={session_id}', self.metadata, ) - self._reconnect_locked() + success, error = self._reconnect_locked() + if not success: + self.error( + f'OPC reconnect failed reconnect_reason={reason} opc_server_id={self.id}', + self.metadata, + ) + if error: + self.error(error.get('message', ''), self.metadata) except Exception: self.error( - f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}', + f'OPC reconnect task failed opc_server_id={self.id} reconnect_reason={reason}', self.metadata, ) self.error(traceback.format_exc(), self.metadata) @@ -681,7 +707,7 @@ class OpcRepository(SientiaMonitoring): f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}', metadata, ) - self._start_reconnect_on_bad(opc_status, session_id) + self._start_reconnect(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}', @@ -714,32 +740,89 @@ class OpcRepository(SientiaMonitoring): pass node_obj.set_value(data, variant_type) + def _write_reconnect_in_progress( + self, metadata: dict[str, Any] + ) -> tuple[bool, dict[str, Any]]: + """ + Fail a write because a background reconnect thread is already running. + + Args: + metadata (dict[str, Any]): Write context passed through to the activity. + + Return: + tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind reconnect_in_progress). + """ + 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', + } + + def _write_connection_lost( + self, metadata: dict[str, Any], opc_status: str + ) -> tuple[bool, dict[str, Any]]: + """ + Fail a write after scheduling reconnect for a closed or stale session. + + Args: + metadata (dict[str, Any]): Write context passed through to the activity. + opc_status (str): Synthetic reason (ProtocolClosed, SessionNotReady). + + Return: + tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind connection_lost). + """ + self._emit_opc_write_metric('unknown', opc_status, metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_CONNECTION_LOST_{self.id}', + message=f'OPC write skipped: connection lost ({opc_status}) | metadata: {metadata}', + level=NotificationLevel.WARNING, + opc_error_kind='connection_lost', + opc_status=opc_status, + ) + def write_data( self, node: str, value: Any, data_type: str, metadata: dict[str, Any] ) -> tuple[bool, dict[str, Any]]: """ - 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', - } + Write data to OPC server with a single attempt and background reconnect scheduling. - is_connected, error = self.validate_connection() + Reconnect is scheduled on Tier-1 Bad*, closed protocol, or stale session readiness. + There is no retry within the same call. + + Args: + node (str): OPC UA node id to write. + value (Any): Value to convert and send. + data_type (str): Logical type key (float, int, bool, str, double). + metadata (dict[str, Any]): Activity context (model_id, model_name, etc.). + + Return: + tuple[bool, dict[str, Any]]: (True, {response_time}) on success, or + (False, structured error info) on failure. + """ + if self._reconnect_thread_in_progress(): + return self._write_reconnect_in_progress(metadata) + + if not self._session_ready.is_set(): + session_id = _opc_authentication_token_str(self.client) + self._start_reconnect('SessionNotReady', session_id) + if self._reconnect_thread_in_progress(): + return self._write_reconnect_in_progress(metadata) + return self._write_connection_lost(metadata, 'SessionNotReady') + + is_connected, _error = self.validate_connection() if not is_connected: - self._emit_opc_write_metric('unknown', 'NotConnected', metadata) - return False, error + session_id = _opc_authentication_token_str(self.client) + self._start_reconnect('ProtocolClosed', session_id) + return self._write_connection_lost(metadata, 'ProtocolClosed') start_time = time.time() session_id = _opc_authentication_token_str(self.client) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 2597864..4b47d5b 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -65,7 +65,7 @@ with workflow.unsafe.imports_passed_through(): ) from laborious.workflows.sub_workflows.prediction_process import PredictionProcess -POD_ID = os.getenv('POD_ID') +POD_ID = os.getenv('HOSTNAME') SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) diff --git a/requirements-dev.txt b/requirements-dev.txt index c4f4ef6..4d0f89d 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -18,3 +18,4 @@ testcontainers[postgres,minio] # PostgreSQL and MinIO containers for E2E tests # Development Tools ipython>=8.12.0 # Enhanced Python shell ipdb>=0.13.13 # IPython debugger +ipykernel==6.30.1 # IPython kernel for Jupyter notebooks diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 6b8b57a..b5ef20d 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -253,6 +253,7 @@ def test_disconnect(opc_repository, mock_client): opc_repository._disconnection_fallback.assert_called_once() assert opc_repository.client is None + assert opc_repository._allow_reconnect is False def test_disconnect_no_client(opc_repository): @@ -323,12 +324,19 @@ def test_write_data_validate_connection_do_nothing(opc_repository): def test_write_data_validate_connection_failed(opc_repository): opc_repository.validate_connection = MagicMock(return_value=(False, {})) opc_repository.client = MagicMock() + opc_repository._start_reconnect = MagicMock() - result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + is_success, error_data = 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() - assert result == (False, {}) + opc_repository._start_reconnect.assert_called_once() + assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed' + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + assert error_data['opc_status'] == 'ProtocolClosed' def test_write_data_get_node_failed(opc_repository): @@ -424,7 +432,7 @@ def test_is_reconnectable_opcua_bad(): 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() + opc_repository._start_reconnect = MagicMock() mock_node = MagicMock() mock_client.get_node = MagicMock(return_value=mock_node) mock_node.write_value.side_effect = BadSessionIdInvalid() @@ -434,7 +442,7 @@ def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, m ) mock_node.write_value.assert_called_once() - opc_repository._start_reconnect_on_bad.assert_called_once() + opc_repository._start_reconnect.assert_called_once() assert is_success is False assert error_data['opc_error_kind'] == 'session_bad' assert error_data['opc_status'] == 'BadSessionIdInvalid' @@ -455,15 +463,83 @@ def test_write_data_reconnect_in_progress_immediate(opc_repository): assert error_data['opc_error_kind'] == 'reconnect_in_progress' -def test_start_reconnect_on_bad_skips_within_interval(opc_repository): +def test_start_reconnect_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') + opc_repository._start_reconnect('BadSessionIdInvalid', 'tok') assert opc_repository._reconnect_thread is None +def test_write_data_protocol_closed_schedules_reconnect(opc_repository): + opc_repository.client = MagicMock() + opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed') + opc_repository._start_reconnect = MagicMock() + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + opc_repository._start_reconnect.assert_called_once() + assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed' + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + assert error_data['opc_status'] == 'ProtocolClosed' + + +def test_write_data_protocol_closed_skips_reconnect_within_interval(opc_repository): + opc_repository.client = MagicMock() + opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed') + opc_repository.last_reconnection_time = datetime.now() + opc_repository.reconnection_interval = 3600 + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + assert opc_repository._reconnect_thread is None + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + + +def test_write_data_after_failed_reconnect_schedules_again(opc_repository): + opc_repository._session_ready.clear() + opc_repository.reconnection_interval = 0 + opc_repository.last_reconnection_time = None + opc_repository._reconnect_locked = MagicMock( + return_value=(False, {'message': 'connect failed'}) + ) + + opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + if opc_repository._reconnect_thread is not None: + opc_repository._reconnect_thread.join(timeout=2) + assert opc_repository._reconnect_locked.call_count == 1 + assert not opc_repository._reconnect_thread_in_progress() + + opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + if opc_repository._reconnect_thread is not None: + opc_repository._reconnect_thread.join(timeout=2) + assert opc_repository._reconnect_locked.call_count == 2 + + +def test_write_data_after_disconnect_does_not_schedule_reconnect(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'closed' + mock_client.aio_obj.uaclient.protocol = proto + opc_repository._disconnection_fallback = MagicMock(return_value=[]) + opc_repository.disconnect() + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + assert opc_repository._reconnect_thread is None + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + + 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 @@ -472,7 +548,7 @@ def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client): 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() + opc_repository._start_reconnect = MagicMock() with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: futures = [ @@ -487,7 +563,7 @@ def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client): ] results = [future.result() for future in futures] - assert 1 <= opc_repository._start_reconnect_on_bad.call_count <= 2 + assert 1 <= opc_repository._start_reconnect.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