diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index 320c97c..0d8ef09 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -17,18 +17,21 @@ with workflow.unsafe.imports_passed_through(): from laborious.activities.mlflow import MLFlow from laborious.activities.model_metrics import ModelMetrics from laborious.activities.opc import OPC + from laborious.activities.prediction_metadata import PredictionMetadata from laborious.activities.storage import Storage from laborious.utils.connectors_config import build_mlflow_config -class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): +class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetadata): """ Central orchestrator for all Temporal activities used by Laborious workflows. Composes Storage (Postgres + MinIO offload), MLFlow (wrapper-based inference and retrain via ``SientiaMLflowRepository``), Gates (data quality and ML response filters), OPC exports, - drift/simple metrics, and PI Web API writes. The worker constructs one ``Activities`` instance - per process and registers its callables on multiple workers bound to different task queues. + drift/simple metrics, PI Web API writes, and PredictionMetadata (the single write point for + gate metadata — see `laborious.activities.prediction_metadata`). The worker constructs one + ``Activities`` instance per process and registers its callables on multiple workers bound to + different task queues. MLflow connectivity: unless ``mlflow_repository`` is injected (tests only), this class builds ``SientiaMLflowRepository`` from ``build_mlflow_config()`` so tracking credentials and URL @@ -70,7 +73,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): - metrics_controller: Optional shared controller; if ``None``, a new one is created. - mlflow_repository: Optional ``SientiaMLflowRepository`` for unit/e2e tests; in production leave unset so the repository is built from environment via ``build_mlflow_config()``. - - mongo_repository: Optional ``MongoDBRepository`` (gate metadata, best-effort) for + - mongo_repository: Optional ``MongoDBRepository`` (prediction metadata, best-effort) for unit/e2e tests; in production leave unset so the repository is built from environment via ``build_mongodb_config()``, same pattern as ``mlflow_repository`` above. @@ -145,6 +148,13 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): Gates.__init__( self, minio_repository=minio_repository, + logger=logger, + notification_handler=notification_handler, + metrics_controller=mc, + ) + + PredictionMetadata.__init__( + self, mongo_repository=mongo_repository, logger=logger, notification_handler=notification_handler, @@ -192,3 +202,4 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): OPC.close(self) ModelMetrics.close(self) API.close(self) + PredictionMetadata.close(self) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 7cbab54..c6fece9 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -5,7 +5,6 @@ with workflow.unsafe.imports_passed_through(): from collections.abc import Callable, Mapping from typing import Any - from bson import ObjectId from pandas import DataFrame from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel @@ -13,7 +12,6 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.repository.minio_repository_sync import MinioRepository - from sientia_do.repository.mongodb_repository_sync import MongoDBRepository from sientia_do.utils.formatters import create_sample_dict from laborious import metrics @@ -125,14 +123,11 @@ class Gates(SientiaMonitoring): """ minio_repository: MinioRepository | None = None - mongo_repository: MongoDBRepository | None = None _MAX_DEBUG_DATAFRAME_ROWS = 100 - _PREDICTION_METADATA_COLLECTION = 'prediction_metadata' def __init__( self, minio_repository: MinioRepository | None = None, - mongo_repository: MongoDBRepository | None = None, logger: Logger | None = None, notification_handler: NotificationHandler | None = None, metrics_controller: MetricsController | None = None, @@ -143,14 +138,11 @@ class Gates(SientiaMonitoring): Args: logger: Logger instance for observability and debugging notification_handler: Notification handler for alerts and monitoring - mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is - written best-effort. `None` is a valid state (see `_write_prediction_metadata`). Raises: Exception: If BaseActivity initialization fails """ self.minio_repository = minio_repository - self.mongo_repository = mongo_repository SientiaMonitoring.__init__( self, logger=logger, @@ -168,11 +160,6 @@ class Gates(SientiaMonitoring): self.minio_repository.close() finally: self.minio_repository = None - if self.mongo_repository is not None: - try: - self.mongo_repository.close() - finally: - self.mongo_repository = None SientiaMonitoring.shutdown(self) def __del__(self): @@ -196,64 +183,6 @@ class Gates(SientiaMonitoring): metadata, ) - def _write_prediction_metadata( - self, document: dict[str, Any] | None, metadata: dict[str, Any] - ) -> str | None: - """ - Best-effort write of gate metadata (triggered filter, policy, counter) to Mongo. - - Implements both business rules for this feature in one place: - - The prediction write must never be blocked or failed by a metadata write. A failed - `insert(...)` is swallowed, never re-raised, logged here with gate-level context (which - `mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s - `cleanup_minio_objects_expired` also logs again at the call site even though the - repository already logs internally and emits its own Prometheus metric - (`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this - one is for whoever is reading logs for this specific gate. - - Nothing is written when there's nothing relevant to report. A falsy `document` (e.g. - no filter triggered) skips Mongo entirely, so callers can call this unconditionally - and get `None` back for "no metadata this time" instead of deciding upfront whether - there's something worth saving. - - `insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here, - client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this - codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char - size of the `predictions.metadata_id` Postgres column that will reference it. - - Caller contract (not enforced here — no caller exists yet in this codebase): only call - this when the gate's resolved `path_flag` is `CONTINUE`. `STOP` terminates the workflow - before any `predictions` row is written (`prediction_process.py`'s `path_flag_handler`); - `REPEAT` reuses the last existing row instead of writing a new one. In both cases, a - document written here would be orphaned — no `predictions.metadata_id` will ever - reference it. Only `CONTINUE` reaches `format_and_export_prediction`, the one path that - writes a new `predictions` row. - - Args: - document: Metadata payload (filter, policy, counter, etc.); `None`/empty skips the - write entirely. No `_id` key expected. - metadata: Workflow metadata for contextual logging. - - Return: - The generated id, or `None` when there's nothing to write, Mongo isn't configured, or - the insert failed. - """ - if not document or self.mongo_repository is None: - return None - - generated_id = str(ObjectId()) - - try: - self.mongo_repository.insert( - self._PREDICTION_METADATA_COLLECTION, - {**document, '_id': generated_id}, - metadata, - ) - except Exception as e: - self.error(f'Failed to write prediction metadata: {e}', metadata) - return None - - return generated_id - @staticmethod def _read_filter_entry(config: dict[str, Any]) -> tuple[str, dict[str, Any]]: """ @@ -271,7 +200,9 @@ class Gates(SientiaMonitoring): return policy, filter_config @activity.defn(name='input_gate') - def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str, str | None]: + def input_gate( + self, input_data: dict[str, Any] + ) -> tuple[str | None, int, str, dict[str, Any] | None]: """ Apply input data quality filters and validation. @@ -286,8 +217,9 @@ class Gates(SientiaMonitoring): 3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT) 4. Provides confidence scores and detailed comments 5. Handles errors gracefully with notification integration - 6. When the resolved path is CONTINUE, best-effort writes which filter(s) triggered, - the policy and a counter to `prediction_metadata` (see `_write_prediction_metadata`) + 6. Builds a metadata document describing which filter(s) triggered, with their policy + and a counter, whenever one does — this activity only computes it, it never writes it + (see `PredictionMetadata.write_prediction_metadata` for the actual write) Args: input_data: Configuration and data for input validation @@ -298,14 +230,14 @@ class Gates(SientiaMonitoring): - path_priority (list[str]): Priority order for path decisions Returns: - tuple: (path_flag, confidence, comment, metadata_id) + tuple: (path_flag, confidence, comment, document) - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None) - confidence (int): Confidence score for the decision - comment (str): Detailed explanation of the decision - - metadata_id (str | None): Id of the `prediction_metadata` document written for - this gate's decision; `None` unless `path_flag` is `CONTINUE` (STOP/REPEAT - never write a new `predictions` row, so writing metadata for them would - orphan it — see `_write_prediction_metadata`'s caller contract) + - document (dict | None): Metadata describing the filter(s) that produced + `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`; + `None` when nothing was filtered. Not written anywhere by this activity — + the caller decides whether and how to persist it. Raises: Exception: If filter execution fails or configuration is invalid @@ -373,17 +305,12 @@ class Gates(SientiaMonitoring): for fil, filter_config in matched ], } - metadata_id = ( - self._write_prediction_metadata(document, metadata) - if path_flag == 'CONTINUE' - else None - ) return ( path_flag, input_path_confidence[path_flag], 'Input data with bad quality', - metadata_id, + document, ) self.info('Nothing was filtered by the input gate', metadata) @@ -786,6 +713,9 @@ class Gates(SientiaMonitoring): - model_id (str): Unique identifier for the ML model - prediction_confidence (float): Confidence score (typically low for errors) - comment (str): Error description or operational comment + - metadata_id (str | None, optional): Id of the `prediction_metadata` Mongo + document to reference for this row; `None`/absent when there's nothing to + reference Returns: dict: Formatted default prediction data with error indicators @@ -803,6 +733,7 @@ class Gates(SientiaMonitoring): 'prediction_confidence': [input_data['prediction_confidence']], 'prediction_status': ['Bad'], 'comments': [input_data['comment']], + 'metadata_id': [input_data.get('metadata_id')], } ) diff --git a/laborious/activities/prediction_metadata.py b/laborious/activities/prediction_metadata.py new file mode 100644 index 0000000..1fdb519 --- /dev/null +++ b/laborious/activities/prediction_metadata.py @@ -0,0 +1,132 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + from typing import Any + + from bson import ObjectId + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.observability.logger import Logger + from sientia_do.observability.metrics_controller import MetricsController + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + from sientia_do.repository.mongodb_repository_sync import MongoDBRepository + + +class PredictionMetadata(SientiaMonitoring): + """ + Centralized, best-effort writer for prediction gate metadata (triggered filter, policy, + counter) to MongoDB. + + Gates only compute and return a metadata document; they never write it themselves. Keeping + the write here, instead of in each gate, puts the one real write point in the subworkflow + that's about to persist a `predictions` row — the only place that reliably knows whether one + is being written at all — and lets other metadata sources reuse the same activity instead of + duplicating write logic wherever they originate. + """ + + mongo_repository: MongoDBRepository | None = None + _PREDICTION_METADATA_COLLECTION = 'prediction_metadata' + + def __init__( + self, + mongo_repository: MongoDBRepository | None = None, + logger: Logger | None = None, + notification_handler: NotificationHandler | None = None, + metrics_controller: MetricsController | None = None, + ): + """ + Initialize the prediction metadata writer. + + Args: + mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is + written best-effort. `None` is a valid state (see `write_prediction_metadata`). + + Raises: + Exception: If BaseActivity initialization fails + """ + self.mongo_repository = mongo_repository + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + def close(self) -> None: + """ + Close the prediction metadata writer and clean up resources. + """ + if self.mongo_repository is not None: + try: + self.mongo_repository.close() + finally: + self.mongo_repository = None + SientiaMonitoring.shutdown(self) + + def __del__(self): + self.close() + + @activity.defn(name='write_prediction_metadata') + def write_prediction_metadata(self, input_data: dict[str, Any]) -> str | None: + """ + Best-effort write of one or more gate metadata documents to Mongo, combined into a single + `prediction_metadata` document. + + Implements both business rules for this feature in one place: + - The prediction write must never be blocked or failed by a metadata write. A failed + `insert(...)` is swallowed, never re-raised, logged here with call-site context (which + `mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s + `cleanup_minio_objects_expired` also logs again at the call site even though the + repository already logs internally and emits its own Prometheus metric + (`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this + one is for whoever is reading logs for this specific call. + - Nothing is written when there's nothing relevant to report. An empty/missing + `documents` list (or one full of falsy/empty entries) skips Mongo entirely, so callers + can call this unconditionally and get `None` back for "no metadata this time" instead + of deciding upfront whether there's something worth saving. + + Takes a *list* of documents, not a single one: each caller contributes its own document + (shaped like `{'filters': [...]}`), and this flattens every `filters` entry across all of + them into one `prediction_metadata` document before the single insert. Today only one + gate ever contributes a document per prediction cycle (the gate chain short-circuits on + the first non-`None` `path_flag`), but the list shape means this activity won't need to + change again once more than one source contributes metadata for the same cycle. + + `insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here, + client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this + codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char + size of the `predictions.metadata_id` Postgres column that will reference it. + + Args: + input_data: Dict with: + - documents (list[dict[str, Any]]): Metadata payloads to combine, each shaped + like `{'filters': [...]}`. Missing/empty, or entries with no `filters`, skip + the write entirely. + - metadata (dict): Workflow metadata for contextual logging. + + Return: + The generated id, or `None` when there's nothing to write, Mongo isn't configured, or + the insert failed. + """ + metadata = input_data['metadata'] + documents = input_data.get('documents') or [] + + filters = [ + entry for document in documents if document for entry in document.get('filters', []) + ] + + if not filters or self.mongo_repository is None: + return None + + generated_id = str(ObjectId()) + + try: + self.mongo_repository.insert( + self._PREDICTION_METADATA_COLLECTION, + {'filters': filters, '_id': generated_id}, + metadata, + ) + except Exception as e: + self.error(f'Failed to write prediction metadata: {e}', metadata) + return None + + return generated_id diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index a71d399..b44d829 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -249,6 +249,8 @@ async def main(): activities.format_transformed_data, activities.format_prediction, activities.format_default_prediction, + # Prediction metadata + activities.write_prediction_metadata, # OPC activities.write_opc_data, # Postgres / MinIO offload diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index cf950ce..59bcf82 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -75,6 +75,11 @@ class FormatAndExportPrediction: Required when path_flag is None - comment (str): Operational comment or error description Required when path_flag is not None + - metadata_id (str | None): Id of the `prediction_metadata` Mongo document to + reference for this row, already written by the caller; `None` when there's + nothing to reference. Only meaningful when path_flag is not None — the + normal (path_flag is None) path never has metadata to reference, so this + key is irrelevant there. Returns: None: The workflow completes successfully when all export operations finish @@ -154,6 +159,7 @@ class FormatAndExportPrediction: 'model_id': input_data['model_id'], 'prediction_confidence': prediction_confidence, 'comment': input_data['comment'], + 'metadata_id': input_data.get('metadata_id'), }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 67b7fd6..7f3bfac 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -132,27 +132,28 @@ class PredictionProcess: 'path_priority': input_data['path_priority'], } - # `_metadata_id` (prediction_metadata document id, only non-None when path_flag ends up - # CONTINUE) isn't threaded through yet — SIENTIAPDE-2167 wires it to Postgres. Star-unpack - # rather than a fixed 4-tuple: `input_gate` is a local activity, so its result is recorded - # in workflow history; an execution already in flight when this field was added would - # replay a 3-element result here, and exact unpacking would raise ValueError on replay - # (review finding, 2026-09-30). + # `document` describes which filter(s) produced `path_flag` (`None` when nothing was + # filtered); `path_flag_handler` is the one that decides whether and how to persist it. + # Star-unpack rather than a fixed 4-tuple: `input_gate` is a local activity, so its result + # is recorded in workflow history; an execution already in flight when this field was + # added would replay a 3-element result here, and exact unpacking would raise ValueError + # on replay (review finding, 2026-09-30). ( path_flag, confidence, comment, - *_metadata_id, + *document_tail, ) = await workflow.execute_local_activity_method( Activities.input_gate, gate_input, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), ) + document = document_tail[0] if document_tail else None # Handle path decision based on filter results if await self.path_flag_handler( - data, path_flag, input_data, confidence, last_timestamp, comment + data, path_flag, input_data, confidence, last_timestamp, comment, document ): return @@ -258,6 +259,33 @@ class PredictionProcess: }, ) + async def _write_prediction_metadata( + self, metadata: dict[str, Any], document: dict[str, Any] | None + ) -> str | None: + """ + Write a gate's metadata document via the centralized write activity. + + Shared by the CONTINUE and REPEAT branches of `path_flag_handler` — both persist a new + `predictions` row (`REPEAT` reuses the previous row's values but still inserts a new + one), so both need a `metadata_id` to reference it. + + Args: + metadata: Workflow metadata for contextual logging. + document: Metadata produced by the gate that resolved this `path_flag`; `None` when + nothing was filtered, or when the gate that produced this `path_flag` doesn't + build one yet. + + Return: + The generated id, or `None` when there's nothing to write (see + `PredictionMetadata.write_prediction_metadata`). + """ + return await workflow.execute_activity_method( + Activities.write_prediction_metadata, + {**metadata, 'documents': [document] if document else []}, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(minutes=1), + ) + async def path_flag_handler( self, data: dict[str, Any], @@ -266,6 +294,7 @@ class PredictionProcess: confidence: int, last_timestamp: str, comment: str, + document: dict[str, Any] | None = None, ) -> bool: """ Handle path decisions based on filter results and confidence levels. @@ -291,6 +320,13 @@ class PredictionProcess: confidence: Confidence level from filter validation last_timestamp: Last processed timestamp comment: Additional information about the filter result + document: Metadata describing the filter(s) that produced `path_flag`, as returned by + the gate (`None` when nothing was filtered, or when the gate that produced this + `path_flag` doesn't build one yet). Written via `write_prediction_metadata` for + `CONTINUE` and `REPEAT` alike — both persist a `predictions` row (`REPEAT` reuses + the previous row's values but still inserts a new one), so both can reference the + resulting `metadata_id`; `STOP` never reaches a write at all, so its `document` is + simply never used here. Returns: bool: True if processing should stop, False to continue @@ -315,7 +351,11 @@ class PredictionProcess: # Stop processing and exit workflow return True elif path_flag == 'REPEAT': - # Repeat last prediction if available + metadata_id = await self._write_prediction_metadata(metadata, document) + # Repeat last prediction if available. `metadata_id` isn't consumed by + # `repeat_last_prediction` yet (its INSERT...SELECT has no `metadata_id` column — + # pending a change in `sientia-dataops-library`); passed through already so no further + # change is needed here once that lands. await workflow.execute_activity_method( Activities.repeat_last_prediction, { @@ -324,12 +364,14 @@ class PredictionProcess: 'table_name': table_name, 'model': model_id, 'last_timestamp': last_timestamp, + 'metadata_id': metadata_id, }, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), ) return True elif path_flag == 'CONTINUE': + metadata_id = await self._write_prediction_metadata(metadata, document) # call write workflow await workflow.execute_child_workflow( 'subworkflow.format_and_export_prediction', @@ -346,6 +388,7 @@ class PredictionProcess: 'table_name': table_name, 'transform_table_name': transform_table_name, 'comment': comment, + 'metadata_id': metadata_id, 'opc_output_config': input_data['opc_output_config'], 'pi_web_api_output_config': input_data['pi_web_api_output_config'], 'prediction_store_policy': input_data['prediction_store_policy'], diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index d861f76..f902b09 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -6,6 +6,7 @@ from laborious.activities.gates import Gates from laborious.activities.mlflow import MLFlow from laborious.activities.model_metrics import ModelMetrics from laborious.activities.opc import OPC +from laborious.activities.prediction_metadata import PredictionMetadata from laborious.activities.storage import Storage @@ -13,6 +14,7 @@ from laborious.activities.storage import Storage @patch('laborious.activities.activities.MLFlow.__init__') @patch('laborious.activities.activities.OPC.__init__') @patch('laborious.activities.activities.Gates.__init__') +@patch('laborious.activities.activities.PredictionMetadata.__init__') @patch('laborious.activities.activities.ModelMetrics.__init__') @patch('laborious.activities.activities.API.__init__') @patch('laborious.activities.activities.MinioRepository') @@ -22,6 +24,7 @@ def test___init__( mock_minio_repository, mock_api_init, mock_model_metrics_init, + mock_prediction_metadata_init, mock_gates_init, mock_opc_init, mock_mlflow_init, @@ -84,6 +87,7 @@ def test___init__( assert isinstance(activities, Gates) assert isinstance(activities, ModelMetrics) assert isinstance(activities, API) + assert isinstance(activities, PredictionMetadata) mock_storage_init.assert_called_once_with( ANY, @@ -122,6 +126,13 @@ def test___init__( mock_gates_init.assert_called_once_with( ANY, minio_repository=mock_minio_repository.return_value, + logger=logger, + notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, + ) + + mock_prediction_metadata_init.assert_called_once_with( + ANY, mongo_repository=mongo_repository, logger=logger, notification_handler=notification_handler, @@ -161,6 +172,7 @@ def test___init__( @patch('laborious.activities.activities.MLFlow') @patch('laborious.activities.activities.OPC') @patch('laborious.activities.activities.Gates') +@patch('laborious.activities.activities.PredictionMetadata') @patch('laborious.activities.activities.ModelMetrics') @patch('laborious.activities.activities.API') @patch('laborious.activities.activities.MinioRepository') @@ -168,6 +180,7 @@ def test_shutdown( _mock_minio_repository, mock_api_init, mock_model_metrics_init, + mock_prediction_metadata_init, mock_gates_init, mock_opc_init, mock_mlflow_init, @@ -230,6 +243,7 @@ def test_shutdown( mock_gates_init.close.assert_called_once() mock_model_metrics_init.close.assert_called_once() mock_api_init.close.assert_called_once() + mock_prediction_metadata_init.close.assert_called_once() @patch('laborious.activities.activities.SientiaMLflowRepository') @@ -238,6 +252,7 @@ def test_shutdown( @patch('laborious.activities.activities.MLFlow.__init__') @patch('laborious.activities.activities.OPC.__init__') @patch('laborious.activities.activities.Gates.__init__') +@patch('laborious.activities.activities.PredictionMetadata.__init__') @patch('laborious.activities.activities.ModelMetrics.__init__') @patch('laborious.activities.activities.API.__init__') @patch('laborious.activities.activities.MinioRepository') @@ -247,6 +262,7 @@ def test___init___builds_mlflow_repository_when_not_provided( mock_minio_repository, _mock_api_init, _mock_model_metrics_init, + _mock_prediction_metadata_init, _mock_gates_init, _mock_opc_init, _mock_mlflow_init, @@ -310,6 +326,7 @@ def test___init___builds_mlflow_repository_when_not_provided( @patch('laborious.activities.activities.MLFlow.__init__') @patch('laborious.activities.activities.OPC.__init__') @patch('laborious.activities.activities.Gates.__init__') +@patch('laborious.activities.activities.PredictionMetadata.__init__') @patch('laborious.activities.activities.ModelMetrics.__init__') @patch('laborious.activities.activities.API.__init__') @patch('laborious.activities.activities.MinioRepository') @@ -319,6 +336,7 @@ def test___init___builds_mongo_repository_when_not_provided( mock_minio_repository, _mock_api_init, _mock_model_metrics_init, + mock_prediction_metadata_init, mock_gates_init, _mock_opc_init, _mock_mlflow_init, @@ -373,10 +391,17 @@ def test___init___builds_mongo_repository_when_not_provided( notification_handler=notification_handler, metrics_controller=mock_metrics_controller.return_value, ) - # The same repository built here is the one handed to Gates, not a second instance. mock_gates_init.assert_called_once_with( ANY, minio_repository=mock_minio_repository.return_value, + logger=logger, + notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, + ) + # The same repository built here is the one handed to PredictionMetadata, not a second + # instance. + mock_prediction_metadata_init.assert_called_once_with( + ANY, mongo_repository=mock_mongo_repository_cls.return_value, logger=logger, notification_handler=notification_handler, diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 54bf66b..8f0eb80 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -35,14 +35,8 @@ def _minio_payload(retrieve_return, status=None): @fixture -def mongo() -> MagicMock: - return MagicMock() - - -@fixture -def gates_activity(mongo: MagicMock): +def gates_activity(): gates = Gates( - mongo_repository=mongo, logger=MagicMock(), notification_handler=MagicMock(), metrics_controller=MagicMock(spec=MetricsController), @@ -67,75 +61,6 @@ metadata = { } -def test_write_prediction_metadata_no_document_skips_mongo(gates_activity, mongo: MagicMock): - # Act - result = gates_activity._write_prediction_metadata(None, metadata) - - # Assert - assert result is None - mongo.insert.assert_not_called() - - -def test_write_prediction_metadata_empty_document_skips_mongo(gates_activity, mongo: MagicMock): - # Act - result = gates_activity._write_prediction_metadata({}, metadata) - - # Assert - assert result is None - mongo.insert.assert_not_called() - - -def test_write_prediction_metadata_not_configured_returns_none(): - # Arrange - gates = Gates( - mongo_repository=None, - logger=MagicMock(), - notification_handler=MagicMock(), - metrics_controller=MagicMock(spec=MetricsController), - ) - gates.error = MagicMock() - - # Act - result = gates._write_prediction_metadata({'filter_triggered': 'EMPTY_DATA'}, metadata) - - # Assert - assert result is None - gates.error.assert_not_called() - - -def test_write_prediction_metadata_success_generates_id_and_inserts( - gates_activity, mongo: MagicMock -): - # Arrange - document = {'filter_triggered': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3} - - # Act - result = gates_activity._write_prediction_metadata(document, metadata) - - # Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)), - # not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side. - assert result is not None - assert len(result) == 24 - int(result, 16) # raises ValueError if not hex - - collection, inserted_document, inserted_metadata = mongo.insert.call_args.args - assert collection == Gates._PREDICTION_METADATA_COLLECTION - assert inserted_document == {**document, '_id': result} - assert inserted_metadata == metadata - - -def test_write_prediction_metadata_insert_failure_swallowed(gates_activity, mongo: MagicMock): - # Arrange - mongo.insert.side_effect = RuntimeError('connection reset') - - # Act - result = gates_activity._write_prediction_metadata({'filter_triggered': 'EMPTY_DATA'}, metadata) - - # Assert: the prediction write must never be blocked or failed by this - assert result is None - gates_activity.error.assert_called_once() - - def test_input_filter_counters_covers_every_input_filter_function(): # A filter added to input_filter_functions without a matching entry here would degrade to # counter=0 (see the `.get(..., lambda *_: 0)` fallback in input_gate) instead of crashing — @@ -143,7 +68,7 @@ def test_input_filter_counters_covers_every_input_filter_function(): assert set(input_filter_counters) == set(input_filter_functions) -def test_input_gate_invalid_filter(gates_activity, mongo: MagicMock): +def test_input_gate_invalid_filter(gates_activity): # Arrange input_data = { **metadata, @@ -160,7 +85,6 @@ def test_input_gate_invalid_filter(gates_activity, mongo: MagicMock): gates_activity.error.assert_called_once_with( 'Filter INVALID_FILTER not found', metadata['metadata'] ) - mongo.insert.assert_not_called() @patch('laborious.activities.gates.input_filter_functions') @@ -192,7 +116,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity ) -def test_input_gate_no_filters(gates_activity, mongo: MagicMock): +def test_input_gate_no_filters(gates_activity): # Arrange input_data = { **metadata, @@ -207,13 +131,10 @@ def test_input_gate_no_filters(gates_activity, mongo: MagicMock): # Assert assert result == (None, 0, '', None) gates_activity.debug.assert_called() - mongo.insert.assert_not_called() -def test_input_gate_with_filter(gates_activity, mongo: MagicMock): - # Arrange: STOP never writes to Mongo — see business rule refinement (2026-09-29): STOP - # terminates the workflow before any `predictions` row is written, so a metadata document - # written here would be orphaned. +def test_input_gate_with_filter(gates_activity): + # Arrange input_data = { **metadata, 'filters': {'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}}}, @@ -222,12 +143,15 @@ def test_input_gate_with_filter(gates_activity, mongo: MagicMock): } # Act - result = gates_activity.input_gate(input_data) + path_flag, confidence, comment, document = gates_activity.input_gate(input_data) - # Assert - assert result == ('STOP', -1, 'Input data with bad quality', None) + # Assert: the gate only computes the document now — it never writes anywhere, and never + # checks path_flag to decide whether to build one (that decision moved to the caller, see + # `PredictionMetadata.write_prediction_metadata` and `prediction_process.py`), so even a STOP + # resolution still gets its document built. + assert (path_flag, confidence, comment) == ('STOP', -1, 'Input data with bad quality') + assert document == {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]} gates_activity.debug.assert_called() - mongo.insert.assert_not_called() def test_input_gate_with_filter_lowercase_keys(gates_activity): @@ -243,7 +167,12 @@ def test_input_gate_with_filter_lowercase_keys(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == ('STOP', -1, 'Input data with bad quality', None) + assert result == ( + 'STOP', + -1, + 'Input data with bad quality', + {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}, + ) def test_input_gate_with_filter_capitalized_keys(gates_activity): @@ -259,10 +188,15 @@ def test_input_gate_with_filter_capitalized_keys(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == ('STOP', -1, 'Input data with bad quality', None) + assert result == ( + 'STOP', + -1, + 'Input data with bad quality', + {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}, + ) -def test_input_gate_with_filter_not_caught(gates_activity, mongo: MagicMock): +def test_input_gate_with_filter_not_caught(gates_activity): # Arrange input_data = { **metadata, @@ -277,12 +211,12 @@ def test_input_gate_with_filter_not_caught(gates_activity, mongo: MagicMock): # Assert assert result == (None, 0, '', None) gates_activity.debug.assert_called() - mongo.insert.assert_not_called() -def test_input_gate_repeat_does_not_write_prediction_metadata(gates_activity, mongo: MagicMock): - # Arrange: REPEAT reuses the last existing row instead of writing a new one, so it must not - # write metadata either — same orphaning risk as STOP. +def test_input_gate_repeat_also_builds_document(gates_activity): + # Arrange: REPEAT used to be treated like STOP (no document at all) — it isn't anymore, since + # REPEAT does insert a new `predictions` row (reusing the previous row's values), so it needs + # a document/metadata_id just like CONTINUE (2026-10-01 redesign). input_data = { **metadata, 'filters': {'EMPTY_DATA': {'POLICY': 'REPEAT', 'CONFIG': {}}}, @@ -294,11 +228,15 @@ def test_input_gate_repeat_does_not_write_prediction_metadata(gates_activity, mo result = gates_activity.input_gate(input_data) # Assert - assert result == ('REPEAT', -1, 'Input data with bad quality', None) - mongo.insert.assert_not_called() + assert result == ( + 'REPEAT', + -1, + 'Input data with bad quality', + {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}, + ) -def test_input_gate_continue_writes_prediction_metadata(gates_activity, mongo: MagicMock): +def test_input_gate_continue_builds_document_with_counter(gates_activity): # Arrange: two rows with a null in one of the configured variables ('x'); a third row is # clean, to confirm the counter reflects only the rows that actually matched the condition. input_data = { @@ -316,23 +254,16 @@ def test_input_gate_continue_writes_prediction_metadata(gates_activity, mongo: M } # Act - path_flag, confidence, comment, metadata_id = gates_activity.input_gate(input_data) + path_flag, confidence, comment, document = gates_activity.input_gate(input_data) # Assert assert (path_flag, confidence, comment) == ('CONTINUE', 2, 'Input data with bad quality') - assert metadata_id is not None - collection, inserted_document, inserted_metadata = mongo.insert.call_args.args - assert collection == Gates._PREDICTION_METADATA_COLLECTION - assert inserted_document == { - 'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}], - '_id': metadata_id, + assert document == { + 'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}] } - assert inserted_metadata == metadata['metadata'] -def test_input_gate_continue_multiple_filters_same_policy_lists_each( - gates_activity, mongo: MagicMock -): +def test_input_gate_multiple_filters_same_policy_lists_each(gates_activity): # Arrange: force both input filters to trigger (not reachable with today's real filter # conditions — EMPTY_DATA and SPECIFIC_VARIABLES_NULL_VALUES are mutually exclusive) to lock # in the list-building behavior approved by Vitor (2026-09-30): one entry per filter sharing @@ -355,17 +286,15 @@ def test_input_gate_continue_multiple_filters_same_policy_lists_each( } # Act - path_flag, _, _, metadata_id = gates_activity.input_gate(input_data) + path_flag, _, _, document = gates_activity.input_gate(input_data) # Assert assert path_flag == 'CONTINUE' - assert metadata_id is not None - _, inserted_document, _ = mongo.insert.call_args.args - assert {f['name'] for f in inserted_document['filters']} == { + assert {f['name'] for f in document['filters']} == { 'SPECIFIC_VARIABLES_NULL_VALUES', 'EMPTY_DATA', } - assert all(f['policy'] == 'CONTINUE' for f in inserted_document['filters']) + assert all(f['policy'] == 'CONTINUE' for f in document['filters']) def test_mlflow_response_gate_invalid_filter(gates_activity): diff --git a/tests/laborious/activities/test_prediction_metadata.py b/tests/laborious/activities/test_prediction_metadata.py new file mode 100644 index 0000000..b7c1143 --- /dev/null +++ b/tests/laborious/activities/test_prediction_metadata.py @@ -0,0 +1,179 @@ +from unittest.mock import MagicMock, call, patch + +from pytest import fixture +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring + +from laborious.activities.prediction_metadata import PredictionMetadata + + +@fixture +def mongo() -> MagicMock: + return MagicMock() + + +@fixture +def prediction_metadata(mongo: MagicMock): + activity = PredictionMetadata( + mongo_repository=mongo, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(spec=MetricsController), + ) + activity.error = MagicMock() + return activity + + +metadata = { + 'metadata': { + 'model_id': 'test_model', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + 'schema_name': 'test_schedule', + }, +} + + +def test_write_prediction_metadata_no_documents_skips_mongo(prediction_metadata, mongo: MagicMock): + # Act + result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': []}) + + # Assert + assert result is None + mongo.insert.assert_not_called() + + +def test_write_prediction_metadata_missing_documents_key_skips_mongo( + prediction_metadata, mongo: MagicMock +): + # Act + result = prediction_metadata.write_prediction_metadata({**metadata}) + + # Assert + assert result is None + mongo.insert.assert_not_called() + + +def test_write_prediction_metadata_documents_with_no_filters_skips_mongo( + prediction_metadata, mongo: MagicMock +): + # Arrange: a falsy document (None) and one with an empty `filters` list — neither contributes + # anything to write. + documents = [None, {'filters': []}] + + # Act + result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents}) + + # Assert + assert result is None + mongo.insert.assert_not_called() + + +def test_write_prediction_metadata_not_configured_returns_none(): + # Arrange + activity = PredictionMetadata( + mongo_repository=None, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(spec=MetricsController), + ) + activity.error = MagicMock() + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]} + + # Act + result = activity.write_prediction_metadata({**metadata, 'documents': [document]}) + + # Assert + assert result is None + activity.error.assert_not_called() + + +def test_write_prediction_metadata_success_generates_id_and_inserts( + prediction_metadata, mongo: MagicMock +): + # Arrange + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}]} + + # Act + result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]}) + + # Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)), + # not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side. + assert result is not None + assert len(result) == 24 + int(result, 16) # raises ValueError if not hex + + collection, inserted_document, inserted_metadata = mongo.insert.call_args.args + assert collection == PredictionMetadata._PREDICTION_METADATA_COLLECTION + assert inserted_document == {'filters': document['filters'], '_id': result} + assert inserted_metadata == metadata['metadata'] + + +def test_write_prediction_metadata_combines_multiple_documents( + prediction_metadata, mongo: MagicMock +): + # Arrange: two documents (e.g. from two different sources contributing metadata for the same + # prediction cycle) — their `filters` entries are flattened into one combined document. + documents = [ + {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]}, + { + 'filters': [ + {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2} + ] + }, + ] + + # Act + result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents}) + + # Assert + assert result is not None + _, inserted_document, _ = mongo.insert.call_args.args + assert inserted_document['filters'] == [ + {'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}, + {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}, + ] + + +def test_write_prediction_metadata_insert_failure_swallowed(prediction_metadata, mongo: MagicMock): + # Arrange + mongo.insert.side_effect = RuntimeError('connection reset') + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]} + + # Act + result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]}) + + # Assert: the prediction write must never be blocked or failed by this + assert result is None + prediction_metadata.error.assert_called_once() + + +@patch.object(SientiaMonitoring, 'shutdown') +def test_close_disposes_mongo_repository(mock_shutdown): + """``PredictionMetadata.close`` should close the Mongo client and clear the reference.""" + mongo = MagicMock() + activity = PredictionMetadata( + mongo_repository=mongo, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(spec=MetricsController), + ) + activity.close() + mongo.close.assert_called_once() + assert activity.mongo_repository is None + # Other instances collected by GC during the test also hit the patched shutdown. + assert mock_shutdown.call_args_list.count(call(activity)) == 1 + + +@patch.object(SientiaMonitoring, 'shutdown') +def test_close_without_mongo_repository(mock_shutdown): + """When no Mongo repository is configured, ``close`` only shuts down monitoring.""" + activity = PredictionMetadata( + mongo_repository=None, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(spec=MetricsController), + ) + activity.close() + # Other instances collected by GC during the test also hit the patched shutdown. + assert mock_shutdown.call_args_list.count(call(activity)) == 1 diff --git a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py index 4fbc62b..9d44b76 100644 --- a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py +++ b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py @@ -326,6 +326,7 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction 'model_id': input_data['model_id'], 'prediction_confidence': input_data['prediction_confidence'], 'comment': input_data['comment'], + 'metadata_id': None, }, retry_policy=ANY, start_to_close_timeout=ANY, @@ -389,6 +390,56 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction assert workflow_mock.execute_local_activity_method.call_count == 1 +@mark.asyncio +@patch( + 'laborious.workflows.sub_workflows.format_and_export_prediction.workflow', + new_callable=AsyncMock, +) +async def test_run_default_path_flag_passes_metadata_id( + workflow_mock, format_and_export_prediction +): + # Arrange: metadata_id reaches format_default_prediction, which is the row that ends up in + # Postgres via export_data_to_postgres (CONTINUE/REPEAT path). + input_data = { + 'metadata': metadata, + 'path_flag': 'default', + 'data': {'test': 'data'}, + 'timestamp': '2021-01-01', + 'model_id': 1, + 'model_name': metadata['metadata']['model_name'], + 'prediction_confidence': 0, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'opc_output_config': {'test': 'config'}, + 'comment': 'test_comment', + 'metadata_id': 'a' * 24, + } + + workflow_mock.execute_activity_method.side_effect = [ + (MagicMock(), MagicMock()), + MagicMock(), + MagicMock(), + ] + + # Act + await format_and_export_prediction.run(input_data) + + # Assert + workflow_mock.execute_local_activity_method.assert_any_call( + Activities.format_default_prediction, + { + **metadata, + 'timestamp': input_data['timestamp'], + 'model_id': input_data['model_id'], + 'prediction_confidence': input_data['prediction_confidence'], + 'comment': input_data['comment'], + 'metadata_id': 'a' * 24, + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + + @mark.asyncio @patch( 'laborious.workflows.sub_workflows.format_and_export_prediction.workflow', @@ -638,6 +689,20 @@ async def test_run_default_path_flag_with_pi_web_api(workflow_mock, format_and_e await format_and_export_prediction.run(input_data) + workflow_mock.execute_local_activity_method.assert_any_call( + Activities.format_default_prediction, + { + **metadata, + 'timestamp': input_data['timestamp'], + 'model_id': input_data['model_id'], + 'prediction_confidence': input_data['prediction_confidence'], + 'comment': input_data['comment'], + 'metadata_id': None, + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + workflow_mock.execute_activity_method.assert_has_calls( [ call( diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index cc84661..92715a2 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -598,6 +598,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process): last_timestamp = '2024-01-01' model_name = 'test_model_name' model_config = {'retention': '30'} + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]} # Act result = await prediction_process.path_flag_handler( @@ -616,18 +617,22 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process): confidence, last_timestamp, '', + document, ) - # Assert + # Assert: STOP never reaches a write, even though it now gets a `document` same as CONTINUE + # and REPEAT do — it just never has anywhere to use it. assert result is True workflow_mock.execute_local_activity_method.assert_not_called() + workflow_mock.execute_activity_method.assert_not_called() workflow_mock.execute_child_workflow.assert_not_called() @mark.asyncio @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) async def test_path_flag_handler_repeat(workflow_mock, prediction_process): - # Arrange + # Arrange: REPEAT now writes metadata just like CONTINUE — it does insert a new `predictions` + # row (reusing the previous row's values), so it needs a `metadata_id` too. data = {'test': 'data'} path_flag = 'repeat' confidence = 0.95 @@ -637,6 +642,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): last_timestamp = '2024-01-01' model_name = 'test_model_name' model_config = {'retention': '30'} + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]} # Act result = await prediction_process.path_flag_handler( @@ -655,11 +661,20 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): confidence, last_timestamp, '', + document, ) # Assert assert result is True - workflow_mock.execute_activity_method.assert_called_once_with( + metadata_id = workflow_mock.execute_activity_method.return_value + assert workflow_mock.execute_activity_method.call_count == 2 + workflow_mock.execute_activity_method.assert_any_call( + Activities.write_prediction_metadata, + {**metadata, 'documents': [document]}, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + workflow_mock.execute_activity_method.assert_any_call( Activities.repeat_last_prediction, { **metadata, @@ -667,6 +682,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): 'table_name': table_name, 'model': model, 'last_timestamp': last_timestamp, + 'metadata_id': metadata_id, }, retry_policy=ANY, start_to_close_timeout=ANY, @@ -674,6 +690,45 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): workflow_mock.execute_child_workflow.assert_not_called() +@mark.asyncio +@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) +async def test_path_flag_handler_repeat_no_document_writes_nothing( + workflow_mock, prediction_process +): + # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet) still + # calls write_prediction_metadata, with an empty `documents` list — the activity itself (not + # this workflow) decides there's nothing to write and returns `None`. + data = {'test': 'data'} + + # Act + result = await prediction_process.path_flag_handler( + data, + 'repeat', + { + 'metadata': metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'transform_table_name': 'test_transform_table', + 'model_id': 'test_model', + 'last_timestamp': '2024-01-01', + 'model_name': 'test_model_name', + 'model_config': {'retention': '30'}, + }, + 0.95, + '2024-01-01', + '', + ) + + # Assert + assert result is True + workflow_mock.execute_activity_method.assert_any_call( + Activities.write_prediction_metadata, + {**metadata, 'documents': []}, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + + @mark.asyncio @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) async def test_path_flag_handler_continue(workflow_mock, prediction_process): @@ -688,6 +743,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): model_name = 'test_model_name' model_config = {'retention': '30'} prediction_store_policy = 'erl:1' + document = { + 'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}] + } # Act result = await prediction_process.path_flag_handler( @@ -713,11 +771,18 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): confidence, last_timestamp, 'Prediction Process', + document, ) # Assert + metadata_id = workflow_mock.execute_activity_method.return_value assert result is True - workflow_mock.execute_activity_method.assert_not_called() + workflow_mock.execute_activity_method.assert_called_once_with( + Activities.write_prediction_metadata, + {**metadata, 'documents': [document]}, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) workflow_mock.execute_child_workflow.assert_called_once_with( 'subworkflow.format_and_export_prediction', { @@ -733,6 +798,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'table_name': table_name, 'transform_table_name': 'test_transform_table', 'comment': 'Prediction Process', + 'metadata_id': metadata_id, 'opc_output_config': {'test': 'config'}, 'pi_web_api_output_config': { 'endpoint': 'https://test-pi-server.com',