rc/SIENTIAPDE-2167- test: add e2e coverage for metadata_id reaching predictions.metadata_id

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(