From 9f218af6b03e80f3a81fbbb462a9218c331c1ff2 Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 8 Oct 2026 16:43:18 +0200 Subject: [PATCH] rc/SIENTIAPDE-2167- test: add e2e coverage for metadata_id reaching predictions.metadata_id --- e2e/conftest.py | 3 +- ...st_predictions_batch_prediction_process.py | 71 +++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/e2e/conftest.py b/e2e/conftest.py index 807a5ce..0295326 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -368,6 +368,7 @@ def _worker_activity_list(test_activities: Activities): test_activities.mlflow_content_gate, test_activities.request_predict, test_activities.repeat_last_prediction, + test_activities.write_prediction_metadata, test_activities.format_prediction, test_activities.format_transformed_data, test_activities.format_default_prediction, @@ -624,5 +625,3 @@ async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc): activity_executor=activity_executor, ) as worker: yield worker - - diff --git a/e2e/test_predictions_batch_prediction_process.py b/e2e/test_predictions_batch_prediction_process.py index f15621a..c3f4767 100644 --- a/e2e/test_predictions_batch_prediction_process.py +++ b/e2e/test_predictions_batch_prediction_process.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch import numpy as np import pandas as pd import pytest +from pymongo import MongoClient from sqlalchemy import text from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker @@ -86,6 +87,76 @@ async def test_scenario_2_1_1_input_gate_triggers_continue( mlflow_repository_stub.stub_wrapper.predict.assert_not_called() +@pytest.mark.asyncio +@pytest.mark.integration +async def test_e2e_metadata_id_reaches_postgres_when_gate_triggers( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, + mongo_container, + mlflow_repository_stub, +): + """ + input_gate CONTINUE: the gate's document is written to Mongo, and the resulting + metadata_id lands on the predictions row — independently verified on both sides. + """ + client = temporal_test_env.client + model_id = 261 + # sensor_1 NULL, sensor_2 present — matches `input_filters.SPECIFIC_VARIABLES_NULL_VALUES` + # (`variables: ["sensor_1"]`, POLICY CONTINUE) in prediction_process_base.json. + insert_sample_data(postgres_engine, model_id, ['NULL', 78.2]) + input_data = get_base_input_data(model_id) + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-continue') + ) + assert_continue(postgres_engine, model_id) + + with postgres_engine.connect() as conn: + metadata_id = conn.execute( + text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}') + ).scalar() + assert metadata_id is not None, 'Expected a metadata_id on the predictions row' + + # Independent readback: a fresh client against the same testcontainer, not the + # `mongo_repository`/`PredictionMetadata` instances the workflow itself wrote through. + mongo_port = mongo_container.get_exposed_port(27017) + verify_client = MongoClient(f'mongodb://localhost:{mongo_port}') + try: + document = verify_client['test_db']['prediction_metadata'].find_one({'_id': metadata_id}) + finally: + verify_client.close() + assert document is not None, f'Expected a prediction_metadata document for id {metadata_id}' + assert document['filters'] == [ + {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 1} + ] + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_e2e_metadata_id_is_null_when_nothing_triggers( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, +): + """Clean prediction, no gate triggers: predictions.metadata_id stays NULL.""" + client = temporal_test_env.client + model_id = 262 + insert_sample_data(postgres_engine, model_id, [60.0, 78.2]) + input_data = get_base_input_data(model_id) + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-none') + ) + assert_prediction(postgres_engine, model_id, prediction=0.5, prediction_confidence=0) + + with postgres_engine.connect() as conn: + metadata_id = conn.execute( + text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}') + ).scalar() + assert metadata_id is None, f'Expected metadata_id to be NULL, got {metadata_id!r}' + + @pytest.mark.asyncio @pytest.mark.integration async def test_scenario_2_1_2_input_gate_triggers_stop(