From 14b854148aa83c17929efcb6ca2174b65e3a2a97 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Mon, 14 Sep 2026 12:15:23 -0300 Subject: [PATCH] SIENTIAPDE-2011: add e2e coverage for query-time filter, fix stale e2e fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit e2e fixtures (default_notification, seed_last_timestamp, default_receiver_group) still assumed the pre-SIENTIAPDE-2007 string-timestamp representation and the old ignore_models field name, so the load_notification_package e2e suite was failing independently of this ticket's changes: - default_notification now seeds `timestamp` as a native datetime (matches production since SIENTIAPDE-2007's Notification.timestamp = now()). - seed_last_timestamp now uses DATETIME_FORMAT_MS_WITH_TZ (matches what put_last_data_timestamp actually writes to Redis today). - default_receiver_group's ignore_models renamed to ignore_models_id, per this ticket's decision. - test_reports_main_workflow.py's explicit timestamp= call sites updated to match. Also fixes a real production bug in put_last_data_timestamp (slot_manager.py) surfaced by running the e2e suite against a real Temporal environment: Temporal's default data converter serializes datetime values to ISO-8601 strings across the activity boundary, and pymongo returns naive datetimes for BSON Date fields, so `data['timestamp'].max().strftime(...)` crashed on a plain str. Unit tests never caught this because they mock the activity boundary and never cross real (de)serialization. Fixed by normalizing through `pandas.to_datetime(..., utc=True)`, which handles native datetime/Timestamp and offset-aware/naive ISO strings uniformly. Adds test_load_package_query_time_filter_union_and_intersection, a new e2e test exercising the actual SIENTIAPDE-2011 behavior end-to-end: union of active groups' levels, intersection of their ignore_models_id, and the requirement that a model ignored by only one of several active groups must not be excluded from the fetch. Note: test_reports_main_workflow.py / test_alerts_main_workflow.py still have a separate, pre-existing failure in process_notifications.py (same family of bug: a notification's timestamp field parsed with a fixed strptime format after crossing the same Temporal serialization boundary) — confirmed present on the unmodified baseline too, unrelated to this ticket, tracked separately. Co-Authored-By: Claude Sonnet 5 --- e2e/helpers.py | 44 +++---- e2e/scenarios.md | 8 +- e2e/test_reports_main_workflow.py | 5 +- ...t_subworkflow_load_notification_package.py | 110 +++++++++++++++--- orchestrator/activities/slot_manager.py | 23 +++- 5 files changed, 143 insertions(+), 47 deletions(-) diff --git a/e2e/helpers.py b/e2e/helpers.py index 3af6f87..cafe1ce 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -29,7 +29,6 @@ MONGO_COLLECTIONS = ( ) DATETIME_FORMAT_MS_WITH_TZ = '%Y-%m-%d %H:%M:%S.%f%z' -DATETIME_FORMAT_WITH_TZ = '%Y-%m-%d %H:%M:%S%z' _TIMESTAMP_MARKER_PATTERN = re.compile(r'^@now(?:([+-])(\d+)([smhd]))?$') _TIMESTAMP_UNIT_TO_KWARG = {'s': 'seconds', 'm': 'minutes', 'h': 'hours', 'd': 'days'} @@ -222,17 +221,19 @@ def seed_last_timestamp(redis_client: Redis, mail_type: str, value: str) -> None Set notification_last_timestamp for a mail type, JSON-encoded. Values must be JSON-encoded so ``redis_repository.get`` (which calls - ``json.loads`` on the raw payload) can deserialize them. The value - must follow the exact format ``sientia_do.notifications.models.Notification`` - writes into ``notification_queue.timestamp``: ``DATETIME_FORMAT_WITH_TZ`` - (e.g. ``"2026-05-22 16:47:02+0000"``) — no microseconds and no colon in - the timezone offset. + ``json.loads`` on the raw payload) can deserialize them. The value must + follow the exact format ``put_last_data_timestamp`` + (``orchestrator/activities/slot_manager.py``) writes into Redis today: + ``DATETIME_FORMAT_MS_WITH_TZ`` (e.g. ``"2026-05-22 16:47:02.000000+0000"``) + — microseconds included, no colon in the timezone offset. ``load_latest_data`` + parses this same format back into a ``datetime`` for the ``$gt`` comparison, + so the two sides must agree exactly. Args: redis_client: Redis client connected to the test instance. mail_type: Mail type identifier (e.g. ``"Alerts"``, ``"Reports"``). - value: Timestamp string in ``DATETIME_FORMAT_WITH_TZ`` - (e.g. ``"2024-06-01 10:30:00+0000"``). + value: Timestamp string in ``DATETIME_FORMAT_MS_WITH_TZ`` + (e.g. ``"2024-06-01 10:30:00.000000+0000"``). """ redis_client.set(f'notification_last_timestamp:{mail_type}', json.dumps(value)) @@ -295,34 +296,35 @@ def default_notification( *, notification_id: str, level: str = 'ERROR', - timestamp: str | None = None, + timestamp: datetime | None = None, model_name: str = 'model-a', model_id: str = '1', ) -> dict[str, Any]: """ Build a minimal notification_queue document mirroring production layout. - The ``timestamp`` field is stored as a string in ``DATETIME_FORMAT_WITH_TZ`` - because that is exactly what ``sientia_do.notifications.models.Notification`` - writes into ``notification_queue`` in production (``now().strftime( - DATETIME_FORMAT_WITH_TZ)``). Tests intentionally use this same format so we + The ``timestamp`` field is stored as a native, timezone-aware ``datetime`` + (BSON Date) because that is exactly what + ``sientia_do.notifications.models.Notification`` writes into + ``notification_queue`` in production since SIENTIAPDE-2007 + (``self.timestamp = now()`` — a real ``datetime``, no more string + formatting). Tests intentionally use this same representation so we surface, rather than hide, real production behavior in downstream - activities. + activities (e.g. ``put_last_data_timestamp``'s ``DataFrame(...)['timestamp'] + .max().strftime(...)``, which requires a real ``datetime``, not a string). Args: notification_id: Unique identifier for the notification. level: Notification level (e.g. ``"ERROR"``, ``"WARNING"``). - timestamp: Optional production-format timestamp string. ``None`` falls - back to a fixed sample value. + timestamp: Optional timezone-aware datetime. ``None`` falls back to a + fixed sample value. model_name: Model name attached to the notification. model_id: Model id attached to the notification. Return: dict[str, Any]: A notification document ready for insertion. """ - ts = timestamp if timestamp is not None else datetime( - 2024, 6, 1, 12, 0, 0, tzinfo=UTC - ).strftime(DATETIME_FORMAT_WITH_TZ) + ts = timestamp if timestamp is not None else datetime(2024, 6, 1, 12, 0, 0, tzinfo=UTC) return { 'notification_id': notification_id, 'level': level, @@ -343,7 +345,7 @@ def default_receiver_group( members: list[str] | None = None, levels: list[str] | None = None, contents: list[str] | None = None, - ignore_models: list[str] | None = None, + ignore_models_id: list[str] | None = None, ) -> dict[str, Any]: """Minimal active receiver_groups document.""" return { @@ -352,5 +354,5 @@ def default_receiver_group( 'members': members or ['admin@example.com'], 'levels': levels or ['ERROR', 'WARNING', 'INFO'], 'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'], - 'ignore_models': ignore_models or [], + 'ignore_models_id': ignore_models_id or [], } diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 169abc2..1bf1eb5 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -71,7 +71,12 @@ Alert past `notification_ttl` with cache cleared → new email sent. ### A.2.1 Group filtering -Receiver group `levels` / `ignore_models` honored. +Receiver group `levels` / `ignore_models_id` honored. The query-time half +(union of `levels`, intersection of `ignore_models_id` across active groups) +is covered at the subworkflow level — see `LoadNotificationPackage` below. +The per-group half (each group only receiving its own precise slice) is not +yet implemented in production code (SIENTIAPDE-2012) and so is not yet +exercised here. ### A.3.1 Empty queue @@ -103,6 +108,7 @@ No notifications → no SMTP, no Postgres row. - No prior Redis timestamp → all matching notifications returned, max timestamp stored. - 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. - Empty Mongo → no Redis timestamp write. ### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py` diff --git a/e2e/test_reports_main_workflow.py b/e2e/test_reports_main_workflow.py index 64ea6f7..84d31d8 100644 --- a/e2e/test_reports_main_workflow.py +++ b/e2e/test_reports_main_workflow.py @@ -1,5 +1,6 @@ """E2E tests for the Reports main workflow.""" +from datetime import UTC, datetime from email.message import EmailMessage, Message import pytest @@ -79,13 +80,13 @@ async def test_scenario_r_1_1_happy_path( notification_id='r-warn', level='WARNING', model_name='m2', - timestamp='2024-06-01 12:01:00+0000', + timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC), ), default_notification( notification_id='r-info', level='INFO', model_name='m3', - timestamp='2024-06-01 12:02:00+0000', + timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC), ), ], ) diff --git a/e2e/test_subworkflow_load_notification_package.py b/e2e/test_subworkflow_load_notification_package.py index 2bb556f..9ba3f61 100644 --- a/e2e/test_subworkflow_load_notification_package.py +++ b/e2e/test_subworkflow_load_notification_package.py @@ -1,5 +1,7 @@ """E2E tests for LoadNotificationPackage subworkflow.""" +from datetime import UTC, datetime + import pytest from redis import Redis from temporalio.testing import WorkflowEnvironment @@ -42,10 +44,12 @@ async def test_load_package_without_prior_timestamp( """ No Redis timestamp → returns notifications and stores max timestamp. - Notifications are seeded with the exact production format produced by - ``sientia_do.notifications.models.Notification`` (string in - ``DATETIME_FORMAT_WITH_TZ``, e.g. ``"2024-06-01 11:00:00+0000"``). The - cached "last timestamp" must mirror that representation. + Notifications are seeded with the exact production representation written + by ``sientia_do.notifications.models.Notification`` since SIENTIAPDE-2007: + a native ``datetime`` (BSON Date), not a formatted string. The cached + "last timestamp" written back to Redis by ``put_last_data_timestamp`` is a + string formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` (microseconds + included) — that is the format this test asserts against. """ seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()]) seed_notifications( @@ -54,11 +58,11 @@ async def test_load_package_without_prior_timestamp( [ default_notification( notification_id='n1', - timestamp='2024-06-01 10:00:00+0000', + timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC), ), default_notification( notification_id='n2', - timestamp='2024-06-01 11:00:00+0000', + timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC), ), ], ) @@ -75,7 +79,7 @@ async def test_load_package_without_prior_timestamp( assert len(result['notification_package']) == 2 stored = redis_client.get('notification_last_timestamp:Alerts') or '' - assert stored.strip('"') == '2024-06-01 11:00:00+0000' + assert stored.strip('"') == '2024-06-01 11:00:00.000000+0000' @pytest.mark.e2e @@ -90,27 +94,27 @@ async def test_load_package_with_prior_timestamp( """ Existing timestamp → only newer notifications returned. - Both the seeded "last timestamp" (Redis) and the notification timestamps - (Mongo) follow the production format used by - ``sientia_do.notifications.models.Notification`` (string in - ``DATETIME_FORMAT_WITH_TZ``). ``load_latest_data`` will translate the - Redis value into a Python ``datetime`` and apply ``{$gt: }`` - against the Mongo string timestamps; this exercise surfaces the real - BSON comparison semantics rather than a synthetic ideal. + The seeded "last timestamp" (Redis) follows the format + ``put_last_data_timestamp`` actually writes (``DATETIME_FORMAT_MS_WITH_TZ``, + microseconds included); the notification timestamps (Mongo) are native + ``datetime`` values, matching current production behavior since + SIENTIAPDE-2007. ``load_latest_data`` parses the Redis string back into a + ``datetime`` and applies ``{$gt: }`` against the Mongo BSON dates — + this exercises the real comparison semantics rather than a synthetic ideal. """ seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()]) - seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00+0000') + seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00.000000+0000') seed_notifications( mongo_uri, E2E_DATABASE, [ default_notification( notification_id='old', - timestamp='2024-06-01 10:00:00+0000', + timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC), ), default_notification( notification_id='new', - timestamp='2024-06-01 11:00:00+0000', + timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC), ), ], ) @@ -129,6 +133,78 @@ async def test_load_package_with_prior_timestamp( assert ids == {'new'} +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_load_package_query_time_filter_union_and_intersection( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + redis_client: Redis, +): + """ + SIENTIAPDE-2011: query-time filter unions active groups' levels and + intersects their ignore_models_id, instead of fetching everything. + + Two active receiver groups narrow the fetch differently: + - group_a: levels=['ERROR'], ignore_models_id=['model-x'] + - group_b: levels=['WARNING'], ignore_models_id=['model-x', 'model-y'] + + The Mongo query must return exactly what the UNION of levels allows + (ERROR or WARNING, never INFO) and must exclude only models in the + INTERSECTION of every active group's ignore list (model-x, ignored by + both). A model ignored by only one group (model-y) must still be + fetched, since the other group still wants it — per-group correctness + for that case is SIENTIAPDE-2012's job, not this query-time filter's. + """ + seed_receiver_groups( + mongo_uri, + E2E_DATABASE, + [ + default_receiver_group( + group_name='group_a', levels=['ERROR'], ignore_models_id=['model-x'] + ), + default_receiver_group( + group_name='group_b', + levels=['WARNING'], + ignore_models_id=['model-x', 'model-y'], + ), + ], + ) + seed_notifications( + mongo_uri, + E2E_DATABASE, + [ + default_notification( + notification_id='error-included', level='ERROR', model_id='model-a' + ), + default_notification( + notification_id='warning-included', level='WARNING', model_id='model-a' + ), + default_notification(notification_id='info-excluded', level='INFO', model_id='model-a'), + default_notification( + notification_id='modelx-excluded', level='ERROR', model_id='model-x' + ), + default_notification( + notification_id='modely-included', level='WARNING', model_id='model-y' + ), + ], + ) + + input_data = load_scenario_input('subworkflow_load_notification_package.json') + _build_metadata(input_data) + + result = await start_and_await_workflow( + temporal_env.client, + LoadNotificationPackage.run, + input_data, + make_workflow_id('load-pkg-filter'), + ) + + ids = {n['notification_id'] for n in result['notification_package']} + assert ids == {'error-included', 'warning-included', 'modely-included'} + + @pytest.mark.e2e @pytest.mark.asyncio async def test_load_package_empty_mongo_no_redis_write( diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index e0f778a..38113ea 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through(): from logging import Logger from typing import Any - from pandas import DataFrame + from pandas import DataFrame, to_datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring @@ -331,14 +331,25 @@ class SlotManager(SientiaMonitoring): self.warning('No data to insert', metadata=metadata) return None - last_data_timestamp = data['timestamp'].max() + # ``timestamp`` started out as a native ``datetime``/BSON Date in Mongo, + # but by the time it gets here it has crossed a Temporal activity + # boundary (this activity's ``input_data`` is whatever + # ``load_latest_data`` returned): Temporal's default data converter + # serializes ``datetime`` values to plain ISO-8601 strings and does not + # restore them on decode, so ``data['timestamp']`` is a ``str`` in a + # real workflow run even though unit tests mocking the activity in + # isolation never see that conversion. pymongo also returns naive + # datetimes for BSON Date fields (no tzinfo), so the ISO string may + # carry no UTC offset at all. ``utc=True`` normalizes every case + # (native datetime/Timestamp, offset-aware or naive ISO string) into a + # UTC-aware ``Timestamp`` — naive values are BSON Dates, always UTC. + last_data_timestamp = to_datetime(data['timestamp'], utc=True).max() self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata) - # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from - # Mongo), which the Redis repository's plain ``json.dumps`` cannot - # serialize. Store it as a string formatted per - # ``DATETIME_FORMAT_MS_WITH_TZ`` instead. + # The Redis repository's plain ``json.dumps`` cannot serialize a + # ``Timestamp``/``datetime`` directly, so store it as a string + # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) try: