diff --git a/docs/E2E_TEST_REPORT.md b/docs/E2E_TEST_REPORT.md new file mode 100644 index 0000000..1d8087a --- /dev/null +++ b/docs/E2E_TEST_REPORT.md @@ -0,0 +1,149 @@ +# E2E test run report + +**Date:** 2026-05-08 +**Command:** `source venv/bin/activate && rtk pytest e2e/ -v --tb=short` +**Environment:** Linux, Python 3.11.15, pytest 9.0.3 + +## Summary + +| Metric | Count | +|--------|------:| +| Collected | 43 | +| **Passed** | **37** | +| **Failed** | **6** | + +Full pytest output (compressed by `rtk`) was written to: + +`~/.local/share/rtk/tee/1778268193_pytest.log` + +--- + +## Failed tests (6) + +1. `e2e/test_drift.py::test_drift_happy_path_persists_all_columns_with_reference_data` +2. `e2e/test_drift.py::test_drift_uses_30pct_fallback_when_reference_unavailable` +3. `e2e/test_drift.py::test_drift_chunk_period_seconds_preserves_seconds_in_chunk_start_date` +4. `e2e/test_predictions_batch_prediction_process.py::test_scenario_2_1_3_input_gate_triggers_repeat` +5. `e2e/test_predictions_batch_prediction_process.py::test_scenario_2_2_3_transform_gate_triggers_repeat` +6. `e2e/test_predictions_batch_prediction_process.py::test_scenario_2_3_3_predict_gate_triggers_repeat` + +**Follow-up:** Jensen–Shannon NULLs for the drift happy-path scenario are fully traced (histogram out-of-range + `density=True`, vs NannyML’s leftover bin) in [DRIFT_JS_NULL_INVESTIGATION.md](./DRIFT_JS_NULL_INVESTIGATION.md). + +--- + +## Failure group A — Drift: `drift_metrics.value` NOT NULL (2 tests) + +### Error + +`psycopg2.errors.NotNullViolation`: null value in column `value` of relation `sientia_data.drift_metrics` violates not-null constraint. + +Example failing row (from logs): `feature=sensor_2`, `method=jensen_shannon`, `value=null`, with `kolmogorov_smirnov` / `wasserstein` populated for the same chunk. + +The bulk INSERT built by `Activities.export_data_to_postgres` includes parameters such as `'value__4': None` for `jensen_shannon` on a given chunk. + +### Root cause + +`calculate_drift` (real `DriftAnalysis` + `ModelMetrics.get_drift_metrics`) can emit **NaN / missing** values for some metric methods (here **Jensen–Shannon**) on some features/chunks. Pandas/SQLAlchemy turns that into SQL `NULL`, while the E2E schema (mirroring production) defines: + +```sql +value numeric NOT NULL +``` + +in `e2e/db_schema.sql` for `sientia_data.drift_metrics`. + +### Recommended fixes (pick one consistent with product rules) + +1. **Application layer (preferred if NULLs are never valid in production):** Before export, sanitize the drift dataframe — e.g. drop rows where `value` is null/NaN, or replace with a defined sentinel (only if product agrees), or skip emitting that method row when the statistic is undefined. +2. **Analytics layer:** Harden the Jensen–Shannon (and similar) paths so they always return a finite float for the supported inputs, or explicitly map “undefined” to an agreed numeric convention. +3. **Schema (only if product allows missing metrics):** Align DDL with reality by making `value` nullable — **only** if production and downstream consumers already expect missing metrics; the E2E comment in `test_drift.py` suggests `feature`/`timestamp` nullable cases exist, but `value` is still listed in `NON_NULL_DRIFT_COLUMNS`. + +### Tests affected + +- `test_drift_happy_path_persists_all_columns_with_reference_data` +- `test_drift_uses_30pct_fallback_when_reference_unavailable` + +--- + +## Failure group B — Drift: chunk period seconds (1 test) + +### Error + +`AssertionError: expected at least one drift row to be persisted` (`e2e/test_drift.py:493`). + +The workflow run completed without failing the test via `WorkflowFailureError`, but **no rows** were found in `sientia_data.drift_metrics` for the model. + +### Likely cause + +In `Drift.run`, persistence runs only when `if drift_data:` is truthy (`laborious/workflows/drift.py`). An **empty** drift result skips `export_data_to_postgres`, so the table stays empty. + +Probable reasons: + +- With **`chunk_period='s'`** and only **three** target rows (30 s spacing), `calculate_drift` / `DriftAnalysis` may produce **no output rows** (insufficient data per chunk or internal filters). +- Less likely here: time-window mismatch — timestamps are built from `datetime.now(UTC)` with `interval: 60` minutes from `drift_base.json`, so data should still fall in the window. + +### Recommended fixes + +1. **Test data:** Increase the number of second-spaced points (and/or span multiple chunk boundaries) so the analyzer reliably emits at least one chunk row. +2. **Product code:** If sub-minute chunking is required to always produce metrics when any data exists, adjust `ModelMetrics` / `DriftAnalysis` integration for small-N second buckets. +3. **Diagnostics:** Run the same scenario with `pytest -s` and confirm logs for “empty `drift_data`” vs export errors. + +Solução a ser aplicada: +Dividir em dois testes: + +1. Teste com dados suficientes para produzir pelo menos uma linha de drift +2. Teste com dados insuficientes para produzir pelo menos uma linha de drift, mas ja esperando os erros e validando que nao foi persistido nada + +### Test affected + +- `test_drift_chunk_period_seconds_preserves_seconds_in_chunk_start_date` + +--- + +## Failure group C — Prediction REPEAT path: unique constraint on predictions (3 tests) + +### Error + +`psycopg2.errors.UniqueViolation`: duplicate key value violates unique constraint **`unique_model_id_timestamp`** on `sientia_data.predictions` (`model_id`, `timestamp`). + +### Root cause + +REPEAT is handled by `Activities.repeat_last_prediction` (registered from `sientia_do.temporal.activities.postgres_sync.Postgres`, via `Storage` inheritance in `laborious/activities/storage.py`). The E2E tests seed a prior row with a **fixed** timestamp: + +```python +# e2e/test_predictions_batch_prediction_process.py — insert_sample_prediction +VALUES ({model_id}, '2024-01-01 12:00:00+00:00', ...) +``` + +The helper `assert_repeat` expects **two** rows with the same `(model_id, prediction, prediction_confidence, prediction_status)` but compares **only** those columns — not `timestamp` (`e2e/helpers.py`). So the intended behavior is: **duplicate business payload**, not necessarily **duplicate primary unique key** `(model_id, timestamp)`. + +If `repeat_last_prediction` **INSERT**s a copy using the **same** `timestamp` as the last prediction, Postgres correctly rejects the second insert. + +### Recommended fixes + +1. **Implement REPEAT as UPSERT:** Use `ON CONFLICT (model_id, timestamp) DO UPDATE` (or the project’s existing `export_data_to_postgres` `on_conflict` pattern used in `FormatAndExportPrediction`) when writing the repeated prediction — if the product definition of REPEAT is “refresh same logical slot.” +2. **Insert with the current batch timestamp:** Copy numeric/status fields from the last row but set `timestamp` to the **new** batch instant (e.g. the workflow’s `last_timestamp` / slice timestamp). This matches `assert_repeat`, which does not assert on `timestamp`. +3. **Test-only change (weakest):** Relax assertions or change seed data — only if production behavior is “duplicate key is expected” (unlikely). + +Solução a ser aplicada: +Alterar o teste para usar o timestamp da ultima execucao do batch, ao inves do timestamp do primeiro batch. +2. **Insert with the current batch timestamp:** Copy numeric/status fields from the last row but set `timestamp` to the **new** batch instant (e.g. the workflow’s `last_timestamp` / slice timestamp). This matches `assert_repeat`, which does not assert on `timestamp`. + +### Tests affected + +- `test_scenario_2_1_3_input_gate_triggers_repeat` +- `test_scenario_2_2_3_transform_gate_triggers_repeat` +- `test_scenario_2_3_3_predict_gate_triggers_repeat` + +--- + +## Passing areas (sanity check) + +All scenarios in `test_child_workflows_e2e.py`, `test_minimal_retrain.py`, `test_minio_offload.py`, `test_predictions_batch_format_export.py`, `test_predictions_batch_main_workflow.py` (except the three REPEAT cases above), and `test_simple_metrics.py` **passed** in this run. + +--- + +## Suggested order of work + +1. Fix **drift `value` NULL** — unblocks two high-value drift E2Es and may clarify the chunk-seconds scenario if exports start succeeding consistently. +2. Fix **`repeat_last_prediction` uniqueness** — unblocks three prediction-process E2Es; implementation likely lives in **`sientia_do`** Postgres activities, not in this repo. +3. Revisit **`test_drift_chunk_period_seconds`** data volume / expectations after drift export is stable. diff --git a/docs/sientia_model_drift_jensen_shannon_change.md b/docs/sientia_model_drift_jensen_shannon_change.md new file mode 100644 index 0000000..099ca36 --- /dev/null +++ b/docs/sientia_model_drift_jensen_shannon_change.md @@ -0,0 +1,55 @@ +# Specification: `sientia_model` — Jensen–Shannon drift (`DriftAnalysis`) + +This document describes what **`sientia_model.analytics.drift_analysis.DriftAnalysis`** should change so downstream consumers (e.g. Laborious `calculate_drift` → Postgres `drift_metrics.value NOT NULL`) no longer receive **NaN** for Jensen–Shannon on valid finite data. + +## Scope + +- **File:** `sientia_model/analytics/drift_analysis.py` +- **Method:** `_jensen_shannon_distance(self, ref: np.ndarray, cur: np.ndarray, bins: int = 20) -> float` +- **Callers:** `detect_univariate_drift` uses this for the `jensen_shannon` method; results are written to `value` in the drift dataframe. + +## Problem + +The current implementation: + +1. Builds bin edges from **`ref`** only: `np.histogram(ref, bins=bins, density=True)`. +2. Builds the chunk histogram with **`density=True`** on the same edges: `np.histogram(cur, bins=edges, density=True)`. + +When **every** value in **`cur`** falls **outside** the closed support implied by those edges (typical case: production chunk drifted above the reference max or below the reference min), NumPy yields **all-zero counts** for `cur`. With **`density=True`**, normalization does **0/0**, producing **NaN** for the whole histogram, which propagates to **`float('nan')`** in `detect_univariate_drift` → **SQL NULL** where `value` is `NOT NULL`. + +This appears in Laborious when: + +- `model_config.target` excludes the main target column from univariate features, so another feature (e.g. `sensor_2`) is compared chunk-by-chunk against the full reference series for that feature. +- Reference and current ranges do not overlap for some chunks (strong drift or different scaling). + +Other methods (`kolmogorov_smirnov`, `wasserstein`) do not use this histogram+density path, so they can stay finite while **Jensen–Shannon** alone becomes null. + +## Required behavior + +1. **Finite output** for finite `ref` and `cur` after removing non-finite values, whenever both sides have **at least one** usable sample. +2. **Explicit handling of out-of-range chunk mass:** probability mass from `cur` that does not fall into any bin defined from `ref` must still be represented (so the chunk distribution sums to 1), analogous to NannyML’s continuous JS approach (tail / “leftover” mass). +3. **Missing values:** drop `NaN` from `ref` and `cur` before computing. If either side is **empty** after that, return **`float('nan')`** (callers may filter or map; schema may still forbid null — product decision outside this spec). + +## Recommended algorithm (replace current body) + +1. `ref = np.asarray(ref, float); cur = np.asarray(cur, float)`. +2. `ref = ref[~np.isnan(ref)]; cur = cur[~np.isnan(cur)]`. +3. If `ref.size == 0` or `cur.size == 0`: return `float('nan')`. +4. `hist_ref, edges = np.histogram(ref, bins=bins)` — **counts**, not `density=True`. +5. `p = hist_ref.astype(float) / ref.size` (reference bin probabilities). +6. `hist_cur, _ = np.histogram(cur, bins=edges)`; `q = hist_cur.astype(float) / cur.size`. +7. `leftover = 1.0 - float(np.sum(q))`. If `leftover > 1e-15` (tolerance for float noise), append **`leftover`** to `q` and **`0.0`** to `p` so both remain proper discrete distributions over the same extended support. +8. Apply small smoothing (existing module constant `EPSILON` is fine): add `EPSILON` to `p` and `q`, renormalize each to sum 1. +9. `m = 0.5 * (p + q)`; compute symmetric JS via KL terms as today, e.g. `inner = 0.5 * (sum(p*log(p/m)) + sum(q*log(q/m)))`. +10. Return `sqrt(max(inner, 0.0))` to guard against tiny negative `inner` from floating-point error. + +## Non-goals / notes + +- **Numerical parity** with the old `density=True` implementation is not required; parity with **NannyML** or **scipy** JS is desirable but optional. The priority is **finite, interpretable** drift when the chunk is outside the reference histogram range. +- **Multivariate** drift in the same file is unchanged by this spec. +- **Tests** in `sientia_model` should cover: (a) chunk entirely above reference max, (b) entirely below reference min, (c) overlapping range, (d) `ref` or `cur` all-NaN after cleaning. + +## Reference (external) + +- NannyML continuous JS uses count-based bin probabilities and a **leftover** mass bin; see `ContinuousJensenShannonDistance` in `nannyml/drift/univariate/methods.py` (`_calculate`, `leftover = 1 - np.sum(...)`). +- Historical NaN-in-reference issue: [NannyML#339](https://github.com/NannyML/nannyml/issues/339) / [#340](https://github.com/NannyML/nannyml/pull/340) (orthogonal to out-of-range mass, but relevant for input cleaning). diff --git a/e2e/conftest.py b/e2e/conftest.py index e66febc..123cf4a 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -1,109 +1,20 @@ """Pytest configuration and fixtures for E2E tests.""" from concurrent.futures import ThreadPoolExecutor -from typing import Any +from pathlib import Path from unittest.mock import MagicMock import pandas as pd import pytest import pytest_asyncio -from sqlalchemy import create_engine, text +from sqlalchemy import create_engine from testcontainers.core.container import DockerContainer from testcontainers.minio import MinioContainer from testcontainers.postgres import PostgresContainer from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker - -class _FakeModelAnalysis: - """ - Configurable double for ``sientia_model.analytics.model_analysis.ModelAnalysis`` - used by E2E drift tests. - - Replaces the real analyzer in ``sientia_model.analytics.model_analysis`` - BEFORE the production import chain runs so - ``laborious.activities.model_metrics`` resolves ``ModelAnalysis`` to this - class at import time. Keeps drift assertions stable across runs (the real - analyzer is data-dependent). - - Tests configure responses through class-level attributes which are reset - between tests by ``reset_model_analysis_stub``: - - - ``_drift_response_factory``: callable ``(univariate, multivariate) -> DataFrame`` - controlling the consolidated drift dataframe seen by ``calculate_drift``. - - ``_univariate_side_effect`` / ``_multivariate_side_effect``: optional - side effects (Exception or callable) for the raw detection methods. - - ``_drift_metrics_dataframe_exception``: when set, raised by - ``get_drift_metrics_dataframe`` to simulate analyzer failures. - """ - - _drift_response_factory: Any = None - _univariate_side_effect: Any = None - _multivariate_side_effect: Any = None - _drift_metrics_dataframe_exception: Exception | None = None - last_instance: Any = None - - def __init__(self, config): - self.config = config - type(self).last_instance = self - self.detect_univariate_drift_calls = [] - self.detect_multivariate_drift_calls = [] - self.get_drift_metrics_dataframe_calls = [] - - def detect_univariate_drift(self, **kwargs): - """Record arguments and return ``{}`` unless ``_univariate_side_effect`` overrides it.""" - self.detect_univariate_drift_calls.append(kwargs) - side_effect = type(self)._univariate_side_effect - if isinstance(side_effect, Exception): - raise side_effect - if callable(side_effect): - return side_effect(**kwargs) - return {} - - def detect_multivariate_drift(self, **kwargs): - """Record arguments and return ``{}`` unless ``_multivariate_side_effect`` overrides it.""" - self.detect_multivariate_drift_calls.append(kwargs) - side_effect = type(self)._multivariate_side_effect - if isinstance(side_effect, Exception): - raise side_effect - if callable(side_effect): - return side_effect(**kwargs) - return {} - - def get_drift_metrics_dataframe(self, univariate_drift, multivariate_drift): - """Return the configured drift dataframe (copy) or raise the configured exception.""" - self.get_drift_metrics_dataframe_calls.append( - {'univariate_drift': univariate_drift, 'multivariate_drift': multivariate_drift} - ) - if type(self)._drift_metrics_dataframe_exception is not None: - raise type(self)._drift_metrics_dataframe_exception - factory = type(self)._drift_response_factory - if factory is None: - return pd.DataFrame() - result = factory(univariate_drift, multivariate_drift) - return result.copy() if isinstance(result, pd.DataFrame) else result - - @classmethod - def reset(cls): - """Clear all configured side effects and the last constructed instance.""" - cls._drift_response_factory = None - cls._univariate_side_effect = None - cls._multivariate_side_effect = None - cls._drift_metrics_dataframe_exception = None - cls.last_instance = None - - -# Patch ``sientia_model.analytics.model_analysis.ModelAnalysis`` BEFORE the -# production import chain runs so drift E2E tests can drive deterministic -# analyzer outputs (the real implementation is data-dependent and would yield -# values that drift across runs). The patch is applied to the real installed -# module so that ``from laborious.activities.activities import Activities`` -# resolves ``ModelAnalysis`` to ``_FakeModelAnalysis`` at import time. -import sientia_model.analytics.model_analysis as _sientia_model_analysis_module # noqa: E402 - -_sientia_model_analysis_module.ModelAnalysis = _FakeModelAnalysis - -from laborious.activities.activities import Activities # noqa: E402 +from laborious.activities.activities import Activities from laborious.workflows.drift import Drift from laborious.workflows.minimal_retrain import MinimalRetrain from laborious.workflows.predictions_batch import PredictionsBatch @@ -116,6 +27,11 @@ from sientia_do.notifications.handlers import CoreNotificationHandler from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController +# Single source of truth for the test database schema. Mirrors the production +# DDL for ``sientia_data`` so any production change can be pasted directly into +# this file (see ``e2e/db_schema.sql``) without touching Python. +DB_SCHEMA_SQL_PATH = Path(__file__).parent / 'db_schema.sql' + @pytest_asyncio.fixture(scope='session') def postgres_container(): @@ -153,113 +69,16 @@ def postgres_engine(postgres_container): def _create_schema_and_tables(engine): - """Create all schemas/tables required by workflow and activity paths.""" + """ + Create all schemas/tables required by workflow and activity paths. + + Loads the DDL from ``e2e/db_schema.sql`` (single source of truth that + mirrors the production schema). The SQL file is executed via the raw + DBAPI cursor so multi-statement DDL is supported. + """ + sql_text = DB_SCHEMA_SQL_PATH.read_text(encoding='utf-8') with engine.begin() as conn: - conn.execute(text('CREATE SCHEMA IF NOT EXISTS predictions_schema')) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.laborious_data ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - variable text NOT NULL, - value numeric NULL, - "timestamp" timestamptz NOT NULL, - created_at timestamptz NOT NULL, - PRIMARY KEY (id) - ); - """ - ) - ) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.predictions ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - prediction numeric NULL, - prediction_confidence numeric NOT NULL, - response_time numeric NOT NULL, - prediction_status text NOT NULL, - "timestamp" timestamptz NOT NULL, - created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - comments text NULL, - PRIMARY KEY (id, created_at) - ); - """ - ) - ) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.transformed_data ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - variable text NOT NULL, - value numeric NULL, - "timestamp" timestamptz NOT NULL, - created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - PRIMARY KEY (id) - ); - """ - ) - ) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.drift ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - feature text NOT NULL, - method text NOT NULL, - value numeric NULL, - drift bool NOT NULL, - chunk int4 NOT NULL, - "timestamp" timestamptz NOT NULL, - timestamp_end text NULL, - accurate bool NOT NULL, - created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - updated_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - PRIMARY KEY (id) - ); - """ - ) - ) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.simple_metrics_data ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - metric text NOT NULL, - value numeric NOT NULL, - "timestamp" timestamptz NOT NULL, - data_size int4 NOT NULL, - interval_minutes int4 NOT NULL, - created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - PRIMARY KEY (id) - ); - """ - ) - ) - conn.execute( - text( - """ - CREATE TABLE IF NOT EXISTS predictions_schema.retrain_reports ( - id SERIAL NOT NULL, - model_id int4 NOT NULL, - model_name text NOT NULL, - "timestamp" text NOT NULL, - status text NOT NULL, - version text NULL, - mlflow_run_id text NULL, - mlflow_experiment_id text NULL, - created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - PRIMARY KEY (id) - ); - """ - ) - ) + conn.exec_driver_sql(sql_text) @pytest_asyncio.fixture(autouse=True) @@ -269,14 +88,6 @@ def setup_postgres_schema_and_tables(postgres_engine): yield -@pytest.fixture(autouse=True) -def reset_model_analysis_stub(): - """Reset class-level state on the ModelAnalysis stub between tests.""" - _FakeModelAnalysis.reset() - yield - _FakeModelAnalysis.reset() - - @pytest_asyncio.fixture def mock_logger(): """Logger double with readable console output for E2E runs.""" @@ -650,36 +461,3 @@ async def temporal_worker_minimal_retrain(temporal_test_env, test_activities): yield worker -@pytest.fixture -def model_analysis_stub(): - """ - Expose the ``_FakeModelAnalysis`` class so drift tests can configure responses. - - Use class-level attributes to control the analyzer outputs: - - - ``stub.set_drift_dataframe(factory)`` to provide rows for ``calculate_drift``. - - ``stub.raise_on_get_drift_metrics_dataframe(exception)`` to simulate failures. - """ - - class _Helper: - """Thin convenience wrapper around _FakeModelAnalysis class state.""" - - cls = _FakeModelAnalysis - - def set_drift_dataframe(self, factory): - self.cls._drift_response_factory = factory - - def raise_on_get_drift_metrics_dataframe(self, exc: Exception): - self.cls._drift_metrics_dataframe_exception = exc - - def set_univariate_side_effect(self, side_effect): - self.cls._univariate_side_effect = side_effect - - def set_multivariate_side_effect(self, side_effect): - self.cls._multivariate_side_effect = side_effect - - @property - def last_instance(self): - return self.cls.last_instance - - return _Helper() diff --git a/e2e/db_schema.sql b/e2e/db_schema.sql new file mode 100644 index 0000000..9057815 --- /dev/null +++ b/e2e/db_schema.sql @@ -0,0 +1,109 @@ +-- ============================================================================= +-- E2E test database schema for the ``sientia_data`` namespace. +-- +-- Mirrors the production DDL one-to-one so any change in production can be +-- pasted directly into this file. The conftest fixture loads this SQL into the +-- testcontainers Postgres before each test run. +-- +-- Notes on differences from production: +-- * Tables that are partitioned in production (e.g. ``simple_metrics``, +-- ``transformed_data``, ``drift_metrics``) are created as plain tables +-- here because the test suite does not exercise partition pruning. +-- * Indexes are intentionally omitted; tests rely on functional behavior, +-- not query plans. +-- ============================================================================= + +CREATE SCHEMA IF NOT EXISTS sientia_data; + +-- ----------------------------------------------------------------------------- +-- sientia_data.laborious_data +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.laborious_data ( + model_id int4 NOT NULL, + variable text NOT NULL, + value numeric NULL, + "timestamp" timestamptz NOT NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + CONSTRAINT unique_timestamp_variable + UNIQUE (model_id, "timestamp", variable) +); + +-- ----------------------------------------------------------------------------- +-- sientia_data.predictions +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.predictions ( + model_id int4 NOT NULL, + prediction numeric NULL, + prediction_confidence numeric NOT NULL, + response_time numeric NOT NULL, + prediction_status text NOT NULL, + "timestamp" timestamptz NOT NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + "comments" text NULL, + CONSTRAINT unique_model_id_timestamp + UNIQUE (model_id, "timestamp") +); + +-- ----------------------------------------------------------------------------- +-- sientia_data.transformed_data +-- Production: PARTITION BY RANGE (created_at). Tests use a plain table. +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.transformed_data ( + id SERIAL NOT NULL, + model_id int4 NOT NULL, + variable text NOT NULL, + value numeric NULL, + "timestamp" timestamptz NOT NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + PRIMARY KEY (id, created_at) +); + +-- ----------------------------------------------------------------------------- +-- sientia_data.drift_metrics +-- Production: PARTITION BY RANGE (created_at). Tests use a plain table. +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.drift_metrics ( + id SERIAL NOT NULL, + model_id text NOT NULL, + feature text NULL, + method text NOT NULL, + value numeric NOT NULL, + alert bool NOT NULL, + chunk_index int4 NOT NULL, + chunk_start_date text NOT NULL, + chunk_end_date text NOT NULL, + accurate bool NOT NULL, + "timestamp" timestamptz NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + PRIMARY KEY (id, created_at) +); + +-- ----------------------------------------------------------------------------- +-- sientia_data.simple_metrics +-- Production: PARTITION BY RANGE (created_at). Tests use a plain table. +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.simple_metrics ( + id SERIAL NOT NULL, + model_id text NOT NULL, + metric text NOT NULL, + value numeric NOT NULL, + "timestamp" timestamptz NULL, + data_size int4 NOT NULL, + interval_minutes int4 NOT NULL, + created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, + PRIMARY KEY (id, created_at) +); + +-- ----------------------------------------------------------------------------- +-- sientia_data.log_retrain +-- No primary key in production; all columns nullable. +-- ----------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS sientia_data.log_retrain ( + mlflow_experiment_id int8 NULL, + mlflow_run_id text NULL, + model_id text NULL, + model_name text NULL, + status text NULL, + "timestamp" timestamptz NULL, + "version" text NULL +); diff --git a/e2e/helpers.py b/e2e/helpers.py index 0ada308..a0ba6be 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -9,7 +9,6 @@ from decimal import Decimal from pathlib import Path from typing import Any -import pandas as pd from sqlalchemy import text from sqlalchemy.engine import Engine @@ -91,14 +90,14 @@ def insert_sample_data(postgres_engine: Engine, model_id: int, values: list[Any] values: Per-sensor values; use string 'NULL' for SQL NULL. """ with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) values_sql = [] for i, value in enumerate(values): values_sql.append(f""" ({model_id}, 'sensor_{i + 1}', {value}, '2024-01-01 12:00:00+00:00', '2024-01-01 12:00:00+00:00') """) insert_sql = f""" - INSERT INTO predictions_schema.laborious_data (model_id, variable, value, timestamp, created_at) + INSERT INTO sientia_data.laborious_data (model_id, variable, value, timestamp, created_at) VALUES {', '.join(values_sql)} """ @@ -130,7 +129,7 @@ def assert_prediction( result_query = conn.execute( text( f'SELECT model_id, prediction, prediction_confidence, prediction_status, comments ' - f'FROM predictions_schema.predictions WHERE model_id = {model_id} ' + f'FROM sientia_data.predictions WHERE model_id = {model_id} ' f'ORDER BY created_at ASC' ) ) @@ -159,7 +158,7 @@ def assert_continue( result_query = conn.execute( text( f'SELECT model_id, prediction, prediction_confidence, prediction_status, comments ' - f'FROM predictions_schema.predictions WHERE model_id = {model_id}' + f'FROM sientia_data.predictions WHERE model_id = {model_id}' ) ) prediction_rows = result_query.fetchall() @@ -179,7 +178,7 @@ def assert_stop(postgres_engine: Engine, model_id: int) -> None: with postgres_engine.connect() as conn: result_query = conn.execute( - text(f'SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = {model_id}') + text(f'SELECT COUNT(*) FROM sientia_data.predictions WHERE model_id = {model_id}') ) count = result_query.scalar() assert count == 0, f'Expected no predictions, but found {count} records' @@ -202,7 +201,7 @@ def assert_repeat(postgres_engine: Engine, model_id: int, last_prediction: tuple result_query = conn.execute( text( f'SELECT model_id, prediction, prediction_confidence, prediction_status ' - f'FROM predictions_schema.predictions WHERE model_id = {model_id} ' + f'FROM sientia_data.predictions WHERE model_id = {model_id} ' f'ORDER BY created_at ASC' ) ) @@ -255,97 +254,15 @@ def insert_target_data_for_drift( with postgres_engine.begin() as conn: conn.execute( - text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}') + text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}') ) if rows_sql: conn.execute( text( - 'INSERT INTO predictions_schema.laborious_data ' + 'INSERT INTO sientia_data.laborious_data ' '(model_id, variable, value, "timestamp", created_at) VALUES ' + ', '.join(rows_sql) ) ) -def build_drift_dataframe( - timestamps: list[str], - features: list[str], - methods: list[str], - statistic: float = 1.0, - drift_flags: dict[tuple[str, str], bool] | None = None, - include_multivariate: bool = True, - multivariate_value: float = 16.0, - multivariate_drift: bool = True, - extra_rows: list[dict[str, Any]] | None = None, -) -> Any: - """ - Build a deterministic dataframe matching the schema returned by - ``sientia_model.analytics.model_analysis.ModelAnalysis.get_drift_metrics_dataframe`` - so drift E2E tests can pin the exact rows persisted to PostgreSQL. - - The output mirrors the analyzer's canonical schema: - ``timestamp, feature, metric, statistic, p_value, alert, chunk_index, - chunk_start_date, chunk_end_date``. ``calculate_drift`` then renames - ``alert -> drift``, ``chunk_index -> chunk``, ``chunk_end_date -> - timestamp_end`` and drops ``p_value`` / ``chunk_start_date``. - - Args: - - timestamps: Truncated chunk start timestamps (``'2024-01-01 12:00'`` for ``min``). - - features: Univariate feature names (one row per feature/method/timestamp). - - methods: Univariate methods such as ``kolmogorov_smirnov``. - - statistic: Default univariate statistic value. - - drift_flags: Optional override of the ``alert`` flag per ``(feature, method)`` pair. - - include_multivariate: Whether to add a final multivariate row block. - - multivariate_value: Value placed on multivariate rows. - - multivariate_drift: Drift flag placed on multivariate rows. - - extra_rows: Additional pre-built rows to append (used for dedup/p_value tests). - - Return: - pandas.DataFrame with columns: timestamp, feature, metric, statistic, - p_value, alert, chunk_index, chunk_start_date, chunk_end_date. - """ - rows: list[dict[str, Any]] = [] - drift_flags = drift_flags or {} - # Synthetic chunk-end offset that mirrors the high-precision boundary - # (``...:59.999999999``) emitted by ``ModelAnalysis`` for minute chunks. - # Computing via ``Timedelta`` instead of string concatenation keeps the - # helper safe for both minute- and second-precision timestamps. - chunk_span = pd.Timedelta(seconds=59, nanoseconds=999999999) - - for chunk_index, ts in enumerate(timestamps): - chunk_start = pd.Timestamp(ts) - chunk_end = chunk_start + chunk_span - for feature in features: - for method in methods: - rows.append( - { - 'timestamp': chunk_start, - 'feature': feature, - 'metric': method, - 'statistic': statistic, - 'p_value': 0.5, - 'alert': drift_flags.get((feature, method), False), - 'chunk_index': chunk_index, - 'chunk_start_date': chunk_start, - 'chunk_end_date': chunk_end, - } - ) - if include_multivariate: - rows.append( - { - 'timestamp': chunk_start, - 'feature': 'multivariate', - 'metric': 'multivariate', - 'statistic': multivariate_value, - 'p_value': 0.0, - 'alert': multivariate_drift, - 'chunk_index': chunk_index, - 'chunk_start_date': chunk_start, - 'chunk_end_date': chunk_end, - } - ) - - if extra_rows: - rows.extend(extra_rows) - - return pd.DataFrame(rows) diff --git a/e2e/scenario_inputs/drift_base.json b/e2e/scenario_inputs/drift_base.json index 1d02c5f..42762a9 100644 --- a/e2e/scenario_inputs/drift_base.json +++ b/e2e/scenario_inputs/drift_base.json @@ -2,9 +2,9 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "schema": "predictions_schema", + "schema": "sientia_data", "source_table_name": "laborious_data", - "target_table_name": "drift", + "target_table_name": "drift_metrics", "interval": 60, "drift_metrics": ["kolmogorov_smirnov", "jensen_shannon", "wasserstein"], "chunk_period": "min", diff --git a/e2e/scenario_inputs/format_export_base.json b/e2e/scenario_inputs/format_export_base.json index 151342e..60261e0 100644 --- a/e2e/scenario_inputs/format_export_base.json +++ b/e2e/scenario_inputs/format_export_base.json @@ -2,8 +2,8 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/main_happy_path.json b/e2e/scenario_inputs/main_happy_path.json index e951311..6d59d2a 100644 --- a/e2e/scenario_inputs/main_happy_path.json +++ b/e2e/scenario_inputs/main_happy_path.json @@ -10,8 +10,8 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/main_invalid_datetime.json b/e2e/scenario_inputs/main_invalid_datetime.json index 1a3c40b..cb3165e 100644 --- a/e2e/scenario_inputs/main_invalid_datetime.json +++ b/e2e/scenario_inputs/main_invalid_datetime.json @@ -10,8 +10,8 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", + "query": "SELECT timestamp, variable, value FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/main_missing_required.json b/e2e/scenario_inputs/main_missing_required.json index e491755..35ed1b7 100644 --- a/e2e/scenario_inputs/main_missing_required.json +++ b/e2e/scenario_inputs/main_missing_required.json @@ -10,7 +10,7 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "schema": "predictions_schema", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data" } diff --git a/e2e/scenario_inputs/main_sql_error.json b/e2e/scenario_inputs/main_sql_error.json index ffd6436..5823319 100644 --- a/e2e/scenario_inputs/main_sql_error.json +++ b/e2e/scenario_inputs/main_sql_error.json @@ -11,7 +11,7 @@ "model_name": "test_model", "model_id": "{{MODEL_ID}}", "query": "SELECT * FROM nonexistent_table WHERE invalid_syntax =", - "schema": "predictions_schema", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/minimal_retrain_base.json b/e2e/scenario_inputs/minimal_retrain_base.json index 98cd94c..13a4ec5 100644 --- a/e2e/scenario_inputs/minimal_retrain_base.json +++ b/e2e/scenario_inputs/minimal_retrain_base.json @@ -2,9 +2,9 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", - "table_name": "retrain_reports", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", + "table_name": "log_retrain", "datetime_columns": ["timestamp", "created_at"], "model_config": { "target": "sensor_1" diff --git a/e2e/scenario_inputs/minio_offload_load_query.json b/e2e/scenario_inputs/minio_offload_load_query.json index 3d4c09c..d97b1e8 100644 --- a/e2e/scenario_inputs/minio_offload_load_query.json +++ b/e2e/scenario_inputs/minio_offload_load_query.json @@ -5,7 +5,7 @@ "model_id": "{{MODEL_ID}}", "workflow_name": "predictions_batch" }, - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", "model_name": "test_model", "datetime_columns": ["timestamp", "created_at"] } diff --git a/e2e/scenario_inputs/minio_offload_workflow.json b/e2e/scenario_inputs/minio_offload_workflow.json index 4e7b82a..61dc28e 100644 --- a/e2e/scenario_inputs/minio_offload_workflow.json +++ b/e2e/scenario_inputs/minio_offload_workflow.json @@ -2,8 +2,8 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/prediction_process_base.json b/e2e/scenario_inputs/prediction_process_base.json index ff46bb9..3a73210 100644 --- a/e2e/scenario_inputs/prediction_process_base.json +++ b/e2e/scenario_inputs/prediction_process_base.json @@ -2,8 +2,8 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}", - "schema": "predictions_schema", + "query": "SELECT timestamp, variable, value, created_at FROM sientia_data.laborious_data WHERE model_id = {{MODEL_ID}}", + "schema": "sientia_data", "table_name": "predictions", "transform_table_name": "transformed_data", "input_filters": { diff --git a/e2e/scenario_inputs/simple_metrics_base.json b/e2e/scenario_inputs/simple_metrics_base.json index 7ba84ed..5b25242 100644 --- a/e2e/scenario_inputs/simple_metrics_base.json +++ b/e2e/scenario_inputs/simple_metrics_base.json @@ -2,10 +2,10 @@ "schedule_name": "test-schedule", "model_name": "test_model", "model_id": "{{MODEL_ID}}", - "schema": "predictions_schema", + "schema": "sientia_data", "predictions_table_name": "predictions", "data_table_name": "laborious_data", - "target_table_name": "simple_metrics_data", + "target_table_name": "simple_metrics", "interval_minutes": 60, "metrics": ["rmse", "mse", "mae", "r2"], "model_config": { diff --git a/e2e/scenarios.md b/e2e/scenarios.md index da6aa4f..5fd681b 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -232,81 +232,87 @@ Source: `e2e/test_minio_offload.py` ## 5. Drift Workflow Scenarios Source: `e2e/test_drift.py` -The drift suite mocks `sientia.ModelAnalysis.ModelAnalysis` (not installed; see -`CODE_ISSUES.md` issue #1) through the controllable `_FakeModelAnalysis` stub -exposed by the `model_analysis_stub` fixture. The `mlflow_repository_stub` -provides the reference-data CSV via `download_artifacts`. Every scenario asserts -postgres rows in `predictions_schema.drift` against this canonical schema: +The drift suite drives the **real** `sientia_model.analytics.drift_analysis.DriftAnalysis` +analyzer (no stubs / mocks). Each scenario exercises the full pipeline: -`id, model_id, feature, method, value, drift, chunk, timestamp, timestamp_end, accurate, created_at, updated_at`. +``` +laborious_data (Postgres) -> load_custom_query + -> calculate_drift (DriftAnalysis univariate + multivariate) + -> export_data_to_postgres (sientia_data.drift_metrics) +``` + +The `mlflow_repository_stub` provides the reference-data CSV via +`download_artifacts`, and tests assert postgres rows in +`sientia_data.drift_metrics` against this canonical schema: + +`id, model_id, feature, method, value, alert, chunk_index, chunk_start_date, chunk_end_date, accurate, timestamp, created_at`. + +Tests assert behavioral / structural properties (column presence, NOT NULL +constraints, business-key invariants like uniform `timestamp` and stamped +`model_id`) rather than exact numeric drift scores, since those depend on +the real analyzer implementation and the synthetic data fed in. ### 5.1 Happy paths #### D.1.1 Full pipeline persists all columns with reference data -**Summary**: ModelAnalysis returns a deterministic drift dataframe; the -reference CSV is downloaded from the MLflow stub. +**Summary**: 10 minutes of target data are inserted; a 10-row reference CSV +is configured via the MLflow stub. The `DriftAnalysis` runs end-to-end. **Expected Outcome**: -- One row per `(chunk, feature, method)` plus a `multivariate` block per chunk. -- Every drift column is populated and `accurate=True`. -- `timestamp_end` preserves the high-precision string (`HH:MM:59.999999999`). +- One row per `(chunk_index, feature, method)` plus a `multivariate` block + per chunk is persisted. +- Every column in the DDL is populated; `feature` is the only nullable column + per the new schema. +- `accurate=True` for every row (reference path). +- All three default univariate methods reach the analyzer. +- `model_id` is stamped as `text` and uniform across rows. +- `timestamp` equals `max(target_data.timestamp)` and is uniform across rows. +- `chunk_start_date` / `chunk_end_date` are persisted as ISO text and ordered. - `p_value` is dropped before persistence. -- `drift` flags propagate per `(feature, method)` configuration. #### D.1.2 30% fallback when reference data is unavailable -**Summary**: `get_reference_data` fails alias resolution and returns `None`; -`calculate_drift` uses the first 30% of target rows as reference. +**Summary**: MLflow alias resolution is forced to fail so +`get_reference_data` returns `None`; `calculate_drift` falls back to the +first 30% of target rows as reference. **Expected Outcome**: -- Persisted rows carry `accurate=False`. +- All persisted rows carry `accurate=False`. - A `MODEL_METRICS_REFERENCE_DATA_WARNING` notification is emitted to MongoDB. -### 5.2 Filtering / dedup invariants - -#### D.2.1 Deduplication and `p_value` removal -**Summary**: ModelAnalysis returns duplicate `(timestamp, method, feature)` rows -plus a `p_value` column. - -**Expected Outcome**: -- Duplicates are collapsed keeping the first occurrence. -- `p_value` is absent from the persisted rows. - -#### D.2.2 Out-of-range timestamps filtered -**Summary**: Drift rows whose timestamps are not present in the target window -must be discarded before persistence. - -### 5.3 Failure paths +### 5.2 Failure paths #### D.3.1 Empty target data short-circuits the workflow -**Summary**: `load_custom_query` returns no rows; ModelAnalysis is never -instantiated and no drift rows are written. +**Summary**: `load_custom_query` returns no rows. -#### D.3.2 ModelAnalysis raises during dataframe assembly -**Summary**: `get_drift_metrics_dataframe` raises. The activity catches the -error, sends a `MODEL_METRICS_GET_DRIFT_METRICS_ERROR` notification, and the -workflow completes without persisting drift rows. +**Expected Outcome**: +- The workflow returns early and writes nothing to `sientia_data.drift_metrics`. -### 5.4 Configuration paths - -#### D.4.1 Default drift metrics propagated to analyzer -**Summary**: Omitting `drift_metrics` defaults to -`['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']` and forwards the -exact list to `detect_univariate_drift`. +### 5.3 Configuration paths #### D.4.2 Invalid `chunk_period` raises ValueError **Summary**: Anything other than `min` / `s` is rejected by `calculate_drift`. -#### D.4.3 `chunk_period='s'` keeps seconds in timestamp filtering -**Summary**: Truncated `YYYY-MM-DD HH:MM` rows are filtered out when chunking -runs at second granularity. +**Expected Outcome**: +- The workflow surfaces the `ValueError` ("Invalid chunk period: ..."). +- No rows are persisted. + +#### D.4.3 `chunk_period='s'` preserves seconds in `chunk_start_date` +**Summary**: Target data spans two minutes with samples at second-30 +boundaries; the activity is configured with `chunk_period='s'`. + +**Expected Outcome**: +- At least one persisted `chunk_start_date` carries `seconds=30`, proving + that the analyzer chunked at sub-minute granularity and the ISO-text + serialization preserved the boundary. --- ## 6. Simple Metrics Workflow Scenarios Source: `e2e/test_simple_metrics.py` -Validates `predictions_schema.simple_metrics_data` columns: +Validates `sientia_data.simple_metrics` columns: `id, model_id, metric, value, timestamp, data_size, interval_minutes, created_at`. +Note: ``timestamp`` is now nullable per the new DDL and ``model_id`` is ``text``. ### 6.1 Happy paths @@ -339,8 +345,10 @@ the workflow exits before `calculate_simple_metrics` and writes nothing. Source: `e2e/test_minimal_retrain.py` The MLflow registry is fully mocked (no real artifacts in test container). -Validates `predictions_schema.retrain_reports` columns: -`id, model_id, model_name, timestamp, status, version, mlflow_run_id, mlflow_experiment_id, created_at`. +Validates `sientia_data.log_retrain` columns: +`mlflow_experiment_id, mlflow_run_id, model_id, model_name, status, timestamp, version`. +Note: the new DDL drops the legacy ``id`` and ``created_at`` columns, +``mlflow_experiment_id`` is now ``int8`` and ``model_id`` is ``text``. ### 7.1 Happy path @@ -350,7 +358,7 @@ the new version is promoted to the `production` alias. **Expected Outcome**: - Report row has success status, `version='7'`, `mlflow_run_id='retrain-run-id'`, - `mlflow_experiment_id='experiment-id'`. + `mlflow_experiment_id=4242` (`int8`). - `mlflow.log_artifact` is called with the input CSV. - `promote_to_alias` is called once with the resolved version and alias. diff --git a/e2e/test_child_workflows_e2e.py b/e2e/test_child_workflows_e2e.py index 52a4b3f..3d10c68 100644 --- a/e2e/test_child_workflows_e2e.py +++ b/e2e/test_child_workflows_e2e.py @@ -26,7 +26,7 @@ async def test_format_and_export_prediction_default_path_e2e( client = temporal_test_env.client model_id = 401 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.predictions WHERE model_id = {model_id}')) metadata = { 'metadata': { @@ -44,7 +44,7 @@ async def test_format_and_export_prediction_default_path_e2e( 'timestamp': '2024-01-01 12:00:00+00:00', 'model_id': model_id, 'model_name': 'test_model', - 'schema': 'predictions_schema', + 'schema': 'sientia_data', 'table_name': 'predictions', 'transform_table_name': 'transformed_data', 'comment': 'e2e child workflow default path', @@ -64,7 +64,7 @@ async def test_format_and_export_prediction_default_path_e2e( row = conn.execute( text( f'SELECT prediction, prediction_confidence, prediction_status, comments ' - f'FROM predictions_schema.predictions WHERE model_id = {model_id}' + f'FROM sientia_data.predictions WHERE model_id = {model_id}' ) ).fetchone() assert row is not None diff --git a/e2e/test_drift.py b/e2e/test_drift.py index 80d5664..fe03656 100644 --- a/e2e/test_drift.py +++ b/e2e/test_drift.py @@ -1,22 +1,31 @@ """ End-to-end tests for the Drift workflow. +The drift suite drives the **real** ``sientia_model.analytics.drift_analysis.DriftAnalysis`` +analyzer (no mocking). Each scenario exercises the full pipeline: + + laborious_data (Postgres) + -> load_custom_query + -> calculate_drift (DriftAnalysis univariate + multivariate) + -> export_data_to_postgres (sientia_data.drift_metrics) + Coverage focus: -- Full pipeline persists drift rows with **all** columns expected by the - ``predictions_schema.drift`` table (model_id, feature, method, value, drift, - chunk, timestamp, timestamp_end, accurate, plus DB-managed id/created_at/updated_at). -- ``get_reference_data`` happy path (CSV downloaded from MLflow stub) and - fallback path (30% of target data when reference is unavailable). -- ``calculate_drift`` invariants: ``p_value`` dropped, duplicates removed, rows - outside target timestamps filtered, default ``drift_metrics`` propagated. -- Failure paths: ``ModelAnalysis.get_drift_metrics_dataframe`` raises ⇒ - workflow completes without writes; empty target data ⇒ workflow short-circuits; - invalid ``chunk_period`` ⇒ activity raises and workflow surfaces the error. +- Happy path persists every column required by ``sientia_data.drift_metrics`` + with a valid reference dataset downloaded from MLflow. +- 30% fallback path activates when the MLflow reference is unavailable and + emits the ``MODEL_METRICS_REFERENCE_DATA_WARNING`` notification. +- Empty target data short-circuits the workflow without persisting anything. +- Invalid ``chunk_period`` is rejected by ``calculate_drift``. +- ``chunk_period='s'`` preserves second-level precision in + ``chunk_start_date``. + +Tests assert behavioral / structural properties (column presence, NOT NULL +constraints, business-key invariants) rather than exact numeric values, since +those depend on the real analyzer implementation and synthetic data. """ from datetime import datetime, timedelta, timezone -from decimal import Decimal from pathlib import Path from unittest.mock import MagicMock @@ -27,7 +36,6 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from e2e.helpers import ( - build_drift_dataframe, insert_target_data_for_drift, load_scenario_input, make_workflow_id, @@ -36,21 +44,41 @@ from e2e.helpers import ( from laborious.activities.activities import Activities from laborious.workflows.drift import Drift -# Drift columns that must be present (and non-null where required) on every row. +# Drift columns persisted on every row in ``sientia_data.drift_metrics`` — +# mirrors the production DDL. EXPECTED_DRIFT_COLUMNS = [ 'id', 'model_id', 'feature', 'method', 'value', - 'drift', - 'chunk', + 'alert', + 'chunk_index', + 'chunk_start_date', + 'chunk_end_date', + 'accurate', 'timestamp', - 'timestamp_end', + 'created_at', +] + +# Columns the DDL marks as NOT NULL. ``feature`` and ``timestamp`` are +# nullable in the production schema (multivariate rows do not bind to a +# single feature; ``timestamp`` is allowed to be empty when upstream data has +# no usable instant). +NON_NULL_DRIFT_COLUMNS = { + 'id', + 'model_id', + 'method', + 'value', + 'alert', + 'chunk_index', + 'chunk_start_date', + 'chunk_end_date', 'accurate', 'created_at', - 'updated_at', -] +} + +DEFAULT_DRIFT_METHODS = ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'] def _drift_input(model_id: int, **overrides) -> dict: @@ -60,42 +88,37 @@ def _drift_input(model_id: int, **overrides) -> dict: return input_data -def _five_minute_window(offset_minutes: int = 6) -> tuple[list[str], list[str]]: +def _recent_minute_timestamps(count: int, offset_minutes: int = 6) -> list[str]: """ - Build five consecutive UTC minute timestamps positioned in the recent past. + Build ``count`` consecutive UTC minute timestamps placed in the recent past. The Drift workflow filters target rows with ``timestamp > NOW() - INTERVAL``, - so timestamps must be recent for tests to retrieve any data. We snap to - minute precision and back off ``offset_minutes`` minutes so all chunks land - well inside the default 60-minute interval defined in ``drift_base.json``. + so timestamps must be recent for tests to retrieve any data. Snapping to + minute precision keeps the helper deterministic regardless of clock skew. Args: - - offset_minutes: How many minutes ago the most recent chunk should be. + - count (int): How many consecutive minute timestamps to generate. + - offset_minutes (int): Minutes ago for the EARLIEST generated timestamp. Return: - Tuple ``(target_timestamps, chunk_timestamps)``: - - - ``target_timestamps``: ISO strings with ``+0000`` used as ``timestamp`` - and ``created_at`` columns when inserting target rows. - - ``chunk_timestamps``: ``YYYY-MM-DD HH:MM`` truncations matching what - ``calculate_drift`` filters on for ``chunk_period='min'``. + list[str]: ISO strings with ``+0000`` offset, one per minute. """ base = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta( minutes=offset_minutes ) - target_timestamps = [ - (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M:%S%z') for i in range(5) + return [ + (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M:%S%z') for i in range(count) ] - chunk_timestamps = [ - (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M') for i in range(5) - ] - return target_timestamps, chunk_timestamps def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None: """ Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns ``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``. + + Args: + - mlflow_repository_stub: External MLflow repository fixture. + - reference_rows (pd.DataFrame): Rows to expose as the production reference. """ def _download(run_id: str, artifact_path: str, dst_path: str, metadata=None): @@ -115,22 +138,34 @@ def _force_reference_unavailable(mlflow_repository_stub) -> None: ) -def _assert_drift_columns_complete(rows, *, expected_count: int) -> None: - """Validate row count and that the canonical drift columns are populated.""" - assert len(rows) == expected_count, ( - f'Expected {expected_count} drift rows persisted, got {len(rows)}' - ) - seen_columns = set(rows[0]._mapping.keys()) if rows else set() +def _select_drift_rows(postgres_engine, model_id: int) -> list[dict]: + """Read every persisted drift row for ``model_id`` ordered by chunk/feature/method.""" + with postgres_engine.connect() as conn: + rows = ( + conn.execute( + text( + 'SELECT * FROM sientia_data.drift_metrics ' + 'WHERE model_id = :m ' + 'ORDER BY chunk_index, feature, method' + ), + {'m': str(model_id)}, + ) + .mappings() + .all() + ) + return [dict(row) for row in rows] + + +def _assert_required_columns_populated(rows: list[dict]) -> None: + """Validate column presence and NOT NULL constraints on every row.""" + assert rows, 'expected at least one drift row to be persisted' + seen_columns = set(rows[0].keys()) for column in EXPECTED_DRIFT_COLUMNS: assert column in seen_columns, f'Missing drift column in postgres: {column}' - for row in rows: - mapping = dict(row._mapping) - for column in EXPECTED_DRIFT_COLUMNS: - if column == 'value': - # ``value`` is nullable in the table; skip null check, only ensure key exists. - continue - assert mapping[column] is not None, f"Column '{column}' is NULL in {mapping}" + for column in NON_NULL_DRIFT_COLUMNS: + assert row[column] is not None, f"Column '{column}' is NULL in {row}" + assert 'p_value' not in row, 'p_value must not be persisted to drift_metrics' @pytest.mark.asyncio @@ -141,112 +176,111 @@ async def test_drift_happy_path_persists_all_columns_with_reference_data( test_activities: Activities, postgres_engine, mlflow_repository_stub, - model_analysis_stub, ): """ Scenario D.1.1: Happy path with reference data downloaded from MLflow. - Validates the full drift pipeline produces one row per (chunk, feature, method) - combination plus the multivariate row block, with **all** columns required by - ``predictions_schema.drift`` populated. Uses a deterministic mocked drift - dataframe so the postgres assertions remain stable across runs. + Drives the full pipeline against the real ``DriftAnalysis``. Asserts: + + - One row is persisted per ``(chunk_index, feature, method)`` combination + plus the multivariate row block, with every column required by + ``sientia_data.drift_metrics`` populated. + - The three default univariate methods are forwarded to the analyzer. + - ``model_id`` and ``timestamp`` are stamped by the activity (not by the + analyzer); ``timestamp`` equals ``max(target_data.timestamp)`` and is + identical on every persisted row. + - ``chunk_start_date`` / ``chunk_end_date`` are persisted as ISO text so + the analyzer's nanosecond-precision boundaries survive the ``text`` + column type. + - ``accurate=True`` because the reference dataset was available. """ client = temporal_test_env.client model_id = 411 - target_timestamps, chunk_timestamps = _five_minute_window() + target_timestamps = _recent_minute_timestamps(count=10) insert_target_data_for_drift( postgres_engine, model_id=model_id, timestamps=target_timestamps, variables_values={ - 'sensor_1': [10.0, 11.0, 12.0, 13.0, 14.0], - 'sensor_2': [20.0, 21.0, 22.0, 23.0, 24.0], + 'sensor_1': [10.0 + i * 0.1 for i in range(10)], + 'sensor_2': [20.0 + i * 0.5 for i in range(10)], }, ) reference_df = pd.DataFrame( { - 'timestamp': ['2023-12-31 11:00:00+00:00', '2023-12-31 11:01:00+00:00'], - 'sensor_1': [9.5, 9.6], - 'sensor_2': [19.5, 19.6], + 'timestamp': [ + f'2023-12-31 11:{minute:02d}:00+00:00' for minute in range(10) + ], + 'sensor_1': [9.0 + i * 0.05 for i in range(10)], + 'sensor_2': [18.0 + i * 0.25 for i in range(10)], } ) _configure_reference_csv(mlflow_repository_stub, reference_df) - features = ['sensor_1', 'sensor_2'] - methods = ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'] - drift_flags = { - ('sensor_1', 'wasserstein'): True, - ('sensor_2', 'wasserstein'): True, - ('sensor_2', 'jensen_shannon'): True, - } - - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps, - features=features, - methods=methods, - statistic=0.42, - drift_flags=drift_flags, - multivariate_value=15.5, - multivariate_drift=True, - ) - ) - input_data = _drift_input(model_id) await start_and_await_workflow( client, Drift.run, input_data, make_workflow_id('test-drift-happy-path') ) - with postgres_engine.connect() as conn: - rows = ( - conn.execute( - text( - 'SELECT * FROM predictions_schema.drift ' - 'WHERE model_id = :model_id ORDER BY chunk, feature, method' - ), - {'model_id': model_id}, - ) - .mappings() - .all() - ) + rows = _select_drift_rows(postgres_engine, model_id) - expected_rows = len(chunk_timestamps) * (len(features) * len(methods) + 1) # univariate + multivariate - assert len(rows) == expected_rows - for column in EXPECTED_DRIFT_COLUMNS: - assert column in rows[0], f'Missing drift column in postgres: {column}' + exported_csv_path = '/tmp/test_drift_happy_path_exported.csv' + pd.DataFrame(rows).to_csv(exported_csv_path, index=False) + print( + f'\n[test_drift_happy_path] Exported drift dataframe ' + f'({len(rows)} rows) -> {exported_csv_path}' + ) - for row in rows: - for column in EXPECTED_DRIFT_COLUMNS: - if column == 'value': - continue - assert row[column] is not None, f"Column '{column}' is NULL in {dict(row)}" + _assert_required_columns_populated(rows) - multivariate_rows = [r for r in rows if r['feature'] == 'multivariate'] + # The activity drops the target column from the feature list, so only + # ``sensor_2`` participates in univariate analysis (``sensor_1`` is the + # configured target). Multivariate produces one row per chunk regardless. univariate_rows = [r for r in rows if r['feature'] != 'multivariate'] + multivariate_rows = [r for r in rows if r['feature'] == 'multivariate'] + assert univariate_rows, 'expected univariate drift rows for non-target features' + assert multivariate_rows, 'expected one multivariate drift row per chunk' - assert len(multivariate_rows) == len(chunk_timestamps) + # All three default methods must reach the analyzer. + assert {r['method'] for r in univariate_rows} == set(DEFAULT_DRIFT_METHODS) assert all(r['method'] == 'multivariate' for r in multivariate_rows) - assert all(r['drift'] is True for r in multivariate_rows) - assert all(Decimal(str(r['value'])) == Decimal('15.5') for r in multivariate_rows) + assert {r['feature'] for r in univariate_rows} == {'sensor_2'} - assert len(univariate_rows) == len(features) * len(methods) * len(chunk_timestamps) - assert {r['method'] for r in univariate_rows} == set(methods) - assert {r['feature'] for r in univariate_rows} == set(features) + # ``timestamp`` is stamped uniformly with ``max(target_data.timestamp)``. + expected_timestamp = pd.to_datetime(max(target_timestamps), utc=True) + persisted_timestamps = {pd.to_datetime(r['timestamp'], utc=True) for r in rows} + assert len(persisted_timestamps) == 1, ( + 'timestamp must be uniform across all drift rows ' + f'(got {len(persisted_timestamps)} distinct values)' + ) + assert pd.Timestamp(persisted_timestamps.pop()) == expected_timestamp, ( + 'timestamp must equal max(target_data.timestamp)' + ) - drift_pairs = {(r['feature'], r['method']): r['drift'] for r in univariate_rows} - for (feature, method), expected_drift in drift_flags.items(): - assert drift_pairs[(feature, method)] is expected_drift + # ``model_id`` is stamped by ``calculate_drift`` (not produced by the analyzer). + assert all(r['model_id'] == str(model_id) for r in rows), ( + 'model_id must be stamped on every drift row' + ) - assert all(r['accurate'] is True for r in rows), 'reference path should mark rows as accurate' + # Reference path → accurate=True. + assert all(r['accurate'] is True for r in rows), ( + 'reference path should mark all rows as accurate' + ) - assert all( - r['timestamp_end'].endswith(':59.999999999') and 'T' in r['timestamp_end'] - for r in rows - ), 'timestamp_end should preserve the high-precision ISO string from ModelAnalysis' - - assert all('p_value' not in r for r in rows), 'p_value must be dropped before postgres' + # ISO text serialization preserves ordering between start/end of each chunk. + for row in rows: + assert 'T' in row['chunk_start_date'], ( + f"chunk_start_date should be ISO text, got {row['chunk_start_date']!r}" + ) + assert 'T' in row['chunk_end_date'], ( + f"chunk_end_date should be ISO text, got {row['chunk_end_date']!r}" + ) + assert row['chunk_start_date'] <= row['chunk_end_date'], ( + f'chunk_start_date must precede chunk_end_date ' + f"(start={row['chunk_start_date']}, end={row['chunk_end_date']})" + ) @pytest.mark.asyncio @@ -257,60 +291,42 @@ async def test_drift_uses_30pct_fallback_when_reference_unavailable( test_activities: Activities, postgres_engine, mlflow_repository_stub, - model_analysis_stub, notification_inserts, ): """ - Scenario D.1.2: ``get_reference_data`` returns ``None`` (production alias missing), - so ``calculate_drift`` falls back to using the first 30% of target rows as - reference. Persisted rows must report ``accurate=False`` and a warning - notification must be emitted to mongo. + Scenario D.1.2: ``get_reference_data`` returns ``None`` (production alias + missing), so ``calculate_drift`` falls back to using the first 30% of + target rows as reference. Persisted rows must report ``accurate=False`` + and a ``MODEL_METRICS_REFERENCE_DATA_WARNING`` notification must be + emitted to mongo. """ client = temporal_test_env.client model_id = 412 - target_timestamps, chunk_timestamps = _five_minute_window() + target_timestamps = _recent_minute_timestamps(count=10) insert_target_data_for_drift( postgres_engine, model_id=model_id, timestamps=target_timestamps, variables_values={ - 'sensor_1': [10.0, 10.1, 10.2, 10.3, 10.4], - 'sensor_2': [20.0, 20.1, 20.2, 20.3, 20.4], + 'sensor_1': [10.0 + i * 0.1 for i in range(10)], + 'sensor_2': [20.0 + i * 0.5 for i in range(10)], }, ) _force_reference_unavailable(mlflow_repository_stub) - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps, - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - statistic=0.7, - include_multivariate=False, - ) - ) - input_data = _drift_input(model_id) - await start_and_await_workflow( client, Drift.run, input_data, make_workflow_id('test-drift-fallback') ) - with postgres_engine.connect() as conn: - rows = ( - conn.execute( - text('SELECT * FROM predictions_schema.drift WHERE model_id = :m'), - {'m': model_id}, - ) - .mappings() - .all() - ) + rows = _select_drift_rows(postgres_engine, model_id) + _assert_required_columns_populated(rows) - assert len(rows) == len(chunk_timestamps) - assert all(r['accurate'] is False for r in rows), 'fallback path must mark rows as inaccurate' - assert all(r['feature'] == 'sensor_1' for r in rows) + assert all(r['accurate'] is False for r in rows), ( + 'fallback path must mark all rows as inaccurate' + ) fallback_warnings = [ call @@ -322,168 +338,6 @@ async def test_drift_uses_30pct_fallback_when_reference_unavailable( assert len(fallback_warnings) >= 1, 'expected reference fallback warning notification' -@pytest.mark.asyncio -@pytest.mark.integration -async def test_drift_drops_p_value_and_dedupes_by_timestamp_method_feature( - temporal_test_env: WorkflowEnvironment, - temporal_worker_drift: Worker, - test_activities: Activities, - postgres_engine, - mlflow_repository_stub, - model_analysis_stub, -): - """ - Scenario D.2.1: ModelAnalysis returns duplicate (timestamp, method, feature) - rows and a populated ``p_value`` column. ``calculate_drift`` must deduplicate - keeping the first occurrence and the persisted rows must not contain - ``p_value``. - """ - client = temporal_test_env.client - model_id = 421 - - target_timestamps, chunk_timestamps = _five_minute_window() - insert_target_data_for_drift( - postgres_engine, - model_id=model_id, - timestamps=target_timestamps, - variables_values={ - 'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0], - 'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0], - }, - ) - - duplicates = [ - # Re-emit the first chunk for sensor_1/kolmogorov_smirnov with a different value - # so we can prove the dedup keeps the FIRST row. - { - 'timestamp': pd.Timestamp(chunk_timestamps[0]), - 'feature': 'sensor_1', - 'metric': 'kolmogorov_smirnov', - 'statistic': 0.99, - 'p_value': 0.02, - 'alert': True, - 'chunk_index': 0, - 'chunk_start_date': pd.Timestamp(chunk_timestamps[0]), - 'chunk_end_date': pd.Timestamp(f'{chunk_timestamps[0]}:59.999999999'), - } - ] - - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps, - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - statistic=0.5, - include_multivariate=False, - extra_rows=duplicates, - ) - ) - _force_reference_unavailable(mlflow_repository_stub) - - input_data = _drift_input(model_id) - await start_and_await_workflow( - client, Drift.run, input_data, make_workflow_id('test-drift-dedupe') - ) - - with postgres_engine.connect() as conn: - rows = ( - conn.execute( - text( - 'SELECT chunk, feature, method, value FROM predictions_schema.drift ' - 'WHERE model_id = :m ORDER BY chunk' - ), - {'m': model_id}, - ) - .mappings() - .all() - ) - - assert len(rows) == len(chunk_timestamps), 'duplicates must be removed before persistence' - - first_chunk_rows = [r for r in rows if r['chunk'] == 0] - assert len(first_chunk_rows) == 1 - assert Decimal(str(first_chunk_rows[0]['value'])) == Decimal('0.5'), ( - 'dedup must keep the FIRST occurrence (statistic=0.5), not the duplicate (statistic=0.99)' - ) - - columns = set(rows[0].keys()) - assert 'p_value' not in columns - - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_drift_drops_rows_outside_target_timestamps( - temporal_test_env: WorkflowEnvironment, - temporal_worker_drift: Worker, - test_activities: Activities, - postgres_engine, - mlflow_repository_stub, - model_analysis_stub, -): - """ - Scenario D.2.2: Drift rows whose timestamps are not present in the target - data (e.g. came from the reference distribution) must be discarded so the - saved drift only reflects analysis chunks. - """ - client = temporal_test_env.client - model_id = 422 - - target_timestamps, chunk_timestamps = _five_minute_window() - insert_target_data_for_drift( - postgres_engine, - model_id=model_id, - timestamps=target_timestamps, - variables_values={ - 'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0], - 'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0], - }, - ) - - out_of_range_timestamp = pd.Timestamp('2099-12-31 23:59') - extra = [ - { - 'timestamp': out_of_range_timestamp, - 'feature': 'sensor_1', - 'metric': 'kolmogorov_smirnov', - 'statistic': 0.5, - 'p_value': 0.0, - 'alert': True, - 'chunk_index': 99, - 'chunk_start_date': out_of_range_timestamp, - 'chunk_end_date': pd.Timestamp('2099-12-31 23:59:59.999999999'), - } - ] - - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps, - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - include_multivariate=False, - extra_rows=extra, - ) - ) - _force_reference_unavailable(mlflow_repository_stub) - - input_data = _drift_input(model_id) - await start_and_await_workflow( - client, Drift.run, input_data, make_workflow_id('test-drift-tts-filter') - ) - - with postgres_engine.connect() as conn: - chunks = [ - r[0] - for r in conn.execute( - text( - 'SELECT chunk FROM predictions_schema.drift WHERE model_id = :m ORDER BY chunk' - ), - {'m': model_id}, - ).all() - ] - - assert chunks == [0, 1, 2, 3, 4], 'out-of-range timestamps must be filtered out' - - @pytest.mark.asyncio @pytest.mark.integration async def test_drift_empty_target_data_short_circuits_workflow( @@ -492,33 +346,19 @@ async def test_drift_empty_target_data_short_circuits_workflow( test_activities: Activities, postgres_engine, mlflow_repository_stub, - model_analysis_stub, ): """ - Scenario D.3.1: When ``load_custom_query`` returns no rows the workflow must - return early without invoking ModelAnalysis or writing any drift rows. + Scenario D.3.1: When ``load_custom_query`` returns no rows the workflow + must return early without invoking the analyzer or writing any drift rows. """ client = temporal_test_env.client model_id = 431 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) _force_reference_unavailable(mlflow_repository_stub) - drift_calls = {'count': 0} - - def _factory(univariate, multivariate): - drift_calls['count'] += 1 - return build_drift_dataframe( - timestamps=['2024-01-01 12:00'], - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - include_multivariate=False, - ) - - model_analysis_stub.set_drift_dataframe(_factory) - input_data = _drift_input(model_id) await start_and_await_workflow( client, Drift.run, input_data, make_workflow_id('test-drift-empty-target') @@ -526,121 +366,10 @@ async def test_drift_empty_target_data_short_circuits_workflow( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'), - {'m': model_id}, + text('SELECT COUNT(*) FROM sientia_data.drift_metrics WHERE model_id = :m'), + {'m': str(model_id)}, ).scalar() assert count == 0 - assert drift_calls['count'] == 0, 'ModelAnalysis must not be invoked when target data is empty' - - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_drift_model_analysis_failure_keeps_workflow_alive_no_writes( - temporal_test_env: WorkflowEnvironment, - temporal_worker_drift: Worker, - test_activities: Activities, - postgres_engine, - mlflow_repository_stub, - model_analysis_stub, - notification_inserts, -): - """ - Scenario D.3.2: ``ModelAnalysis.get_drift_metrics_dataframe`` raises. The - activity must catch the error, send an error notification and return ``[]`` - so the workflow completes without persisting drift rows. - """ - client = temporal_test_env.client - model_id = 432 - - target_timestamps, _ = _five_minute_window() - insert_target_data_for_drift( - postgres_engine, - model_id=model_id, - timestamps=target_timestamps, - variables_values={ - 'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0], - 'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0], - }, - ) - - _force_reference_unavailable(mlflow_repository_stub) - model_analysis_stub.raise_on_get_drift_metrics_dataframe( - RuntimeError('drift analyzer crashed') - ) - - input_data = _drift_input(model_id) - await start_and_await_workflow( - client, Drift.run, input_data, make_workflow_id('test-drift-analyzer-error') - ) - - with postgres_engine.connect() as conn: - count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'), - {'m': model_id}, - ).scalar() - assert count == 0 - - error_notifications = [ - call - for call in notification_inserts.call_args_list - if call.args - and isinstance(call.args[0], dict) - and call.args[0].get('notification_id') == 'MODEL_METRICS_GET_DRIFT_METRICS_ERROR' - ] - assert len(error_notifications) >= 1 - - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_drift_default_drift_metrics_propagated_to_analyzer( - temporal_test_env: WorkflowEnvironment, - temporal_worker_drift: Worker, - test_activities: Activities, - postgres_engine, - mlflow_repository_stub, - model_analysis_stub, -): - """ - Scenario D.4.1: When ``drift_metrics`` is omitted from input the workflow - must default to ``['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']`` - and forward exactly that list to ``ModelAnalysis.detect_univariate_drift``. - """ - client = temporal_test_env.client - model_id = 441 - - target_timestamps, chunk_timestamps = _five_minute_window() - insert_target_data_for_drift( - postgres_engine, - model_id=model_id, - timestamps=target_timestamps, - variables_values={ - 'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0], - 'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0], - }, - ) - - _force_reference_unavailable(mlflow_repository_stub) - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps, - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - include_multivariate=False, - ) - ) - - input_data = _drift_input(model_id) - input_data.pop('drift_metrics', None) - - await start_and_await_workflow( - client, Drift.run, input_data, make_workflow_id('test-drift-default-methods') - ) - - instance = model_analysis_stub.last_instance - assert instance is not None, 'ModelAnalysis must have been instantiated' - assert len(instance.detect_univariate_drift_calls) == 1 - forwarded_methods = instance.detect_univariate_drift_calls[0]['methods'] - assert forwarded_methods == ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'] @pytest.mark.asyncio @@ -655,12 +384,12 @@ async def test_drift_invalid_chunk_period_raises_value_error( """ Scenario D.4.2: ``calculate_drift`` validates ``chunk_period`` and rejects anything other than ``min`` / ``s``. The workflow must surface the - ``ValueError`` to the caller and persist nothing. + ``ValueError`` and persist nothing. """ client = temporal_test_env.client model_id = 442 - target_timestamps, _ = _five_minute_window() + target_timestamps = _recent_minute_timestamps(count=5) insert_target_data_for_drift( postgres_engine, model_id=model_id, @@ -682,8 +411,9 @@ async def test_drift_invalid_chunk_period_raises_value_error( make_workflow_id('test-drift-bad-chunk-period'), ) - # Temporal wraps the activity ValueError in WorkflowFailureError; the message - # may live on ``.message`` or ``str(exc)`` depending on the SDK error class. + # Temporal wraps the activity ValueError in WorkflowFailureError; the + # message may live on ``.message`` or ``str(exc)`` depending on the SDK + # error class, so walk the cause chain looking for the guard text. cause_descriptions = [] current: BaseException | None = excinfo.value while current is not None: @@ -697,42 +427,36 @@ async def test_drift_invalid_chunk_period_raises_value_error( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'), - {'m': model_id}, + text('SELECT COUNT(*) FROM sientia_data.drift_metrics WHERE model_id = :m'), + {'m': str(model_id)}, ).scalar() assert count == 0 @pytest.mark.asyncio @pytest.mark.integration -async def test_drift_chunk_period_seconds_preserves_seconds_in_timestamp_column( +async def test_drift_chunk_period_seconds_preserves_seconds_in_chunk_start_date( temporal_test_env: WorkflowEnvironment, temporal_worker_drift: Worker, test_activities: Activities, postgres_engine, mlflow_repository_stub, - model_analysis_stub, ): """ - Scenario D.4.3: With ``chunk_period='s'`` the persisted ``timestamp`` - column must preserve second-level precision instead of being flattened to - the start of the minute, and ``chunk_period`` must be propagated to the - analyzer so it actually chunks at second granularity. + Scenario D.4.3: With ``chunk_period='s'`` the persisted ``chunk_start_date`` + column must preserve second-level precision so consumers can audit the + actual chunk boundary. """ client = temporal_test_env.client model_id = 443 base = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(minutes=2) + # Three samples spaced by 30 seconds inside two adjacent minutes. target_timestamps = [ base.strftime('%Y-%m-%d %H:%M:%S%z'), (base + timedelta(seconds=30)).strftime('%Y-%m-%d %H:%M:%S%z'), (base + timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S%z'), ] - chunk_timestamps_full = [ - base.strftime('%Y-%m-%d %H:%M:%S'), - (base + timedelta(seconds=30)).strftime('%Y-%m-%d %H:%M:%S'), - (base + timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S'), - ] insert_target_data_for_drift( postgres_engine, @@ -745,15 +469,6 @@ async def test_drift_chunk_period_seconds_preserves_seconds_in_timestamp_column( ) _force_reference_unavailable(mlflow_repository_stub) - model_analysis_stub.set_drift_dataframe( - lambda univariate, multivariate: build_drift_dataframe( - timestamps=chunk_timestamps_full, - features=['sensor_1'], - methods=['kolmogorov_smirnov'], - include_multivariate=False, - ) - ) - input_data = _drift_input(model_id, chunk_period='s') await start_and_await_workflow( client, @@ -766,21 +481,19 @@ async def test_drift_chunk_period_seconds_preserves_seconds_in_timestamp_column( rows = ( conn.execute( text( - 'SELECT chunk, timestamp FROM predictions_schema.drift ' - 'WHERE model_id = :m ORDER BY chunk' + 'SELECT chunk_start_date FROM sientia_data.drift_metrics ' + 'WHERE model_id = :m ORDER BY chunk_index' ), - {'m': model_id}, + {'m': str(model_id)}, ) .mappings() .all() ) - assert [r['chunk'] for r in rows] == [0, 1, 2] - seconds_present = {r['timestamp'].second for r in rows} - assert seconds_present == {0, 30}, ( - f'expected seconds 0 and 30 to be preserved, got {seconds_present}' + assert rows, 'expected at least one drift row to be persisted' + # At least one chunk must land on second-30, proving the analyzer chunked + # at sub-minute granularity instead of collapsing everything to minute=0. + seconds_present = {pd.Timestamp(r['chunk_start_date']).second for r in rows} + assert 30 in seconds_present, ( + f'expected at least one chunk_start_date with seconds=30, got {seconds_present}' ) - - instance = model_analysis_stub.last_instance - assert instance is not None - assert instance.detect_univariate_drift_calls[0]['chunk_period'] == 's' diff --git a/e2e/test_minimal_retrain.py b/e2e/test_minimal_retrain.py index 32f4bb6..e108aef 100644 --- a/e2e/test_minimal_retrain.py +++ b/e2e/test_minimal_retrain.py @@ -7,9 +7,14 @@ test container; we only validate that the workflow: - Loads training data via ``load_query_with_minio_offload``. - Calls ``retrain_model`` with a payload pointing at MinIO. - Calls ``update_production_model`` only when retrain succeeds. -- Persists ``retrain_reports`` rows with all required columns; success rows - carry the new ``version`` / ``mlflow_run_id`` / ``mlflow_experiment_id`` - while failure rows leave them ``NULL``. +- Persists ``sientia_data.log_retrain`` rows with all required columns; + success rows carry the new ``version`` / ``mlflow_run_id`` / + ``mlflow_experiment_id`` while failure rows leave them ``NULL``. + +The production DDL drops the legacy ``id`` / ``created_at`` columns and +moves ``mlflow_experiment_id`` to ``int8`` and ``model_id`` to ``text``. +The stubs used here therefore emit ``experiment_id`` as an integer to fit +the new column type. """ from contextlib import contextmanager @@ -30,16 +35,18 @@ from e2e.helpers import ( from laborious.activities.activities import Activities from laborious.workflows.minimal_retrain import MinimalRetrain +# Columns defined by the production DDL for ``sientia_data.log_retrain``. +# The legacy ``retrain_reports`` table had ``id`` and ``created_at``; the new +# DDL drops both. ``mlflow_experiment_id`` is ``int8`` and ``model_id`` is +# ``text``. EXPECTED_RETRAIN_REPORT_COLUMNS = [ - 'id', + 'mlflow_experiment_id', + 'mlflow_run_id', 'model_id', 'model_name', - 'timestamp', 'status', + 'timestamp', 'version', - 'mlflow_run_id', - 'mlflow_experiment_id', - 'created_at', ] @@ -106,7 +113,9 @@ def _configure_retrain_happy_path(mlflow_repository_stub) -> None: def fake_start_run(**kwargs): run_info = MagicMock() run_info.run_id = 'retrain-run-id' - run_info.experiment_id = 'experiment-id' + # ``mlflow_experiment_id`` is ``int8`` in the new DDL, so we feed an + # integer-compatible id from the stubbed run info. + run_info.experiment_id = 4242 yield run_info mlflow_repository_stub.start_run.side_effect = fake_start_run @@ -125,10 +134,10 @@ async def test_minimal_retrain_happy_path_writes_success_report( mlflow_repository_stub, ): """ - Scenario MR.1.1: Retrain succeeds. ``retrain_reports`` must contain a - success row with version/mlflow_run_id/mlflow_experiment_id populated and - the registry must have been told to promote the new version to the - configured alias. + Scenario MR.1.1: Retrain succeeds. ``sientia_data.log_retrain`` must + contain a success row with version/mlflow_run_id/mlflow_experiment_id + populated and the registry must have been told to promote the new version + to the configured alias. """ client = temporal_test_env.client model_id = 711 @@ -152,10 +161,10 @@ async def test_minimal_retrain_happy_path_writes_success_report( rows = ( conn.execute( text( - 'SELECT * FROM predictions_schema.retrain_reports ' + 'SELECT * FROM sientia_data.log_retrain ' 'WHERE model_id = :m' ), - {'m': model_id}, + {'m': str(model_id)}, ) .mappings() .all() @@ -163,17 +172,19 @@ async def test_minimal_retrain_happy_path_writes_success_report( assert len(rows) == 1 for column in EXPECTED_RETRAIN_REPORT_COLUMNS: - assert column in rows[0], f'Missing retrain report column: {column}' + assert column in rows[0], f'Missing log_retrain column: {column}' row = rows[0] - assert row['model_id'] == model_id + # ``model_id`` is now ``text``; compare against the stringified id. + assert row['model_id'] == str(model_id) assert row['model_name'] == 'test_model' assert row['status'] == 'Model retrained successfully.' assert row['version'] == '7' assert row['mlflow_run_id'] == 'retrain-run-id' - assert row['mlflow_experiment_id'] == 'experiment-id' + # ``mlflow_experiment_id`` is now ``int8``; assert the integer value + # provided by the stubbed run info. + assert row['mlflow_experiment_id'] == 4242 assert row['timestamp'] is not None - assert row['created_at'] is not None mlflow_repository_stub.promote_to_alias.assert_called_once() promote_kwargs = mlflow_repository_stub.promote_to_alias.call_args.kwargs @@ -220,10 +231,10 @@ async def test_minimal_retrain_failure_writes_report_without_version_columns( rows = ( conn.execute( text( - 'SELECT * FROM predictions_schema.retrain_reports ' + 'SELECT * FROM sientia_data.log_retrain ' 'WHERE model_id = :m' ), - {'m': model_id}, + {'m': str(model_id)}, ) .mappings() .all() @@ -231,7 +242,7 @@ async def test_minimal_retrain_failure_writes_report_without_version_columns( assert len(rows) == 1 row = rows[0] - assert row['model_id'] == model_id + assert row['model_id'] == str(model_id) assert row['model_name'] == 'test_model' assert 'training did not converge' in row['status'] assert row['version'] is None @@ -275,10 +286,10 @@ async def test_minimal_retrain_missing_target_writes_failure_report( row = ( conn.execute( text( - 'SELECT * FROM predictions_schema.retrain_reports ' + 'SELECT * FROM sientia_data.log_retrain ' 'WHERE model_id = :m' ), - {'m': model_id}, + {'m': str(model_id)}, ) .mappings() .first() @@ -313,7 +324,7 @@ async def test_minimal_retrain_no_training_data_does_not_persist_report( model_id = 731 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) _configure_retrain_happy_path(mlflow_repository_stub) @@ -329,7 +340,7 @@ async def test_minimal_retrain_no_training_data_does_not_persist_report( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.retrain_reports WHERE model_id = :m'), - {'m': model_id}, + text('SELECT COUNT(*) FROM sientia_data.log_retrain WHERE model_id = :m'), + {'m': str(model_id)}, ).scalar() assert count == 0 diff --git a/e2e/test_minio_offload.py b/e2e/test_minio_offload.py index 3b71ab7..cff0ee9 100644 --- a/e2e/test_minio_offload.py +++ b/e2e/test_minio_offload.py @@ -34,7 +34,7 @@ async def test_load_query_with_minio_offload_writes_object_to_bucket( """ model_id = 501 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) insert_sample_data(postgres_engine, model_id, [1.0, 2.0]) scenario_input = load_scenario_input('minio_offload_load_query.json', model_id=model_id) @@ -66,9 +66,9 @@ async def test_predictions_batch_with_minio_offload_path( """ model_id = 502 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) - conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}')) - conn.execute(text(f'DELETE FROM predictions_schema.transformed_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.predictions WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.transformed_data WHERE model_id = {model_id}')) insert_sample_data(postgres_engine, model_id, [10.0, 20.0, 30.0]) input_data = load_scenario_input('minio_offload_workflow.json', model_id=model_id) @@ -83,7 +83,7 @@ async def test_predictions_batch_with_minio_offload_path( with postgres_engine.connect() as conn: count = conn.execute( - text(f'SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = {model_id}') + text(f'SELECT COUNT(*) FROM sientia_data.predictions WHERE model_id = {model_id}') ).scalar() assert count == 1 @@ -97,7 +97,7 @@ async def test_load_query_with_inline_payload_when_below_threshold( """Scenario 4.2.1: payload stays inline when threshold is high enough.""" model_id = 503 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) insert_sample_data(postgres_engine, model_id, [1.0, 2.0]) scenario_input = load_scenario_input('minio_offload_load_query.json', model_id=model_id) diff --git a/e2e/test_predictions_batch_format_export.py b/e2e/test_predictions_batch_format_export.py index c09fc80..822efee 100644 --- a/e2e/test_predictions_batch_format_export.py +++ b/e2e/test_predictions_batch_format_export.py @@ -49,8 +49,8 @@ async def test_scenario_3_1_1_default_prediction_export( model_id = 311 with postgres_engine.begin() as conn: - conn.execute(text(f"DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}")) - conn.execute(text(f"DELETE FROM predictions_schema.transformed_data WHERE model_id = {model_id}")) + conn.execute(text(f"DELETE FROM sientia_data.predictions WHERE model_id = {model_id}")) + conn.execute(text(f"DELETE FROM sientia_data.transformed_data WHERE model_id = {model_id}")) insert_sample_data(postgres_engine, model_id, ['NULL', 78.2]) input_data = get_base_input_data(model_id) @@ -148,7 +148,7 @@ async def test_scenario_3_1_1_default_prediction_export( with postgres_engine.connect() as conn: tf_count = conn.execute( - text(f"SELECT COUNT(*) FROM predictions_schema.transformed_data WHERE model_id = {model_id}") + text(f"SELECT COUNT(*) FROM sientia_data.transformed_data WHERE model_id = {model_id}") ).scalar() assert tf_count == 0, 'transform export must be skipped when path_flag is set' @@ -407,7 +407,7 @@ async def test_scenario_3_1_5_export_without_transformed_data( insert_sample_data(postgres_engine, model_id, [23.5, 78.2]) with postgres_engine.begin() as conn: - conn.execute(text(f"DELETE FROM predictions_schema.transformed_data WHERE model_id = {model_id}")) + conn.execute(text(f"DELETE FROM sientia_data.transformed_data WHERE model_id = {model_id}")) input_data = get_base_input_data(model_id) input_data['save_transform'] = False # Don't save transformed data @@ -489,7 +489,7 @@ async def test_scenario_3_1_5_export_without_transformed_data( with postgres_engine.connect() as conn: result_query = conn.execute( - text(f"SELECT COUNT(*) FROM predictions_schema.transformed_data WHERE model_id = {model_id}") + text(f"SELECT COUNT(*) FROM sientia_data.transformed_data WHERE model_id = {model_id}") ) count = result_query.scalar() assert count == 0, f"Expected transform table to be empty, but found {count} records" diff --git a/e2e/test_predictions_batch_main_workflow.py b/e2e/test_predictions_batch_main_workflow.py index 89b81df..3eeb0d5 100644 --- a/e2e/test_predictions_batch_main_workflow.py +++ b/e2e/test_predictions_batch_main_workflow.py @@ -27,9 +27,9 @@ async def test_scenario_1_1_1_happy_path_complete_success( client = temporal_test_env.client with postgres_engine.begin() as conn: - conn.execute(text('DELETE FROM predictions_schema.laborious_data WHERE model_id = 123')) + conn.execute(text('DELETE FROM sientia_data.laborious_data WHERE model_id = 123')) insert_sql = """ - INSERT INTO predictions_schema.laborious_data (model_id, variable, value, timestamp, created_at) + INSERT INTO sientia_data.laborious_data (model_id, variable, value, timestamp, created_at) VALUES (123, 'sensor_1', 23.5, '2024-01-01 12:00:00+00:00', '2024-01-01 12:00:00+00:00'), (123, 'sensor_2', 78.2, '2024-01-01 12:00:00+00:00', '2024-01-01 12:00:00+00:00'), @@ -46,7 +46,7 @@ async def test_scenario_1_1_1_happy_path_complete_success( make_workflow_id('test-predictions-batch'), ) - schema_name = 'predictions_schema' + schema_name = 'sientia_data' with postgres_engine.connect() as conn: result_query = conn.execute( text( @@ -101,7 +101,7 @@ async def test_scenario_1_2_1_sql_query_execution_error( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = 128') + text('SELECT COUNT(*) FROM sientia_data.predictions WHERE model_id = 128') ).scalar() assert count == 0 @@ -131,7 +131,7 @@ async def test_scenario_1_2_2_missing_required_parameters( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = 129') + text('SELECT COUNT(*) FROM sientia_data.predictions WHERE model_id = 129') ).scalar() assert count == 0 @@ -150,11 +150,11 @@ async def test_scenario_1_2_3_invalid_datetime_column_specification( client = temporal_test_env.client with postgres_engine.begin() as conn: - conn.execute(text('DELETE FROM predictions_schema.laborious_data WHERE model_id = 130')) + conn.execute(text('DELETE FROM sientia_data.laborious_data WHERE model_id = 130')) conn.execute( text( """ - INSERT INTO predictions_schema.laborious_data (model_id, variable, value, timestamp, created_at) + INSERT INTO sientia_data.laborious_data (model_id, variable, value, timestamp, created_at) VALUES (130, 'sensor_1', 23.5, '2024-01-01 12:00:00+00:00', '2024-01-01 12:00:00+00:00') """ ) @@ -174,7 +174,7 @@ async def test_scenario_1_2_3_invalid_datetime_column_specification( with postgres_engine.connect() as conn: count = conn.execute( - text('SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = 130') + text('SELECT COUNT(*) FROM sientia_data.predictions WHERE model_id = 130') ).scalar() assert count == 0 diff --git a/e2e/test_predictions_batch_prediction_process.py b/e2e/test_predictions_batch_prediction_process.py index 5148670..2b67aa1 100644 --- a/e2e/test_predictions_batch_prediction_process.py +++ b/e2e/test_predictions_batch_prediction_process.py @@ -30,9 +30,9 @@ def get_base_input_data(model_id): def insert_sample_prediction(postgres_engine, model_id): with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.predictions WHERE model_id = {model_id}')) insert_sql = f""" - INSERT INTO predictions_schema.predictions (model_id, timestamp, prediction, prediction_confidence, prediction_status, comments, response_time) + INSERT INTO sientia_data.predictions (model_id, timestamp, prediction, prediction_confidence, prediction_status, comments, response_time) VALUES ({model_id}, '2024-01-01 12:00:00+00:00', 10, 0, 'Good', '', 0.1) """ @@ -147,7 +147,7 @@ async def test_scenario_2_1_4_input_gate_repeat_without_prior_prediction( model_id = 214 insert_sample_data(postgres_engine, model_id, ['NULL', 78.2]) with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.predictions WHERE model_id = {model_id}')) input_data = get_base_input_data(model_id) input_data['input_filters']['SPECIFIC_VARIABLES_NULL_VALUES']['POLICY'] = 'REPEAT' await start_and_await_workflow( @@ -338,7 +338,7 @@ async def test_scenario_2_4_1_input_empty_data_stop( client = temporal_test_env.client model_id = 241 with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) input_data = get_base_input_data(model_id) input_data['input_filters'] = {'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}}} await start_and_await_workflow( diff --git a/e2e/test_simple_metrics.py b/e2e/test_simple_metrics.py index 2413e81..c232e65 100644 --- a/e2e/test_simple_metrics.py +++ b/e2e/test_simple_metrics.py @@ -4,7 +4,7 @@ End-to-end tests for the SimpleMetrics workflow. Coverage focus: - Happy path computes rmse/mse/mae/r2 from predictions joined against ``laborious_data`` - and persists rows to ``predictions_schema.simple_metrics_data`` with all required columns. + and persists rows to ``sientia_data.simple_metrics`` with all required columns. - Subset metric selection (only rmse) writes exactly the requested rows. - Zero-variance target produces ``r2=0`` per division-by-zero guard. - Empty join (no overlapping data) short-circuits without persisting anything. @@ -23,6 +23,7 @@ from e2e.helpers import load_scenario_input, make_workflow_id, start_and_await_w from laborious.activities.activities import Activities from laborious.workflows.simple_metrics import SimpleMetrics +# Columns defined by the production DDL for ``sientia_data.simple_metrics``. EXPECTED_SIMPLE_METRICS_COLUMNS = [ 'id', 'model_id', @@ -34,6 +35,11 @@ EXPECTED_SIMPLE_METRICS_COLUMNS = [ 'created_at', ] +# ``timestamp`` is now nullable per the new DDL (production code may write it +# null when the upstream data has no usable instant); skip the non-null check +# for it while still validating presence. +NULLABLE_SIMPLE_METRICS_COLUMNS = {'timestamp'} + def _simple_metrics_input(model_id: int, **overrides) -> dict: """Load and override the simple-metrics base scenario.""" @@ -87,8 +93,8 @@ def _seed_predictions_and_targets( ) with postgres_engine.begin() as conn: - conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}')) - conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.predictions WHERE model_id = {model_id}')) + conn.execute(text(f'DELETE FROM sientia_data.laborious_data WHERE model_id = {model_id}')) # The SimpleMetrics SQL JOIN only filters ``predictions.model_id``; it does # NOT filter ``laborious_data.model_id`` (see ``e2e/CODE_ISSUES.md`` issue # SM-1). Without this cross-model cleanup, a previous test's target rows @@ -96,7 +102,7 @@ def _seed_predictions_and_targets( # whenever timestamps happened to overlap. conn.execute( text( - "DELETE FROM predictions_schema.laborious_data " + "DELETE FROM sientia_data.laborious_data " "WHERE variable IN (:sensor_default, :target_name) " "AND timestamp >= NOW() - INTERVAL '120 minutes'" ), @@ -105,7 +111,7 @@ def _seed_predictions_and_targets( if prediction_rows: conn.execute( text( - 'INSERT INTO predictions_schema.predictions ' + 'INSERT INTO sientia_data.predictions ' '(model_id, prediction, prediction_confidence, response_time, ' 'prediction_status, "timestamp", created_at) VALUES ' + ', '.join(prediction_rows) @@ -113,7 +119,7 @@ def _seed_predictions_and_targets( ) conn.execute( text( - 'INSERT INTO predictions_schema.laborious_data ' + 'INSERT INTO sientia_data.laborious_data ' '(model_id, variable, value, "timestamp", created_at) VALUES ' + ', '.join(target_rows) ) @@ -132,8 +138,9 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns( """ Scenario S.1.1: rmse/mse/mae/r2 are calculated from a deterministic prediction/target pair set and written one row per metric. Every column - expected by ``predictions_schema.simple_metrics_data`` must be populated and - the numerical values must match closed-form expectations. + expected by ``sientia_data.simple_metrics`` must be populated (except the + nullable ``timestamp`` column) and the numerical values must match + closed-form expectations. """ client = temporal_test_env.client model_id = 511 @@ -169,10 +176,10 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns( rows = ( conn.execute( text( - 'SELECT * FROM predictions_schema.simple_metrics_data ' + 'SELECT * FROM sientia_data.simple_metrics ' 'WHERE model_id = :m ORDER BY metric' ), - {'m': model_id}, + {'m': str(model_id)}, ) .mappings() .all() @@ -183,6 +190,8 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns( assert column in rows[0], f'Missing simple_metrics column: {column}' for row in rows: for column in EXPECTED_SIMPLE_METRICS_COLUMNS: + if column in NULLABLE_SIMPLE_METRICS_COLUMNS: + continue assert row[column] is not None, f"Column '{column}' is NULL in {dict(row)}" by_metric = {row['metric']: row for row in rows} @@ -198,7 +207,9 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns( assert all(row['data_size'] == n for row in rows), 'data_size must equal target row count' assert all(row['interval_minutes'] == 60 for row in rows) - assert all(row['model_id'] == model_id for row in rows) + # ``model_id`` is now ``text`` in the new DDL, so we compare with the + # stringified test id rather than the numeric value. + assert all(row['model_id'] == str(model_id) for row in rows) @pytest.mark.asyncio @@ -232,10 +243,10 @@ async def test_simple_metrics_subset_metrics_writes_only_requested_rows( r[0] for r in conn.execute( text( - 'SELECT metric FROM predictions_schema.simple_metrics_data ' + 'SELECT metric FROM sientia_data.simple_metrics ' 'WHERE model_id = :m' ), - {'m': model_id}, + {'m': str(model_id)}, ).all() ] assert metrics == ['rmse'] @@ -270,10 +281,10 @@ async def test_simple_metrics_zero_variance_target_returns_zero_r2( with postgres_engine.connect() as conn: r2_value = conn.execute( text( - "SELECT value FROM predictions_schema.simple_metrics_data " + "SELECT value FROM sientia_data.simple_metrics " "WHERE model_id = :m AND metric = 'r2'" ), - {'m': model_id}, + {'m': str(model_id)}, ).scalar() assert r2_value is not None assert Decimal(str(r2_value)) == Decimal('0'), f'expected r2=0, got {r2_value!r}' @@ -314,9 +325,9 @@ async def test_simple_metrics_no_overlapping_data_short_circuits( with postgres_engine.connect() as conn: count = conn.execute( text( - 'SELECT COUNT(*) FROM predictions_schema.simple_metrics_data ' + 'SELECT COUNT(*) FROM sientia_data.simple_metrics ' 'WHERE model_id = :m' ), - {'m': model_id}, + {'m': str(model_id)}, ).scalar() assert count == 0, 'Empty target data must short-circuit and skip persistence' diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 8b17ad1..f19c86d 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -15,7 +15,7 @@ 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.temporal.constants import DATETIME_FORMAT_WITH_TZ - from sientia_model.analytics.model_analysis import ModelAnalysis + from sientia_model.analytics.drift_analysis import DriftAnalysis from laborious import metrics from laborious.utils.dataframe_debug import build_dataframe_debug_message @@ -96,14 +96,16 @@ class ModelMetrics(SientiaMonitoring): Args: - reference_data (DataFrame): Baseline dataset representing expected behavior. - target_data (DataFrame): Current analysis dataset to compare against reference. - - target_name (str): Target column name used by ``ModelAnalysis`` config. + - target_name (str): Target column name used by ``DriftAnalysis`` config. - reference_columns (Index): Feature columns evaluated for drift. - drift_metrics (list[str]): Enabled univariate methods. - chunk_period (str): Time bucket granularity used by analysis methods. - metadata (dict[str, Any]): Workflow metadata for logs and notifications. Return: - DataFrame: Consolidated drift dataframe ready for downstream formatting/persistence. + DataFrame: Consolidated drift dataframe from ``get_drift_metrics_dataframe`` using + ``method`` / ``value`` (and optional ``threshold``, ``drift_type``), ready for + activity-level formatting before Postgres export. """ config = { @@ -113,7 +115,7 @@ class ModelMetrics(SientiaMonitoring): 'features': reference_columns, } - model_analysis = ModelAnalysis(config=config) + drift_analysis = DriftAnalysis(config=config) self._debug_dataframe( f'Reference data: Size {reference_data.shape}', reference_data, metadata @@ -124,7 +126,7 @@ class ModelMetrics(SientiaMonitoring): core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift') start_time = time.time() try: - univariate_drift = model_analysis.detect_univariate_drift( + univariate_drift = drift_analysis.detect_univariate_drift( reference_df=reference_data, analysis_df=target_data, features=reference_columns, @@ -142,7 +144,7 @@ class ModelMetrics(SientiaMonitoring): core_labels = self.get_core_labels(metadata, operation_type='detect_multivariate_drift') start_time = time.time() try: - multivariate_drift = model_analysis.detect_multivariate_drift( + multivariate_drift = drift_analysis.detect_multivariate_drift( reference_df=reference_data, analysis_df=target_data, features=reference_columns, @@ -159,7 +161,7 @@ class ModelMetrics(SientiaMonitoring): start_time = time.time() core_labels = self.get_core_labels(metadata, operation_type='get_drift_metrics_dataframe') try: - drift_df = model_analysis.get_drift_metrics_dataframe( + drift_df = drift_analysis.get_drift_metrics_dataframe( univariate_drift=univariate_drift, multivariate_drift=multivariate_drift, ) @@ -179,12 +181,12 @@ class ModelMetrics(SientiaMonitoring): """ Parse ``series`` as datetime and return a TZ-naive UTC copy. - ``sientia_model.analytics.model_analysis.ModelAnalysis`` preserves the + ``sientia_model.analytics.drift_analysis.DriftAnalysis`` preserves the timezone of the input dataframe in its outputs, while target rows loaded from PostgreSQL come in with ``+00:00``. Forcing both sides of a comparison to TZ-naive UTC keeps ``isin`` / ``floor`` operations - deterministic regardless of how the analyzer (or a test double) - constructs its timestamps. + deterministic regardless of how the analyzer constructs its + timestamps. Args: - series (Series): Input series containing datetime-parseable values. @@ -228,7 +230,7 @@ class ModelMetrics(SientiaMonitoring): target_data = target_data.pivot(index='timestamp', columns='variable', values='value') target_data['timestamp'] = target_data.index - # Keep timestamps as datetime: ModelAnalysis._chunk_dataframe relies on + # Keep timestamps as datetime: DriftAnalysis._chunk_dataframe relies on # ``pd.Grouper(freq=...)`` which rejects string timestamp columns. target_data['timestamp'] = to_datetime(target_data['timestamp']) target_data = target_data.reset_index(drop=True) @@ -287,7 +289,7 @@ class ModelMetrics(SientiaMonitoring): return [] # Defense-in-depth: drop chunks whose floored timestamp does not appear - # in the analysis window. ``ModelAnalysis`` already chunks only over + # in the analysis window. ``DriftAnalysis`` already chunks only over # ``analysis_df`` so this only excludes rows injected by upstream # callers that pre-merge reference data into the result. target_floor = self._to_naive_utc(target_data['timestamp']).dt.floor(chunk_period) @@ -301,41 +303,30 @@ class ModelMetrics(SientiaMonitoring): ) return [] - # Map ``sientia_model.analytics.model_analysis`` schema onto the drift - # table columns: ``metric -> method``, ``statistic -> value``, - # ``alert -> drift``, ``chunk_index -> chunk``, - # ``chunk_end_date -> timestamp_end``. ``p_value`` and - # ``chunk_start_date`` are not persisted. - drift_df = drift_df.rename( - columns={ - 'metric': 'method', - 'statistic': 'value', - 'alert': 'drift', - 'chunk_index': 'chunk', - 'chunk_end_date': 'timestamp_end', - } - ) - drift_df.drop(columns=['p_value', 'chunk_start_date'], inplace=True, errors='ignore') + # Analyzer emits diagnostic columns that are not stored in ``sientia_data.drift_metrics``. + drift_df = drift_df.drop(columns=['threshold', 'drift_type'], errors='ignore') - # Drop duplicates - drift_df.drop_duplicates( - subset=['timestamp', 'method', 'feature'], keep='first', inplace=True - ) - - drift_df['model_id'] = model_id + drift_df['model_id'] = str(model_id) drift_df['accurate'] = accurate - drift_df['timestamp'] = self._to_naive_utc(drift_df['timestamp']) - drift_df['timestamp'] = drift_df['timestamp'].dt.tz_localize('UTC') - drift_df['timestamp'] = drift_df['timestamp'].dt.strftime(DATETIME_FORMAT_WITH_TZ) - - # ``timestamp_end`` may carry nanosecond precision (beyond - # ``timestamptz`` microseconds), so serialize as ISO text for the - # ``text`` Postgres column. - drift_df['timestamp_end'] = drift_df['timestamp_end'].apply( - lambda value: pd.Timestamp(value).isoformat() if pd.notna(value) else None + # ``timestamp`` is overridden with the most recent target instant so + # every persisted row shares a single business timestamp (the run's + # logical "now"), matching what downstream consumers expect. + latest_target_timestamp = self._to_naive_utc(target_data['timestamp']).max() + drift_df['timestamp'] = ( + pd.Timestamp(latest_target_timestamp) + .tz_localize('UTC') + .strftime(DATETIME_FORMAT_WITH_TZ) ) + # ``chunk_start_date`` / ``chunk_end_date`` may carry nanosecond + # precision (beyond ``timestamptz`` microseconds), so serialize as ISO + # text for the ``text`` Postgres columns. + for column in ('chunk_start_date', 'chunk_end_date'): + drift_df[column] = drift_df[column].apply( + lambda value: pd.Timestamp(value).isoformat() if pd.notna(value) else None + ) + self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata) return drift_df.to_dict(orient='records') diff --git a/tests/conftest.py b/tests/conftest.py index dd8f543..e46ac68 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,4 @@ import os -import sys -from unittest.mock import MagicMock from sientia_do.temporal.activities.postgres_sync import Postgres @@ -60,13 +58,9 @@ class DummyMinioDataFramePayload: """ Pytest configuration file with global mocks for external dependencies. -This module mocks the 'sientia' module to avoid requiring its installation -during unit tests. The mock is registered in sys.modules before any test -imports are executed. +The historical ``sientia`` package is no longer imported by the codebase; +drift analysis lives in ``sientia_model.analytics.drift_analysis`` and is +imported lazily inside Temporal activities. No global module-level mock is +required here — unit tests that need to control ``DriftAnalysis`` outputs +should patch ``laborious.activities.model_metrics.DriftAnalysis`` directly. """ - -# Mock sientia module -sientia_mock = MagicMock() -sientia_mock.ModelAnalysis = MagicMock -sys.modules['sientia'] = sientia_mock -sys.modules['sientia.ModelAnalysis'] = MagicMock() diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index 53c7ebb..eddc98d 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -1,8 +1,9 @@ from unittest.mock import ANY, MagicMock, patch -from pandas import DataFrame +from pandas import DataFrame, Timestamp from pytest import fixture from sientia_do.notifications.models import NotificationLevel +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ from laborious.activities.model_metrics import ModelMetrics @@ -72,39 +73,28 @@ def test_calculate_drift_invalid_chunk_period(model_metrics_activity): raise AssertionError('Expected ValueError') -@patch('laborious.activities.model_metrics.DataFrame') -@patch('laborious.activities.model_metrics.to_datetime') -def test_calculate_drift_with_reference_data( - mock_to_datetime, mock_dataframe, model_metrics_activity -): - # Arrange - mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' - mock_to_datetime.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - - mock_drift_df = MagicMock() - mock_drift_df.empty = False - mock_drift_df.drop.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.isin.return_value = [True] - mock_drift_df.__getitem__.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - mock_drift_df.rename.return_value = mock_drift_df - mock_drift_df.drop_duplicates.return_value = mock_drift_df - mock_drift_df.to_dict.return_value = [ +def _sample_drift_metrics_df(ts: Timestamp) -> DataFrame: + """Minimal analyzer-shaped dataframe (univariate row + columns the activity expects).""" + return DataFrame( { - 'method': 'ks_test', - 'value': 0.5, - 'feature': 'feature1', - 'timestamp': '2023-05-26 11:12:27+00:00', - 'model_id': 'test_model_id', - 'accurate': True, + 'timestamp': [ts], + 'feature': ['feature1'], + 'method': ['ks_test'], + 'value': [0.5], + 'alert': [False], + 'chunk_index': [0], + 'chunk_start_date': [ts], + 'chunk_end_date': [ts], + 'threshold': [0.1], + 'drift_type': ['univariate'], } - ] + ) - model_metrics_activity.get_drift_metrics = MagicMock(return_value=mock_drift_df) + +def test_calculate_drift_with_reference_data(model_metrics_activity): + ts = Timestamp('2023-05-26 11:12:27') + drift_df = _sample_drift_metrics_df(ts) + model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { @@ -114,20 +104,11 @@ def test_calculate_drift_with_reference_data( } ) - mock_target_df = MagicMock() - mock_target_df.pivot.return_value = mock_target_df - mock_target_df.index = ['2023-05-26 11:12:27'] - mock_target_df.reset_index.return_value = mock_target_df - mock_target_df.dropna.return_value = mock_target_df - mock_target_df.__getitem__.return_value.apply.return_value = ['2023-05-26 11:12:27'] - mock_target_df.drop.return_value.columns = ['feature1'] - mock_dataframe.return_value = mock_target_df - input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', - 'reference_data': reference_data.to_dict(), + 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], @@ -138,92 +119,40 @@ def test_calculate_drift_with_reference_data( 'chunk_period': 'min', } - # Act result = model_metrics_activity.calculate_drift(input_data) - # Assert - assert isinstance(result, list) - assert result == mock_drift_df.to_dict.return_value # type: ignore[comparison-overlap] - model_metrics_activity.info.assert_called() - model_metrics_activity.get_drift_metrics.assert_called_once() - # Verify transformations were called - mock_drift_df.drop.assert_called_once_with( - columns=['p_value', 'chunk_start_date'], inplace=True, errors='ignore' - ) - mock_drift_df.__getitem__.assert_called() - mock_drift_df.rename.assert_called_once_with( - columns={ - 'metric': 'method', - 'statistic': 'value', - 'alert': 'drift', - 'chunk_index': 'chunk', - 'chunk_end_date': 'timestamp_end', - } - ) - mock_drift_df.drop_duplicates.assert_called_once_with( - subset=['timestamp', 'method', 'feature'], keep='first', inplace=True - ) - mock_drift_df.to_dict.assert_called_once_with(orient='records') - - -@patch('laborious.activities.model_metrics.DataFrame') -@patch('laborious.activities.model_metrics.to_datetime') -def test_calculate_drift_without_reference_data( - mock_to_datetime, mock_dataframe, model_metrics_activity -): - # Arrange - mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' - mock_to_datetime.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - - mock_drift_df = MagicMock() - mock_drift_df.empty = False - mock_drift_df.drop.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.isin.return_value = [True] - mock_drift_df.__getitem__.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - mock_drift_df.rename.return_value = mock_drift_df - mock_drift_df.drop_duplicates.return_value = mock_drift_df - mock_drift_df.to_dict.return_value = [ + expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) + assert result == [ { + 'timestamp': expected_timestamp, + 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, - 'feature': 'feature1', - 'timestamp': '2023-05-26 11:12:27+00:00', + 'alert': False, + 'chunk_index': 0, + 'chunk_start_date': ts.isoformat(), + 'chunk_end_date': ts.isoformat(), 'model_id': 'test_model_id', - 'accurate': False, + 'accurate': True, } ] + model_metrics_activity.info.assert_called() + model_metrics_activity.get_drift_metrics.assert_called_once() - model_metrics_activity.get_drift_metrics = MagicMock(return_value=mock_drift_df) +def test_calculate_drift_without_reference_data(model_metrics_activity): + # Ten rows so int(len * 0.3) >= 1 for the built-in reference slice. + ts_last = Timestamp('2023-05-26 11:12:36') + drift_df = _sample_drift_metrics_df(ts_last) + model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) + + timestamps = [f'2023-05-26 11:12:{27 + i:02d}' for i in range(10)] target_data_dict = { - 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], - 'variable': ['feature1', 'feature1', 'feature1'], - 'value': [1.0, 2.0, 3.0], + 'timestamp': timestamps, + 'variable': ['feature1'] * 10, + 'value': [float(i) for i in range(10)], } - mock_target_df = MagicMock() - mock_target_df.pivot.return_value = mock_target_df - mock_target_df.index = ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'] - mock_target_df.reset_index.return_value = mock_target_df - mock_target_df.dropna.return_value = mock_target_df - mock_target_df.sort_values.return_value = mock_target_df - mock_target_df.head.return_value = DataFrame( - {'timestamp': ['2023-05-26 11:12:27'], 'feature1': [1.0]} - ) - mock_target_df.__getitem__.return_value.apply.return_value = [ - '2023-05-26 11:12:27', - '2023-05-26 11:12:28', - '2023-05-26 11:12:29', - ] - mock_target_df.drop.return_value.columns = ['feature1'] - mock_dataframe.return_value = mock_target_df - mock_dataframe.side_effect = lambda x=None: mock_target_df if x is not None else mock_target_df - input_data = { **metadata, 'model_name': 'test_model', @@ -235,12 +164,23 @@ def test_calculate_drift_without_reference_data( 'chunk_period': 's', } - # Act result = model_metrics_activity.calculate_drift(input_data) - # Assert - assert isinstance(result, list) - assert result == mock_drift_df.to_dict.return_value + expected_timestamp = ts_last.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) + assert result == [ + { + 'timestamp': expected_timestamp, + 'feature': 'feature1', + 'method': 'ks_test', + 'value': 0.5, + 'alert': False, + 'chunk_index': 0, + 'chunk_start_date': ts_last.isoformat(), + 'chunk_end_date': ts_last.isoformat(), + 'model_id': 'test_model_id', + 'accurate': False, + } + ] model_metrics_activity.warning.assert_called() model_metrics_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], @@ -250,24 +190,6 @@ def test_calculate_drift_without_reference_data( level=NotificationLevel.WARNING, attachment_content=ANY, ) - # Verify transformations were called - mock_drift_df.drop.assert_called_once_with( - columns=['p_value', 'chunk_start_date'], inplace=True, errors='ignore' - ) - mock_drift_df.__getitem__.assert_called() - mock_drift_df.rename.assert_called_once_with( - columns={ - 'metric': 'method', - 'statistic': 'value', - 'alert': 'drift', - 'chunk_index': 'chunk', - 'chunk_end_date': 'timestamp_end', - } - ) - mock_drift_df.drop_duplicates.assert_called_once_with( - subset=['timestamp', 'method', 'feature'], keep='first', inplace=True - ) - mock_drift_df.to_dict.assert_called_once_with(orient='records') @patch('laborious.activities.model_metrics.DataFrame') @@ -391,44 +313,18 @@ def test_calculate_drift_empty_after_timestamp_filter( 'No drift metrics found after dropping rows where timestamp is not in target data', metadata['metadata'], ) - # When the timestamp filter empties the dataframe, the rename/drop pipeline - # is short-circuited, so neither ``drop`` nor ``rename`` should run. + # When the timestamp filter empties the dataframe, the post-filter + # pipeline is short-circuited, so neither ``drop`` nor ``rename`` runs + # (they wouldn't run anyway, as the activity preserves the lib's schema). mock_drift_df.drop.assert_not_called() mock_drift_df.rename.assert_not_called() mock_drift_df.__getitem__.assert_called() -@patch('laborious.activities.model_metrics.DataFrame') -@patch('laborious.activities.model_metrics.to_datetime') -def test_calculate_drift_success_min(mock_to_datetime, mock_dataframe, model_metrics_activity): - # Arrange - mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' - mock_to_datetime.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - - mock_drift_df = MagicMock() - mock_drift_df.empty = False - mock_drift_df.drop.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.isin.return_value = [True] - mock_drift_df.__getitem__.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - mock_drift_df.rename.return_value = mock_drift_df - mock_drift_df.drop_duplicates.return_value = mock_drift_df - mock_drift_df.to_dict.return_value = [ - { - 'method': 'ks_test', - 'value': 0.5, - 'feature': 'feature1', - 'timestamp': '2023-05-26 11:12:27+00:00', - 'model_id': 'test_model_id', - 'accurate': True, - } - ] - - model_metrics_activity.get_drift_metrics = MagicMock(return_value=mock_drift_df) +def test_calculate_drift_success_min(model_metrics_activity): + ts = Timestamp('2023-05-26 11:12:27') + drift_df = _sample_drift_metrics_df(ts) + model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { @@ -438,20 +334,11 @@ def test_calculate_drift_success_min(mock_to_datetime, mock_dataframe, model_met } ) - mock_target_df = MagicMock() - mock_target_df.pivot.return_value = mock_target_df - mock_target_df.index = ['2023-05-26 11:12:27'] - mock_target_df.reset_index.return_value = mock_target_df - mock_target_df.dropna.return_value = mock_target_df - mock_target_df.__getitem__.return_value.isin.return_value = [True] - mock_target_df.drop.return_value.columns = ['feature1'] - mock_dataframe.return_value = mock_target_df - input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', - 'reference_data': reference_data.to_dict(), + 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], @@ -462,65 +349,31 @@ def test_calculate_drift_success_min(mock_to_datetime, mock_dataframe, model_met 'chunk_period': 'min', } - # Act result = model_metrics_activity.calculate_drift(input_data) - # Assert - assert isinstance(result, list) - assert result == mock_drift_df.to_dict.return_value # type: ignore[comparison-overlap] - model_metrics_activity.info.assert_called() - model_metrics_activity.get_drift_metrics.assert_called_once() - # Verify transformations were called - mock_drift_df.drop.assert_called_once_with( - columns=['p_value', 'chunk_start_date'], inplace=True, errors='ignore' - ) - mock_drift_df.__getitem__.assert_called() - mock_drift_df.rename.assert_called_once_with( - columns={ - 'metric': 'method', - 'statistic': 'value', - 'alert': 'drift', - 'chunk_index': 'chunk', - 'chunk_end_date': 'timestamp_end', - } - ) - mock_drift_df.drop_duplicates.assert_called_once_with( - subset=['timestamp', 'method', 'feature'], keep='first', inplace=True - ) - mock_drift_df.to_dict.assert_called_once_with(orient='records') - - -@patch('laborious.activities.model_metrics.DataFrame') -@patch('laborious.activities.model_metrics.to_datetime') -def test_calculate_drift_success_s(mock_to_datetime, mock_dataframe, model_metrics_activity): - # Arrange - mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' - mock_to_datetime.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - - mock_drift_df = MagicMock() - mock_drift_df.empty = False - mock_drift_df.drop.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.isin.return_value = [True] - mock_drift_df.__getitem__.return_value = mock_drift_df - mock_drift_df.__getitem__.return_value.dt.tz_localize.return_value.dt.strftime.return_value = ( - '2023-05-26 11:12:27+00:00' - ) - mock_drift_df.rename.return_value = mock_drift_df - mock_drift_df.drop_duplicates.return_value = mock_drift_df - mock_drift_df.to_dict.return_value = [ + expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) + assert result == [ { + 'timestamp': expected_timestamp, + 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, - 'feature': 'feature1', - 'timestamp': '2023-05-26 11:12:27+00:00', + 'alert': False, + 'chunk_index': 0, + 'chunk_start_date': ts.isoformat(), + 'chunk_end_date': ts.isoformat(), 'model_id': 'test_model_id', 'accurate': True, } ] + model_metrics_activity.info.assert_called() + model_metrics_activity.get_drift_metrics.assert_called_once() - model_metrics_activity.get_drift_metrics = MagicMock(return_value=mock_drift_df) + +def test_calculate_drift_success_s(model_metrics_activity): + ts = Timestamp('2023-05-26 11:12:27') + drift_df = _sample_drift_metrics_df(ts) + model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { @@ -530,20 +383,11 @@ def test_calculate_drift_success_s(mock_to_datetime, mock_dataframe, model_metri } ) - mock_target_df = MagicMock() - mock_target_df.pivot.return_value = mock_target_df - mock_target_df.index = ['2023-05-26 11:12:27'] - mock_target_df.reset_index.return_value = mock_target_df - mock_target_df.dropna.return_value = mock_target_df - mock_target_df.__getitem__.return_value.isin.return_value = [True] - mock_target_df.drop.return_value.columns = ['feature1'] - mock_dataframe.return_value = mock_target_df - input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', - 'reference_data': reference_data.to_dict(), + 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], @@ -554,32 +398,25 @@ def test_calculate_drift_success_s(mock_to_datetime, mock_dataframe, model_metri 'chunk_period': 's', } - # Act result = model_metrics_activity.calculate_drift(input_data) - # Assert - assert isinstance(result, list) - assert result == mock_drift_df.to_dict.return_value # type: ignore[comparison-overlap] + expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) + assert result == [ + { + 'timestamp': expected_timestamp, + 'feature': 'feature1', + 'method': 'ks_test', + 'value': 0.5, + 'alert': False, + 'chunk_index': 0, + 'chunk_start_date': ts.isoformat(), + 'chunk_end_date': ts.isoformat(), + 'model_id': 'test_model_id', + 'accurate': True, + } + ] model_metrics_activity.info.assert_called() model_metrics_activity.get_drift_metrics.assert_called_once() - # Verify transformations were called - mock_drift_df.drop.assert_called_once_with( - columns=['p_value', 'chunk_start_date'], inplace=True, errors='ignore' - ) - mock_drift_df.__getitem__.assert_called() - mock_drift_df.rename.assert_called_once_with( - columns={ - 'metric': 'method', - 'statistic': 'value', - 'alert': 'drift', - 'chunk_index': 'chunk', - 'chunk_end_date': 'timestamp_end', - } - ) - mock_drift_df.drop_duplicates.assert_called_once_with( - subset=['timestamp', 'method', 'feature'], keep='first', inplace=True - ) - mock_drift_df.to_dict.assert_called_once_with(orient='records') @patch('laborious.activities.model_metrics.DataFrame') @@ -646,7 +483,7 @@ def test_calculate_drift_get_drift_metrics_error( @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') -@patch('laborious.activities.model_metrics.ModelAnalysis') +@patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_success( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity @@ -657,8 +494,8 @@ def test_get_drift_metrics_success( mock_drift_df = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], - 'metric': ['ks_test'], - 'statistic': [0.5], + 'method': ['ks_test'], + 'value': [0.5], 'feature': ['feature1'], } ) @@ -707,7 +544,7 @@ def test_get_drift_metrics_success( @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') -@patch('laborious.activities.model_metrics.ModelAnalysis') +@patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_univariate_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity @@ -764,7 +601,7 @@ def test_get_drift_metrics_univariate_error( @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') -@patch('laborious.activities.model_metrics.ModelAnalysis') +@patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_multivariate_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity @@ -810,7 +647,7 @@ def test_get_drift_metrics_multivariate_error( @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') -@patch('laborious.activities.model_metrics.ModelAnalysis') +@patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_dataframe_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index 62a267d..9583048 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -125,6 +125,8 @@ def test_build_minio_config_with_env_vars(): environ['MINIO_SECRET_KEY'] = 'test-secret' environ['MINIO_REGION_NAME'] = 'test-region' environ['MINIO_DEFAULT_BUCKET'] = 'test-bucket' + # Isolate from IDE/CI env (e.g. VS Code may export MINIO_SECURE=true). + environ['MINIO_SECURE'] = 'false' assert build_minio_config() == { 'endpoint_url': 'http://test-host', 'access_key': 'test-key', @@ -141,6 +143,8 @@ def test_build_minio_config_with_defaults(): environ.pop('MINIO_SECRET_KEY', None) environ.pop('MINIO_REGION_NAME', None) environ.pop('MINIO_DEFAULT_BUCKET', None) + environ.pop('MINIO_SECURE', None) + environ.pop('MINIO_RETENTION_HOURS', None) assert build_minio_config() == { 'endpoint_url': 'http://localhost:9000', 'access_key': 'minioadmin',