diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 6f439a0..b09f2d6 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -50,6 +50,12 @@ First run creates TTL index on notification collection used by scouter pipelines Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one Temporal schedule per window, each with its own `frequency` as interval. +### 1.7.2 Adaptive frequency escalation/recovery round trip + +Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_at`) on one +window → next run switches only that schedule to `frequency_alert`; the recovery write switches it +back to `frequency`. + --- ## 2. Alerts workflow diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index 3f2d0ce..99e9414 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -1,6 +1,6 @@ """E2E tests for the Orchestrator main workflow.""" -from datetime import timedelta +from datetime import UTC, datetime, timedelta import pytest from pymongo import MongoClient @@ -259,3 +259,57 @@ async def test_scenario_1_7_1_simple_metrics_parallel_windows( short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m' assert await _schedule_every(temporal_host, short) == timedelta(hours=1) assert await _schedule_every(temporal_host, long) == timedelta(days=1) + + +def _laborious_adaptive_write(mongo_uri: str, schedule_name: str, escalated: bool) -> None: + """Replicates laborious `update_simple_metrics_adaptive_state` on an escalated flip.""" + mongo = MongoClient(mongo_uri) + try: + mongo[E2E_DATABASE]['pipelines'].update_one( + {'schedule_name': schedule_name}, + { + '$set': { + 'adaptive_state': {'escalated': escalated, 'healthy_streak': 0}, + # BSON datetime, as laborious writes it; a string would break the tick. + 'updated_at': datetime.now(UTC), + } + }, + ) + finally: + mongo.close() + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + temporal_host: str, + redis_client: Redis, +): + """1.7.2: Escalation written by laborious switches only that window to frequency_alert; recovery switches it back.""" + scenario = load_scenario_input('orchestrator_simple_metrics_windows') + _apply_orchestrator_seeds(scenario, mongo_uri, redis_client) + short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m' + + async def run(tag: str) -> None: + await start_and_await_workflow( + temporal_env.client, + Orchestrator.run, + scenario['workflow_input'], + make_workflow_id(f'orchestrator-sm-adaptive-{tag}'), + ) + + await run('create') + assert await _schedule_every(temporal_host, short) == timedelta(hours=1) + + _laborious_adaptive_write(mongo_uri, short, escalated=True) + await run('escalate') + assert await _schedule_every(temporal_host, short) == timedelta(minutes=5) + assert await _schedule_every(temporal_host, long) == timedelta(days=1) + + _laborious_adaptive_write(mongo_uri, short, escalated=False) + await run('recover') + assert await _schedule_every(temporal_host, short) == timedelta(hours=1)