test(e2e): add batch-size cap coverage for SIENTIAPDE-2013

Also fixes the test_activities fixture, which built mongodb_config by
hand and was missing the new required notification_queue_batch_size key
-- every e2e test would have hit a KeyError without this.

New test seeds 5 notifications with a batch size of 2 and confirms three
successive runs return them oldest-first in order (n1,n2 / n3,n4 / n5),
proving the cursor only advances to what was actually returned rather
than skipping documents a capped query didn't fetch. Verified against
real testcontainers Mongo/Redis/Temporal, not mocks.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-17 09:39:35 -03:00
parent 9c153041ad
commit d46bd55cb4
3 changed files with 65 additions and 0 deletions

View File

@@ -330,6 +330,7 @@ async def test_activities(
'connection_string': f'mongodb://localhost:{mongo_port}', 'connection_string': f'mongodb://localhost:{mongo_port}',
'database_name': E2E_DATABASE, 'database_name': E2E_DATABASE,
'ttl_index_seconds': 3600, 'ttl_index_seconds': 3600,
'notification_queue_batch_size': 0,
}, },
email_config={ email_config={
'sender_email': 'e2e@example.com', 'sender_email': 'e2e@example.com',

View File

@@ -110,6 +110,7 @@ No notifications → no SMTP, no Postgres row.
- Prior timestamp → only newer notifications returned. - Prior timestamp → only newer notifications returned.
- Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch. - Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch.
- Empty Mongo → no Redis timestamp write. - Empty Mongo → no Redis timestamp write.
- Batch size cap (SIENTIAPDE-2013) → with 5 pending notifications and a batch size of 2, each run returns exactly the next-oldest 2 (sorted ascending by timestamp), and the Redis cursor advances only to what was actually returned, so three successive runs drain the backlog in order instead of skipping anything.
### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py` ### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py`

View File

@@ -228,3 +228,66 @@ async def test_load_package_empty_mongo_no_redis_write(
) )
assert redis_client.get('notification_last_timestamp:Alerts') is None assert redis_client.get('notification_last_timestamp:Alerts') is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_load_package_batch_size_caps_and_returns_oldest_first(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
redis_client: Redis,
test_activities,
):
"""
SIENTIAPDE-2013: a capped run returns the oldest pending documents first,
and the incremental cursor picks up the remainder on the next run.
Five notifications exist, but the batch size is capped at 2. Each run must
return exactly the next-oldest 2 (never an arbitrary subset), and the
Redis cursor must advance only to the max timestamp of what was actually
returned -- not the true max across all 5 -- or the skipped documents
would never be fetched again.
"""
test_activities.notification_queue_batch_size = 2
seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
seed_notifications(
mongo_uri,
E2E_DATABASE,
[
default_notification(
notification_id=f'n{i}',
timestamp=datetime(2024, 6, 1, 12, i, 0, tzinfo=UTC),
)
for i in range(1, 6)
],
)
input_data = load_scenario_input('subworkflow_load_notification_package.json')
_build_metadata(input_data)
first_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-1'),
)
assert [n['notification_id'] for n in first_run['notification_package']] == ['n1', 'n2']
second_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-2'),
)
assert [n['notification_id'] for n in second_run['notification_package']] == ['n3', 'n4']
third_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-3'),
)
assert [n['notification_id'] for n in third_run['notification_package']] == ['n5']