diff --git a/e2e/conftest.py b/e2e/conftest.py index ab81973..0a67ccf 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -92,9 +92,7 @@ async def register_namespace_if_missing(env: WorkflowEnvironment, namespace: str raise TimeoutError(f'Namespace {namespace} not ready within 5s') -async def ensure_namespace_search_attributes( - env: WorkflowEnvironment, namespace: str -) -> None: +async def ensure_namespace_search_attributes(env: WorkflowEnvironment, namespace: str) -> None: """ Register the orchestrator search attributes on a namespace, if missing. diff --git a/e2e/scenario_inputs/orchestrator_drift_window.json b/e2e/scenario_inputs/orchestrator_drift_window.json new file mode 100644 index 0000000..a034a9c --- /dev/null +++ b/e2e/scenario_inputs/orchestrator_drift_window.json @@ -0,0 +1,38 @@ +{ + "workflow_input": { + "schedule_name": "orchestrator-e2e-drift-window", + "pipelines_query": { + "collection": "pipelines", + "aggregation": [{"$match": {"active": true}}] + }, + "opc_servers_query": { + "collection": "opc_servers", + "filters": {"active": true} + } + }, + "pipelines": [ + { + "schedule_name": "model-dr-drift-default", + "workflow_type": "drift", + "model_id": "model-dr", + "model": {"name": "Model model-dr"}, + "active": true, + "updated_at": "@now", + "frequency": "1m", + "chunk_period": "hour" + }, + { + "schedule_name": "model-dr-drift-explicit", + "workflow_type": "drift", + "model_id": "model-dr", + "model": {"name": "Model model-dr"}, + "active": true, + "updated_at": "@now", + "frequency": "1m", + "interval_minutes": 60, + "chunk_period": "s" + } + ], + "opc_servers": [], + "active_ingestors": ["heartbeat:ingestor:1"] +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index b09f2d6..f0b74d5 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -56,6 +56,14 @@ Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_ window → next run switches only that schedule to `frequency_alert`; the recovery write switches it back to `frequency`. +### 1.8.1 Drift window and chunk_period + +Two `drift` pipelines on `frequency='1m'`: one without `interval_minutes` and with an invalid +`chunk_period='hour'` → workflow input `interval=2` (frequency + 1-minute margin), +`chunk_period='min'` and exactly one `DRIFT_CHUNK_PERIOD_INVALID` notification; one with +`interval_minutes=60` and `chunk_period='s'` → both forwarded unchanged, no warning +(SIENTIAPDE-2212). + --- ## 2. Alerts workflow diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index 99e9414..bdbc505 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -313,3 +313,54 @@ async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip( _laborious_adaptive_write(mongo_uri, short, escalated=False) await run('recover') assert await _schedule_every(temporal_host, short) == timedelta(hours=1) + + +async def _schedule_input(host: str, schedule_id: str) -> dict: + """Return the decoded workflow input of a laborious-namespace schedule.""" + client = await Client.connect(host, namespace='laborious') + description = await client.get_schedule_handle(schedule_id).describe() + action = description.schedule.action + assert isinstance(action, ScheduleActionStartWorkflow) + [input_data] = await client.data_converter.decode(action.args) + return input_data + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_scenario_1_8_1_drift_window_and_chunk_period( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + temporal_host: str, + redis_client: Redis, + notification_handler, +): + """1.8.1: Drift window defaults to frequency + 1 min margin; invalid chunk_period warns and falls back.""" + scenario = load_scenario_input('orchestrator_drift_window') + _apply_orchestrator_seeds(scenario, mongo_uri, redis_client) + + await start_and_await_workflow( + temporal_env.client, + Orchestrator.run, + scenario['workflow_input'], + make_workflow_id('orchestrator-drift-window'), + ) + + default = await _schedule_input(temporal_host, 'model-dr-drift-default') + assert default['interval'] == 2 + assert default['chunk_period'] == 'min' + assert default['chunk_period_configured'] == 'hour' + + explicit = await _schedule_input(temporal_host, 'model-dr-drift-explicit') + assert explicit['interval'] == 60 + assert explicit['chunk_period'] == 's' + assert 'chunk_period_configured' not in explicit + + warnings = list( + notification_handler.mongo_collection.find( + {'notification_id': 'DRIFT_CHUNK_PERIOD_INVALID'} + ) + ) + assert len(warnings) == 1 + assert warnings[0]['model_id'] == 'model-dr'