gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #59 from Aignosi/feature/SIENTIAPDE-2191-adaptive-state

Browse Source 
feat(simple_metrics): adaptive state on the pipelines doc (SIENTIAPDE-2191)
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-09 14:25:56 -03:00
committed by GitHub
parent 9f8237a46e 6cc69b0f93
commit 6fce847fce
20 changed files with 1039 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@@ -489,14 +489,16 @@ The **SimpleMetrics** workflow calculates regression metrics (RMSE, MSE, MAE, R
1. **Data Loading**: Loads prediction vs target data via a JOIN query
2. **Metrics Calculation**: Calculates configured regression metrics; alerts on `simple_metrics_thresholds` breaches (one alert per model, the message names the window) unless the window has fewer rows than `simple_metrics_min_samples`
3. **Data Export**: Exports metrics to PostgreSQL
4. **Adaptive State**: when the schedule's `pipelines` doc has `frequency_alert`, escalates `adaptive_state` on a breach and recovers after the window is renewed without one (`min(12, ceil(interval_minutes / frequency_alert))` runs, at least 1; empty, under-sampled or threshold-less windows count as healthy). The orchestrator runs the schedule at `frequency_alert` while escalated. Runs on every schedule (it reads the `pipelines` doc to check for `frequency_alert`), so a MongoDB failure here fails the run; the metrics are already exported by then
#### Architecture Diagram
```mermaid
flowchart LR
A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres]
A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres] --> D[4. update_simple_metrics_adaptive_state]
A -.-> Database[(Database)]
C -.-> PostgreSQL[(PostgreSQL)]
D -.-> MongoDB[(MongoDB pipelines)]
```
#### Input Parameters
@@ -527,7 +529,7 @@ flowchart LR
- PostgreSQL database
- MLFlow server
- MinIO object storage (for MLFlow artifacts)
- MongoDB server (for notifications)
- MongoDB server (notifications, prediction metadata, and read/write on the orchestrator's `pipelines` collection for the simple-metrics adaptive state)
- OPC server(s) if using OPC export
- PI Web API server if using PI Web API export
@@ -903,7 +905,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes |
| `MONGODB_USERNAME` | MongoDB username | `root` | Yes |
| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes |
| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes |
| `MONGODB_DATABASE_NAME` | MongoDB database name; must be the same database as the orchestrator's `pipelines` | `sientia` | Yes |
| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No |
| `LOG_LEVEL` | Application log level | `INFO` | No |
| `PROJECT_NAME` | Project name for metrics | `laborious` | No |

1
e2e/conftest.py
View File

@@ -445,6 +445,7 @@ def _simple_metrics_worker_activity_list(test_activities: Activities):
test_activities.load_custom_query,
test_activities.calculate_simple_metrics,
test_activities.export_data_to_postgres,
test_activities.update_simple_metrics_adaptive_state,
]

9
e2e/scenarios.md
View File

@@ -413,6 +413,15 @@ four metrics with closed-form expected values.
#### S.1.2 Subset metrics
**Summary**: Requesting `metrics=['rmse']` writes only the rmse row.
#### S.1.3 Threshold breach escalates the adaptive state
**Summary**: With `simple_metrics_thresholds` breached and a `pipelines` doc
(same `schedule_name`) carrying `frequency_alert`, the workflow sets
`adaptive_state={escalated: true, healthy_streak: 0}` and stamps `updated_at`.
**Expected Outcome**:
- `adaptive_state` written on the `pipelines` doc.
- `updated_at` is a BSON datetime (the orchestrator calls `.replace(tzinfo=UTC)` on it).
### 6.2 Edge cases
#### S.2.1 Zero-variance target returns r2=0

52
e2e/test_simple_metrics.py
View File

@@ -15,6 +15,7 @@ from decimal import Decimal
import math
import pytest
from pymongo import MongoClient
from sqlalchemy import text
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
@@ -332,3 +333,54 @@ async def test_simple_metrics_no_overlapping_data_short_circuits(
{'m': str(model_id)},
).scalar()
assert count == 0, 'Empty target data must short-circuit and skip persistence'
@pytest.mark.asyncio
@pytest.mark.integration
async def test_simple_metrics_breach_escalates_adaptive_state(
temporal_test_env: WorkflowEnvironment,
temporal_worker_simple_metrics: Worker,
test_activities: Activities,
postgres_engine,
mongo_container,
):
"""
Scenario S.1.3: a threshold breach on a schedule whose pipelines doc has
``frequency_alert`` escalates ``adaptive_state`` and stamps ``updated_at``
as a BSON datetime (the type the orchestrator reads).
"""
model_id = 541
schedule_name = 'test-simple-metrics-adaptive'
_seed_predictions_and_targets(
postgres_engine, model_id=model_id, pairs=[(1.0, 2.0), (2.0, 4.0), (3.0, 5.0)]
)
mongo = MongoClient(f'mongodb://localhost:{mongo_container.get_exposed_port(27017)}')
pipelines = mongo['test_db']['pipelines']
try:
pipelines.delete_many({'schedule_name': schedule_name})
pipelines.insert_one(
{'schedule_name': schedule_name, 'frequency': '1h', 'frequency_alert': '5m'}
)
input_data = _simple_metrics_input(
model_id,
schedule_name=schedule_name,
model_config={
'target': 'sensor_target',
'simple_metrics_thresholds': {'rmse_max': 0.01},
},
)
await start_and_await_workflow(
temporal_test_env.client,
SimpleMetrics.run,
input_data,
make_workflow_id('test-simple-metrics-adaptive'),
)
doc = pipelines.find_one({'schedule_name': schedule_name})
finally:
mongo.close()
assert doc['adaptive_state'] == {'escalated': True, 'healthy_streak': 0}
assert isinstance(doc['updated_at'], datetime)

18
laborious/activities/activities.py
View File

@@ -16,22 +16,22 @@ with workflow.unsafe.imports_passed_through():
from laborious.activities.gates import Gates
from laborious.activities.mlflow import MLFlow
from laborious.activities.model_metrics import ModelMetrics
from laborious.activities.mongo import MongoActivities
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, PredictionMetadata):
class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, MongoActivities):
"""
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, 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.
drift/simple metrics, PI Web API writes, and MongoActivities (the single write point for
gate metadata and the simple metrics adaptive state — see `laborious.activities.mongo`).
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
@@ -73,7 +73,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetad
- 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`` (prediction metadata, best-effort) for
- mongo_repository: Optional ``MongoDBRepository`` (prediction metadata, adaptive state) 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.
@@ -153,7 +153,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetad
metrics_controller=mc,
)
PredictionMetadata.__init__(
MongoActivities.__init__(
self,
mongo_repository=mongo_repository,
logger=logger,
@@ -202,4 +202,4 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetad
OPC.close(self)
ModelMetrics.close(self)
API.close(self)
PredictionMetadata.close(self)
MongoActivities.close(self)

2
laborious/activities/gates.py
View File

@@ -219,7 +219,7 @@ class Gates(SientiaMonitoring):
5. Handles errors gracefully with notification integration
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)
(see `MongoActivities.write_prediction_metadata` for the actual write)
Args:
input_data: Configuration and data for input validation

116
laborious/activities/model_metrics.py
View File

@@ -4,6 +4,7 @@ with workflow.unsafe.imports_passed_through():
import time
import traceback
import warnings
from math import ceil
from typing import Any
import pandas as pd
@@ -26,6 +27,73 @@ warnings.filterwarnings(
)
def threshold_breaches(
rows: list[dict[str, Any]], thresholds: dict[str, float] | None
) -> list[str]:
"""
Breach messages for metric rows outside their configured thresholds.
Convention: `<metric>_max` breaches when value > threshold, `<metric>_min` when value <
threshold. Shared by the breach alert and the adaptive-state rule so both agree on what a
breach is.
"""
thresholds = thresholds or {}
breaches = []
for row in rows:
metric_name, value = row['metric'], row['value']
max_key, min_key = f'{metric_name}_max', f'{metric_name}_min'
if max_key in thresholds and value > thresholds[max_key]:
breaches.append(f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}')
elif min_key in thresholds and value < thresholds[min_key]:
breaches.append(f'{metric_name}={value} below {min_key}={thresholds[min_key]}')
return breaches
_MAX_RECOVERY_RUNS = 12
_FREQUENCY_UNIT_SECONDS = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
def frequency_to_seconds(frequency: str) -> int:
"""`'5m'` -> 300. Same `'<int><s|m|h|d>'` format the orchestrator schedules with."""
# ponytail: copy of orchestrator/utils/converters.py:parse_frequency (no shared helper in
# sientia_do), same int() parsing so both repos accept the same strings, plus a > 0 guard;
# move it to sientia_do if a third repo needs it.
unit = _FREQUENCY_UNIT_SECONDS.get(frequency[-1:])
if unit is None:
raise ValueError(f'Invalid frequency: {frequency!r}')
value = int(frequency[:-1])
if value <= 0:
raise ValueError(f'Invalid frequency: {frequency!r}')
return value * unit
def recovery_runs(interval_minutes: int, frequency_alert: str) -> int:
"""
N of the hysteresis rule: healthy runs needed to renew the whole window without a breach,
capped at 12 (SIENTIAPDE-1995 decision 3).
"""
runs = ceil(interval_minutes * 60 / frequency_to_seconds(frequency_alert))
return min(_MAX_RECOVERY_RUNS, max(1, runs))
def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dict[str, Any]:
"""
Hysteresis rule of the adaptive frequency (SIENTIAPDE-1995): escalate on the first breach,
recover after `runs` consecutive non-breach runs.
"""
if breached:
return {'escalated': True, 'healthy_streak': 0}
if not state.get('escalated'):
return {'escalated': False, 'healthy_streak': 0}
try:
streak = int(state.get('healthy_streak', 0)) + 1
except (TypeError, ValueError, OverflowError):
streak = 1 # hand-edited streak (null, text, inf): restart the count, stay escalated
if streak >= runs:
return {'escalated': False, 'healthy_streak': 0}
return {'escalated': True, 'healthy_streak': streak}
class ModelMetrics(SientiaMonitoring):
"""
Metrics and statistical analysis activities for the Laborious pipeline.
@@ -430,37 +498,21 @@ class ModelMetrics(SientiaMonitoring):
self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
# Threshold alerting (optional — no crash when absent)
thresholds = input_data.get('thresholds')
if thresholds:
# Convention: _max thresholds breach when value > threshold,
# _min thresholds breach when value < threshold.
for row in output_data:
metric_name = row['metric']
value = row['value']
max_key = f'{metric_name}_max'
min_key = f'{metric_name}_min'
breach_msg = None
if max_key in thresholds and value > thresholds[max_key]:
breach_msg = f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}'
elif min_key in thresholds and value < thresholds[min_key]:
breach_msg = f'{metric_name}={value} below {min_key}={thresholds[min_key]}'
if breach_msg:
message = (
f'Threshold breach for model {model_id} '
f'(window {interval_minutes} min): {breach_msg}'
)
self.warning(message, metadata)
self.send_notification(
# Model-level trigger: the orchestrator dedups alerts on
# `<trigger>:<notification_id>`, so one breach seen by several windows
# (one schedule each) becomes one alert (SIENTIAPDE-1995).
metadata={**metadata, 'schedule_name': f'{model_id}-simple_metrics'},
notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
message=message,
block='model_metrics',
level=NotificationLevel.WARNING,
)
for breach_msg in threshold_breaches(output_data, input_data.get('thresholds')):
message = (
f'Threshold breach for model {model_id} '
f'(window {interval_minutes} min): {breach_msg}'
)
self.warning(message, metadata)
self.send_notification(
# Model-level trigger: the orchestrator dedups alerts on
# `<trigger>:<notification_id>`, so one breach seen by several windows
# (one schedule each) becomes one alert (SIENTIAPDE-1995).
metadata={**metadata, 'schedule_name': f'{model_id}-simple_metrics'},
notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
message=message,
block='model_metrics',
level=NotificationLevel.WARNING,
)
return data.to_dict(orient='records')

227
laborious/activities/mongo.py Normal file
View File

@@ -0,0 +1,227 @@
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
from sientia_do.temporal.constants import now
from laborious.activities.model_metrics import (
next_adaptive_state,
recovery_runs,
threshold_breaches,
)
class MongoActivities(SientiaMonitoring):
"""
Activities that read from or write to MongoDB, sharing one `MongoDBRepository`.
- `write_prediction_metadata`: centralized, best-effort writer for prediction gate metadata
(triggered filter, policy, counter). 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.
- `update_simple_metrics_adaptive_state`: adaptive frequency state on the `pipelines` doc of
a simple metrics schedule (SIENTIAPDE-1995).
"""
mongo_repository: MongoDBRepository | None = None
_PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
_PIPELINES_COLLECTION = 'pipelines'
def __init__(
self,
mongo_repository: MongoDBRepository | None = None,
logger: Logger | None = None,
notification_handler: NotificationHandler | None = None,
metrics_controller: MetricsController | None = None,
):
"""
Initialize the Mongo activities.
Args:
mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
written best-effort and the simple metrics adaptive state is kept. `None` is a
valid state: both activities become no-ops.
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 Mongo repository 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
@activity.defn(name='update_simple_metrics_adaptive_state')
def update_simple_metrics_adaptive_state(self, input_data: dict[str, Any]) -> dict | None:
"""
Update the adaptive frequency state on the schedule's `pipelines` doc.
The orchestrator derives the schedule frequency from `adaptive_state.escalated` and
`frequency_alert`. State is read from Mongo, not from the workflow input, because the
input only refreshes on the orchestrator tick and would be stale while escalated.
`updated_at` is bumped only when `escalated` flips: that is what makes the orchestrator
rewrite the schedule, and a streak change alone does not change the frequency.
Args:
input_data: Dict with:
- metadata (dict): Workflow metadata; `schedule_name` identifies the doc.
- simple_metrics (list[dict]): Rows from `calculate_simple_metrics`; empty for an
empty or under-sampled window, which counts as a non-breach run.
- thresholds (dict): `simple_metrics_thresholds` from `model_config` (may be None: no breach).
- interval_minutes (int): Window size, used to compute the recovery runs.
Return:
The new state, or `None` when adaptive mode is off, Mongo is not configured, the doc
is missing or `frequency_alert` is invalid.
Raises:
Exception: Mongo read/write failures, so the workflow run fails explicitly.
"""
metadata = input_data['metadata']
if self.mongo_repository is None:
return None
# Mongo errors propagate on purpose: the run must fail explicitly so the problem shows
# up in the Temporal metrics (PR #59 review). Metrics were already exported by then.
query = {'schedule_name': metadata['schedule_name']}
pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata)
if not pipeline:
# Likely misconfig: MONGODB_DATABASE_NAME differs from the orchestrator's, or the
# schedule_name does not match its doc.
self.warning(
f'No pipelines doc for schedule {metadata["schedule_name"]}: '
'adaptive state skipped',
metadata,
)
return None
frequency_alert = pipeline.get('frequency_alert')
if not frequency_alert:
return None
try:
runs = recovery_runs(input_data['interval_minutes'], str(frequency_alert))
except (TypeError, ValueError):
# Config error, not a failure: the orchestrator ignores an invalid `frequency_alert`
# too (keeps `frequency`) and already sends SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR
# for it. `interval_minutes` is untyped Mongo config as well (e.g. "60" or null).
self.warning(
f'Invalid frequency_alert={frequency_alert!r} or '
f'interval_minutes={input_data["interval_minutes"]!r}: adaptive state skipped',
metadata,
)
return None
breached = bool(
threshold_breaches(input_data['simple_metrics'], input_data.get('thresholds'))
)
state = pipeline.get('adaptive_state')
if not isinstance(state, dict):
# Missing or hand-edited garbage: not escalated, same as the orchestrator reads it.
state = {'escalated': False, 'healthy_streak': 0}
new_state = next_adaptive_state(state, breached, runs)
if new_state == state:
return new_state
update: dict[str, Any] = {'adaptive_state': new_state}
if new_state['escalated'] != bool(state.get('escalated')):
# BSON datetime, never a string: the orchestrator normalizes `updated_at` with
# `.replace(tzinfo=UTC).strftime(...)` and a string would fail its whole tick.
update['updated_at'] = now()
self.mongo_repository.update_one(
self._PIPELINES_COLLECTION, query, {'$set': update}, metadata
)
self.info(f'Adaptive state of {query["schedule_name"]}: {state} -> {new_state}', metadata)
return new_state

132
laborious/activities/prediction_metadata.py
View File

@@ -1,132 +0,0 @@
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

2
laborious/activities/storage.py
View File

@@ -184,7 +184,7 @@ class Storage(Postgres, SientiaMonitoring):
REPEAT persists a new row (unlike STOP, which persists nothing), reusing the previous
row's `prediction`/`response_time`/`prediction_status`/`prediction_confidence` values with
a fresh `timestamp`/`created_at`. `metadata_id` references the `prediction_metadata`
document for this cycle, if any (see `PredictionMetadata.write_prediction_metadata`);
document for this cycle, if any (see `MongoActivities.write_prediction_metadata`);
`None` is a valid value — the column is nullable.
Args (query_items):

1
laborious/worker/worker.py
View File

@@ -217,6 +217,7 @@ async def main():
activities.load_custom_query,
activities.calculate_simple_metrics,
activities.export_data_to_postgres,
activities.update_simple_metrics_adaptive_state,
],
logger=logger,
runtime='core',

87
laborious/workflows/simple_metrics.py
View File

@@ -71,48 +71,63 @@ class SimpleMetrics:
start_to_close_timeout=timedelta(seconds=300),
)
if not target_data:
return
# Too few rows for a reliable value: metrics are still persisted, but a noisy short
# window must not alert (SIENTIAPDE-1995).
enough_samples = not min_samples or len(target_data) >= min_samples
if not enough_samples:
# window must neither alert nor escalate (SIENTIAPDE-1995).
enough_samples = bool(target_data) and (not min_samples or len(target_data) >= min_samples)
if target_data and not enough_samples:
workflow.logger.info(
f'{len(target_data)} rows < simple_metrics_min_samples={min_samples}: '
'threshold alerting skipped'
'threshold alerting and escalation skipped'
)
simple_metrics = await workflow.execute_local_activity_method(
Activities.calculate_simple_metrics,
{
**metadata,
'model_id': model_id,
'target_data': target_data,
'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
'interval_minutes': interval_minutes,
'model_type': model_type,
'thresholds': thresholds if enough_samples else None,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=300),
)
if not simple_metrics:
return
await workflow.execute_activity_method(
Activities.export_data_to_postgres,
{
**metadata,
'data': simple_metrics,
'schema': input_data['schema'],
'table_name': input_data['target_table_name'],
'timestamp_conversion': {
'column': 'timestamp',
'format': DATETIME_FORMAT_WITH_TZ,
simple_metrics = None
if target_data:
simple_metrics = await workflow.execute_local_activity_method(
Activities.calculate_simple_metrics,
{
**metadata,
'model_id': model_id,
'target_data': target_data,
'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
'interval_minutes': interval_minutes,
'model_type': model_type,
'thresholds': thresholds if enough_samples else None,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=300),
)
if simple_metrics:
await workflow.execute_activity_method(
Activities.export_data_to_postgres,
{
**metadata,
'data': simple_metrics,
'schema': input_data['schema'],
'table_name': input_data['target_table_name'],
'timestamp_conversion': {
'column': 'timestamp',
'format': DATETIME_FORMAT_WITH_TZ,
},
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=300),
)
# No try/except: a failure here must fail the run so it shows up in the Temporal metrics
# (metrics are already persisted above).
await workflow.execute_activity_method(
Activities.update_simple_metrics_adaptive_state,
{
**metadata,
# Only a valid breach escalates: an empty or under-sampled window is sent as no
# rows, which counts as a healthy run (spec decision 8). Always called, even
# without thresholds (no breach), so an escalated doc recovers after they are
# removed.
'simple_metrics': (simple_metrics or []) if enough_samples else [],
'thresholds': thresholds,
'interval_minutes': interval_minutes,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=300),
start_to_close_timeout=timedelta(minutes=1),
)

2
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -267,7 +267,7 @@ class PredictionProcess:
Return:
The generated id, or `None` when there's nothing to write, including when the
activity call itself fails (see `PredictionMetadata.write_prediction_metadata`).
activity call itself fails (see `MongoActivities.write_prediction_metadata`).
"""
if not document:
# Skip the remote activity call entirely rather than dispatching it just to have

28
tests/laborious/activities/test_activities.py
View File

@@ -5,8 +5,8 @@ from laborious.activities.api import API
from laborious.activities.gates import Gates
from laborious.activities.mlflow import MLFlow
from laborious.activities.model_metrics import ModelMetrics
from laborious.activities.mongo import MongoActivities
from laborious.activities.opc import OPC
from laborious.activities.prediction_metadata import PredictionMetadata
from laborious.activities.storage import Storage
@@ -14,7 +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.MongoActivities.__init__')
@patch('laborious.activities.activities.ModelMetrics.__init__')
@patch('laborious.activities.activities.API.__init__')
@patch('laborious.activities.activities.MinioRepository')
@@ -24,7 +24,7 @@ def test___init__(
mock_minio_repository,
mock_api_init,
mock_model_metrics_init,
mock_prediction_metadata_init,
mock_mongo_activities_init,
mock_gates_init,
mock_opc_init,
mock_mlflow_init,
@@ -87,7 +87,7 @@ def test___init__(
assert isinstance(activities, Gates)
assert isinstance(activities, ModelMetrics)
assert isinstance(activities, API)
assert isinstance(activities, PredictionMetadata)
assert isinstance(activities, MongoActivities)
mock_storage_init.assert_called_once_with(
ANY,
@@ -131,7 +131,7 @@ def test___init__(
metrics_controller=mock_metrics_controller.return_value,
)
mock_prediction_metadata_init.assert_called_once_with(
mock_mongo_activities_init.assert_called_once_with(
ANY,
mongo_repository=mongo_repository,
logger=logger,
@@ -172,7 +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.MongoActivities')
@patch('laborious.activities.activities.ModelMetrics')
@patch('laborious.activities.activities.API')
@patch('laborious.activities.activities.MinioRepository')
@@ -180,7 +180,7 @@ def test_shutdown(
_mock_minio_repository,
mock_api_init,
mock_model_metrics_init,
mock_prediction_metadata_init,
mock_mongo_activities_init,
mock_gates_init,
mock_opc_init,
mock_mlflow_init,
@@ -243,7 +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()
mock_mongo_activities_init.close.assert_called_once()
@patch('laborious.activities.activities.SientiaMLflowRepository')
@@ -252,7 +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.MongoActivities.__init__')
@patch('laborious.activities.activities.ModelMetrics.__init__')
@patch('laborious.activities.activities.API.__init__')
@patch('laborious.activities.activities.MinioRepository')
@@ -262,7 +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_mongo_activities_init,
_mock_gates_init,
_mock_opc_init,
_mock_mlflow_init,
@@ -326,7 +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.MongoActivities.__init__')
@patch('laborious.activities.activities.ModelMetrics.__init__')
@patch('laborious.activities.activities.API.__init__')
@patch('laborious.activities.activities.MinioRepository')
@@ -336,7 +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_mongo_activities_init,
mock_gates_init,
_mock_opc_init,
_mock_mlflow_init,
@@ -398,9 +398,9 @@ 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 PredictionMetadata, not a second
# The same repository built here is the one handed to MongoActivities, not a second
# instance.
mock_prediction_metadata_init.assert_called_once_with(
mock_mongo_activities_init.assert_called_once_with(
ANY,
mongo_repository=mock_mongo_repository_cls.return_value,
logger=logger,

4
tests/laborious/activities/test_gates.py
View File

@@ -147,8 +147,8 @@ def test_input_gate_with_filter(gates_activity):
# 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.
# `MongoActivities.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()

84
tests/laborious/activities/test_model_metrics.py
View File

@@ -7,7 +7,13 @@ from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
from sientia_model.analytics.drift_analysis import DriftInsufficientDataError
from laborious.activities.model_metrics import ModelMetrics
from laborious.activities.model_metrics import (
ModelMetrics,
frequency_to_seconds,
next_adaptive_state,
recovery_runs,
threshold_breaches,
)
@fixture
@@ -1221,3 +1227,79 @@ def test_to_naive_utc_keeps_naive_timestamps():
assert result.dt.tz is None
assert result.tolist() == [Timestamp('2024-01-01 06:00:00')]
@mark.parametrize(
'rows, thresholds, expected_count',
[
([{'metric': 'rmse', 'value': 2.0}], {'rmse_max': 1.0}, 1),
([{'metric': 'rmse', 'value': 0.5}], {'rmse_max': 1.0}, 0),
([{'metric': 'r2', 'value': 0.5}], {'r2_min': 0.9}, 1),
([{'metric': 'r2', 'value': 0.95}], {'r2_min': 0.9}, 0),
([{'metric': 'rmse', 'value': 1.0}], {'rmse_max': 1.0}, 0), # equal is not a breach
([{'metric': 'r2', 'value': 0.9}], {'r2_min': 0.9}, 0),
([{'metric': 'mae', 'value': 99.0}], {'rmse_max': 1.0}, 0),
([{'metric': 'rmse', 'value': 99.0}], None, 0),
([], {'rmse_max': 1.0}, 0),
(
[{'metric': 'rmse', 'value': 2.0}, {'metric': 'r2', 'value': 0.1}],
{'rmse_max': 1.0, 'r2_min': 0.9},
2,
),
],
)
def test_threshold_breaches(rows, thresholds, expected_count):
assert len(threshold_breaches(rows, thresholds)) == expected_count
def _state(escalated: bool, streak: int) -> dict:
return {'escalated': escalated, 'healthy_streak': streak}
@mark.parametrize(
'state, breached, runs, expected',
[
({}, True, 3, _state(True, 0)), # first breach escalates
({}, False, 3, _state(False, 0)), # missing state = healthy
(_state(False, 0), False, 3, _state(False, 0)), # healthy stays healthy
(_state(True, 0), False, 3, _state(True, 1)), # healthy run while escalated
(_state(True, 1), False, 3, _state(True, 2)),
(_state(True, 2), False, 3, _state(False, 0)), # N-th healthy run recovers
(_state(True, 2), True, 3, _state(True, 0)), # breach resets the streak
(_state(True, 0), False, 1, _state(False, 0)), # N=1 recovers on first healthy run
(_state(True, None), False, 3, _state(True, 1)), # hand-edited streak restarts
(_state(True, 'x'), False, 3, _state(True, 1)),
(_state(True, float('inf')), False, 3, _state(True, 1)),
],
)
def test_next_adaptive_state(state, breached, runs, expected):
assert next_adaptive_state(state, breached, runs) == expected
@mark.parametrize(
'frequency, seconds',
[('30s', 30), ('5m', 300), ('2h', 7200), ('1d', 86400)],
)
def test_frequency_to_seconds(frequency, seconds):
assert frequency_to_seconds(frequency) == seconds
@mark.parametrize('frequency', ['', '5', 'm', '5x', 'abc', '0m'])
def test_frequency_to_seconds_rejects_invalid(frequency):
with raises(ValueError):
frequency_to_seconds(frequency)
@mark.parametrize(
'interval_minutes, frequency_alert, expected',
[
(60, '5m', 12), # 60/5 = 12, at the cap
(60, '15m', 4),
(60, '7m', 9), # ceil(8.57)
(60, '30s', 12), # 120 capped at 12
(1440, '1h', 12), # 24 capped at 12
(5, '1h', 1), # window shorter than one run: at least 1
],
)
def test_recovery_runs(interval_minutes, frequency_alert, expected):
assert recovery_runs(interval_minutes, frequency_alert) == expected

357
tests/laborious/activities/test_mongo.py Normal file
View File

@@ -0,0 +1,357 @@
from datetime import datetime
from unittest.mock import MagicMock, call, patch
from pytest import fixture, mark, raises
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from laborious.activities.mongo import MongoActivities
@fixture
def mongo() -> MagicMock:
return MagicMock()
@fixture
def prediction_metadata(mongo: MagicMock):
activity = MongoActivities(
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 = MongoActivities(
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 == MongoActivities._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):
"""``MongoActivities.close`` should close the Mongo client and clear the reference."""
mongo = MagicMock()
activity = MongoActivities(
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 = MongoActivities(
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
_SM_METADATA = {**metadata['metadata'], 'schedule_name': 'm1-simple_metrics-60m'}
_BREACH_INPUT = {
'metadata': _SM_METADATA,
'simple_metrics': [{'metric': 'rmse', 'value': 2.0}],
'thresholds': {'rmse_max': 1.0},
'interval_minutes': 60,
}
_HEALTHY_INPUT = {**_BREACH_INPUT, 'simple_metrics': [{'metric': 'rmse', 'value': 0.5}]}
_EMPTY_INPUT = {**_BREACH_INPUT, 'simple_metrics': []}
@fixture
def mongo_activities(mongo: MagicMock):
activity = MongoActivities(
mongo_repository=mongo,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=MagicMock(spec=MetricsController),
)
activity.info = MagicMock()
activity.warning = MagicMock()
return activity
def _pipeline(**fields) -> dict:
return {'schedule_name': 'm1-simple_metrics-60m', 'frequency': '1h', **fields}
def _set_of(mongo: MagicMock) -> dict:
collection, query, update, _metadata = mongo.update_one.call_args.args
assert collection == 'pipelines'
assert query == {'schedule_name': 'm1-simple_metrics-60m'}
return update['$set']
def test_adaptive_state_no_mongo_is_noop():
activity = MongoActivities(
mongo_repository=None,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=MagicMock(spec=MetricsController),
)
assert activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
def test_adaptive_state_missing_pipeline_doc_is_noop(mongo_activities, mongo):
mongo.find_one.return_value = None
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
mongo.update_one.assert_not_called()
# Likely misconfig (other MONGODB_DATABASE_NAME or schedule_name mismatch): not silent.
mongo_activities.warning.assert_called_once()
assert 'm1-simple_metrics-60m' in mongo_activities.warning.call_args.args[0]
def test_adaptive_state_without_frequency_alert_is_noop(mongo_activities, mongo):
mongo.find_one.return_value = _pipeline()
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
mongo.update_one.assert_not_called()
mongo_activities.warning.assert_not_called()
def test_adaptive_state_breach_escalates_and_bumps_updated_at(mongo_activities, mongo):
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
assert result == {'escalated': True, 'healthy_streak': 0}
update = _set_of(mongo)
assert update['adaptive_state'] == {'escalated': True, 'healthy_streak': 0}
# BSON datetime, never a string: the orchestrator calls .replace(tzinfo=...) on it.
assert isinstance(update['updated_at'], datetime)
assert update['updated_at'].tzinfo is not None
def test_adaptive_state_healthy_run_increments_streak_without_updated_at(mongo_activities, mongo):
# 60 min window / 5m = N 12, so one healthy run only moves the streak.
mongo.find_one.return_value = _pipeline(
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
)
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
assert result == {'escalated': True, 'healthy_streak': 1}
assert _set_of(mongo) == {'adaptive_state': {'escalated': True, 'healthy_streak': 1}}
def test_adaptive_state_recovers_after_window_renewed(mongo_activities, mongo):
# 60 min window / 30m = N 2: second consecutive healthy run recovers.
mongo.find_one.return_value = _pipeline(
frequency_alert='30m', adaptive_state={'escalated': True, 'healthy_streak': 1}
)
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
assert result == {'escalated': False, 'healthy_streak': 0}
update = _set_of(mongo)
assert update['adaptive_state'] == {'escalated': False, 'healthy_streak': 0}
assert isinstance(update['updated_at'], datetime)
def test_adaptive_state_empty_window_counts_as_healthy_run(mongo_activities, mongo):
# Spec decision 8: an empty/under-sampled window never escalates but counts toward recovery.
mongo.find_one.return_value = _pipeline(
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 3}
)
result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
assert result == {'escalated': True, 'healthy_streak': 4}
def test_adaptive_state_empty_window_never_escalates(mongo_activities, mongo):
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
assert result == {'escalated': False, 'healthy_streak': 0}
mongo.update_one.assert_not_called()
def test_adaptive_state_unchanged_does_not_write(mongo_activities, mongo):
mongo.find_one.return_value = _pipeline(
frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 0}
)
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
assert result == {'escalated': False, 'healthy_streak': 0}
mongo.update_one.assert_not_called()
@mark.parametrize('frequency_alert', ['5min', '0m', 5])
def test_adaptive_state_invalid_frequency_alert_is_skipped(
mongo_activities, mongo, frequency_alert
):
mongo.find_one.return_value = _pipeline(frequency_alert=frequency_alert)
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
mongo.update_one.assert_not_called()
mongo_activities.warning.assert_called_once()
@mark.parametrize('interval_minutes', ['60', None])
def test_adaptive_state_invalid_interval_minutes_is_skipped(
mongo_activities, mongo, interval_minutes
):
# Untyped Mongo config: "60" * 60 is a string, None * 60 a TypeError.
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
result = mongo_activities.update_simple_metrics_adaptive_state(
{**_BREACH_INPUT, 'interval_minutes': interval_minutes}
)
assert result is None
mongo.update_one.assert_not_called()
mongo_activities.warning.assert_called_once()
@mark.parametrize('adaptive_state', ['garbage', ['escalated']])
def test_adaptive_state_malformed_state_is_not_escalated(mongo_activities, mongo, adaptive_state):
mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state)
result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
assert result == {'escalated': True, 'healthy_streak': 0}
mongo.update_one.assert_called_once()
@mark.parametrize('failing_call', ['find_one', 'update_one'])
def test_adaptive_state_mongo_error_propagates(mongo_activities, mongo, failing_call):
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
getattr(mongo, failing_call).side_effect = RuntimeError('mongo down')
with raises(RuntimeError, match='mongo down'):
mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)

179
tests/laborious/activities/test_prediction_metadata.py
View File

@@ -1,179 +0,0 @@
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

6
tests/laborious/worker/test_worker.py
View File

@@ -16,6 +16,7 @@ def _build_fake_activities():
inst.export_data_to_postgres = MagicMock()
inst.load_custom_query = MagicMock()
inst.calculate_simple_metrics = MagicMock()
inst.update_simple_metrics_adaptive_state = MagicMock()
inst.get_reference_data = MagicMock()
inst.calculate_drift = MagicMock()
inst.request_predict = MagicMock()
@@ -188,6 +189,11 @@ async def test_main_success_exit_zero(monkeypatch):
# Drift and SimpleMetrics load no model code, so every runtime shares the `core` queues.
assert prepare_calls[1].kwargs['runtime'] == 'core'
assert prepare_calls[2].kwargs['runtime'] == 'core'
# prepare_calls[1] is the SimpleMetrics worker (order: MinimalRetrain, SimpleMetrics, Drift, PredictionsBatch).
assert (
fake_activities.update_simple_metrics_adaptive_state
in prepare_calls[1].kwargs['activities']
)
@mark.asyncio

148
tests/laborious/workflows/test_simple_metrics.py
View File

@@ -1,6 +1,6 @@
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from pytest import fixture, mark, raises
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
from laborious.activities.activities import Activities
@@ -42,12 +42,17 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics):
target_data = {'data': 'test_target_data'}
simple_metrics_data = {'metrics': 'test_simple_metrics_data'}
workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None])
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
# Act
await simple_metrics.run(input_data)
# Assert - load, export and adaptive state (no thresholds still updates it)
assert workflow_mock.execute_activity_method.call_count == 3
workflow_mock.logger.warning.assert_not_called()
# Assert - Check load_custom_query call
# Query format matches psycopg2.sql output (identifiers with double quotes, literals with single quotes)
expected_query = f"""
@@ -130,13 +135,22 @@ async def test_run_empty_target_data(workflow_mock: AsyncMock, simple_metrics: S
target_data = None
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(return_value=target_data)
workflow_mock.execute_local_activity_method = AsyncMock()
# Act
await simple_metrics.run(input_data)
# Assert - Should not call calculate_simple_metrics or export
assert workflow_mock.execute_activity_method.call_count == 1
# Assert - No calculate or export; load + adaptive state (empty window = healthy run)
workflow_mock.execute_local_activity_method.assert_not_called()
calls = workflow_mock.execute_activity_method.call_args_list
assert [c.args[0] for c in calls] == [
Activities.load_custom_query,
Activities.update_simple_metrics_adaptive_state,
]
assert calls[1].args[1]['simple_metrics'] == []
workflow_mock.logger.warning.assert_not_called()
@mark.asyncio
@@ -159,15 +173,22 @@ async def test_run_empty_simple_metrics(workflow_mock: AsyncMock, simple_metrics
target_data = {'data': 'test_target_data'}
simple_metrics_data = None
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(return_value=target_data)
workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
# Act
await simple_metrics.run(input_data)
# Assert - Should call calculate_simple_metrics but not export
workflow_mock.execute_activity_method.assert_called_once()
# Assert - calculate returns None: no export, adaptive state gets [] (the `or []` is live)
workflow_mock.execute_local_activity_method.assert_called_once()
calls = workflow_mock.execute_activity_method.call_args_list
assert [c.args[0] for c in calls] == [
Activities.load_custom_query,
Activities.update_simple_metrics_adaptive_state,
]
assert calls[1].args[1]['simple_metrics'] == []
workflow_mock.logger.warning.assert_not_called()
@mark.asyncio
@@ -190,12 +211,17 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim
target_data = {'data': 'test_target_data'}
simple_metrics_data = {'metrics': 'test_simple_metrics_data'}
workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None])
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
# Act
await simple_metrics.run(input_data)
# Assert - all three activities ran
assert workflow_mock.execute_activity_method.call_count == 3
workflow_mock.logger.warning.assert_not_called()
# Assert - Check calculate_simple_metrics call with default metrics
workflow_mock.execute_activity_method.assert_any_call(
Activities.load_custom_query,
@@ -245,7 +271,7 @@ _METRICS = [{'metric': 'rmse', 'value': 9.0}]
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metrics, n_rows):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * n_rows, None])
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * n_rows, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
await simple_metrics.run(
@@ -254,8 +280,8 @@ async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metr
activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
assert activity_input['thresholds'] is None
# Persist, don't alert.
assert workflow_mock.execute_activity_method.call_count == 2
# Persist, don't alert: load, export, adaptive state.
assert workflow_mock.execute_activity_method.call_count == 3
export_call = workflow_mock.execute_activity_method.call_args_list[1]
assert export_call.args[0] == Activities.export_data_to_postgres
assert export_call.args[1]['data'] == _METRICS
@@ -265,7 +291,7 @@ async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metr
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None])
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
await simple_metrics.run(
@@ -274,7 +300,52 @@ async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics
activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
assert activity_input['thresholds'] == {'rmse_max': 1.0}
assert workflow_mock.execute_activity_method.call_count == 2
assert workflow_mock.execute_activity_method.call_count == 3
_BREACH_ROWS = [{'metric': 'rmse', 'value': 9.0}]
_THRESHOLDS = {'rmse_max': 1.0}
@mark.asyncio
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_updates_adaptive_state_after_export(workflow_mock, simple_metrics):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(
side_effect=[[_ROW] * 3, None, {'escalated': True, 'healthy_streak': 0}]
)
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
calls = workflow_mock.execute_activity_method.call_args_list
assert calls[1].args[0] is Activities.export_data_to_postgres
assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
assert calls[2].args[1] == {
**metadata,
'simple_metrics': _BREACH_ROWS,
'thresholds': _THRESHOLDS,
'interval_minutes': 60,
}
@mark.asyncio
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_without_thresholds_still_updates_adaptive_state(workflow_mock, simple_metrics):
# No thresholds = never a breach; still calling lets an escalated doc recover after the
# thresholds are removed.
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 3, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
await simple_metrics.run(_input())
calls = workflow_mock.execute_activity_method.call_args_list
assert len(calls) == 3
assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
assert calls[2].args[1]['thresholds'] is None
assert calls[2].args[1]['simple_metrics'] == _BREACH_ROWS
workflow_mock.logger.warning.assert_not_called()
@mark.asyncio
@@ -284,7 +355,7 @@ async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_guard_off_keeps_thresholds(workflow_mock, simple_metrics, min_samples, warns):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None])
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
await simple_metrics.run(
@@ -294,3 +365,54 @@ async def test_run_guard_off_keeps_thresholds(workflow_mock, simple_metrics, min
activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
assert activity_input['thresholds'] == {'rmse_max': 1.0}
assert workflow_mock.logger.warning.call_count == int(warns)
@mark.asyncio
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_empty_window_counts_as_healthy_run(workflow_mock, simple_metrics):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[], None])
workflow_mock.execute_local_activity_method = AsyncMock()
await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
workflow_mock.execute_local_activity_method.assert_not_called()
calls = workflow_mock.execute_activity_method.call_args_list
assert len(calls) == 2 # load + adaptive state, no export
assert calls[1].args[0] is Activities.update_simple_metrics_adaptive_state
assert calls[1].args[1]['simple_metrics'] == []
@mark.asyncio
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_below_min_samples_sends_no_rows_to_adaptive_state(workflow_mock, simple_metrics):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
await simple_metrics.run(
_input(simple_metrics_thresholds=_THRESHOLDS, simple_metrics_min_samples=5)
)
calls = workflow_mock.execute_activity_method.call_args_list
assert calls[1].args[0] is Activities.export_data_to_postgres # metrics still persisted
assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
assert calls[2].args[1]['simple_metrics'] == [] # never a breach
@mark.asyncio
@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
async def test_run_adaptive_state_failure_fails_workflow(workflow_mock, simple_metrics):
workflow_mock.logger = MagicMock()
workflow_mock.execute_activity_method = AsyncMock(
side_effect=[[_ROW] * 3, None, RuntimeError('dispatch failed')]
)
workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
input_data = _input(simple_metrics_thresholds=_THRESHOLDS)
with raises(RuntimeError, match='dispatch failed'):
await simple_metrics.run(input_data)
# Metrics were exported before the failing call
calls = workflow_mock.execute_activity_method.call_args_list
assert calls[1].args[0] is Activities.export_data_to_postgres