From f323da7b56e709d66def84636f2daa3644931c34 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 25 Sep 2026 12:27:17 -0300 Subject: [PATCH] SIENTIAPDE-1646 Harden retrain merge and OPC write against silent failures - pandas merge now uses validate='one_to_one' so duplicated timestamps from wrapper.retrain fail loudly instead of silently multiplying rows - OPC write_data raises OpcClientNotInitializedError instead of a bare assert when client is torn down between validation and write, so callers get a reported error instead of an AssertionError - add regression test covering the torn-down client path Co-Authored-By: Claude Sonnet 5 --- laborious/activities/mlflow.py | 25 +++++++++++-------- laborious/utils/repository/opc_repository.py | 5 +++- .../utils/repository/test_opc_repository.py | 16 ++++++++++++ 3 files changed, 35 insertions(+), 11 deletions(-) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 91a073d..df812dc 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -529,21 +529,26 @@ class MLFlow(SientiaMonitoring): ) # Keep heavy model fitting outside MLflow run timing. - prediction_data = wrapper.retrain(data) - prediction_data.rename(columns={target: 'prediction'}, inplace=True) + prediction_data = wrapper.retrain(data).rename(columns={target: 'prediction'}) - # Merge prediction data with retrain data + # Merge prediction data with retrain data. Both sides are indexed by the pivoted + # timestamp, so one row per timestamp on each side; `validate` fails loudly instead + # of silently multiplying rows if the wrapper ever returns duplicated timestamps. evaluation_data = pd.merge( - data, prediction_data, left_index=True, right_index=True, how='left' + data, + prediction_data, + left_index=True, + right_index=True, + how='left', + validate='one_to_one', ) - # Rename target column to "target" - evaluation_data.rename(columns={target: 'target'}, inplace=True) - - # Reset index and put as column "timestamp" + # Rename target column to "target", then move the index into a "timestamp" column + evaluation_data = evaluation_data.rename(columns={target: 'target'}) evaluation_data['timestamp'] = evaluation_data.index - evaluation_data.reset_index(drop=True, inplace=True) - evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True) + evaluation_data = evaluation_data.reset_index(drop=True).sort_values( + by='timestamp', ascending=True + ) run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}' diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 7f79583..4b1f688 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -843,7 +843,10 @@ class OpcRepository(SientiaMonitoring): session_id = _opc_authentication_token_str(self.client) try: - assert self.client is not None + if self.client is None: + raise OpcClientNotInitializedError( + f'OPC client is not initialized for server {self.id}; cannot write {node}' + ) node_obj = self.client.get_node(node) except Exception as e: if is_reconnectable_opcua_bad(e): diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 886dd73..cf0851a 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -444,6 +444,22 @@ def test_write_data_get_node_failed(opc_repository): assert error_data['attachment_content'] is not None +def test_write_data_without_client_fails_as_get_node_error(opc_repository): + """A client torn down between validation and the write is reported, not asserted away.""" + opc_repository.validate_connection = MagicMock(return_value=(True, {})) + opc_repository.client = None + + is_success, error_data = opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + assert is_success is False + assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}' + assert 'OPC client is not initialized' in error_data['message'] + last_call = opc_repository.emit_metric_sync.call_args_list[-1] + assert last_call.args[1]['result'] == 'GetNodeError:OpcClientNotInitializedError' + + def test_write_data_invalid_data_type(opc_repository, mock_client): opc_repository.validate_connection = MagicMock(return_value=(True, {})) opc_repository.client = mock_client