diff --git a/CLAUDE.md b/CLAUDE.md index 66e7092..5ab46c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,7 +45,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable ### Layering - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`). -- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` vs all levels). +- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections. - **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing. 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/formatters.py b/orchestrator/activities/formatters.py index 67efdf5..79a2701 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -7,11 +7,15 @@ with workflow.unsafe.imports_passed_through(): from math import ceil from typing import Any, TypedDict, cast - 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 - from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now + from sientia_do.temporal.constants import ( + DATETIME_FORMAT_MS_WITH_TZ, + DATETIME_FORMAT_WITH_TZ, + now, + ) from orchestrator.utils.orchestrator_functions import ( build_tag_config, @@ -755,9 +759,24 @@ class Formatters(SientiaMonitoring): key = f'{notification_id}:{trigger}' if key not in data: + # ``notification['timestamp']`` started out as a native + # ``datetime``/BSON Date in Mongo, but by the time it gets here + # it has crossed at least one Temporal activity boundary: + # Temporal's default data converter serializes ``datetime`` + # values to plain ISO-8601 strings and does not restore them on + # decode, and pymongo returns naive datetimes for BSON Date + # fields (no tzinfo), so the ISO string may carry no UTC offset + # at all. Normalizing with ``utc=True`` and re-formatting to + # ``DATETIME_FORMAT_WITH_TZ`` (same pattern as + # ``SlotManager.put_last_data_timestamp``) keeps the value a + # fixed-format string that ``export_data_to_postgres``'s + # ``timestamp_conversion`` can parse after the next boundary + # crossing, regardless of what shape it arrived in here. data[key] = { 'status': group_config['status'], - 'timestamp': notification['timestamp'], + 'timestamp': to_datetime(notification['timestamp'], utc=True).strftime( + DATETIME_FORMAT_WITH_TZ + ), 'groups': [group_name], 'message': notification['message'], 'level': notification['level'], diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index e0f778a..bbb9ea6 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,18 +331,33 @@ 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. - last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) + # 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. Kept as a + # separate variable (rather than reassigning ``last_data_timestamp``) + # since it's a different type — mypy infers a local's type from its + # first assignment and flags a same-name reassignment to an + # incompatible type as an error. + last_data_timestamp_str = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) try: - self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5) + self.redis_repository.set(key, last_data_timestamp_str, ttl=60 * 60 * 5) except Exception as e: self.send_notification( metadata=metadata, @@ -354,7 +369,7 @@ class SlotManager(SientiaMonitoring): ) raise e - return last_data_timestamp + return last_data_timestamp_str @activity.defn(name='filter_notification_alerts') def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: diff --git a/orchestrator/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 8b603f2..614899c 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -68,18 +68,23 @@ class LoadNotificationPackage: # Load notification package from collection "notification_queue", using a # static filter + sending_configs = await sending_configs_handler + + base_data_filter = self._compute_query_time_filter( + sending_configs, input_data['base_data_filter'] + ) + notification_package = await workflow.start_local_activity_method( Activities.load_latest_data, { **metadata, 'collection_name': 'notification_queue', 'last_data_timestamp': last_timestamp, - 'base_data_filter': input_data['base_data_filter'], + 'base_data_filter': base_data_filter, }, start_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy, ) - sending_configs = await sending_configs_handler if not sending_configs or not notification_package: return { @@ -106,3 +111,56 @@ class LoadNotificationPackage: 'notification_package': notification_package, 'sending_configs': sending_configs, } + + def _compute_query_time_filter( + self, sending_configs: list[dict[str, Any]], default_filter: dict[str, Any] + ) -> dict[str, Any]: + """ + Compute the Mongo query-time filter from all active receiver groups. + + - levels: union of every group's `levels`. If ANY active group has no + restriction (missing/empty `levels`), that group wants every level, so + the query can't restrict levels at all -> falls back to whatever level + restriction `default_filter` itself carries (e.g. ERROR-only for Alerts). + + When every active group DOES restrict `levels`, the union REPLACES + `default_filter`'s own level restriction rather than ANDing with it — + this is deliberate, not an oversight: SIENTIAPDE-2011's design doc and + ticket text both call for Alerts' hard-coded `level: ERROR` to become + "whatever levels the active groups actually care about", falling back + to ERROR-only only when no group narrows further. Until SIENTIAPDE-2012 + (per-group `levels` filtering) lands, this means a single active group + configuring `levels` to include WARNING/INFO widens the Alerts fetch for + every active group, not just that one — an accepted interim gap, not a + bug in this function. + - ignore_models_id: intersection of every group's `ignore_models_id`. A model + is only safe to drop from the fetch if EVERY active group ignores it. + A group with no `ignore_models_id` is treated as "ignores nothing" (empty + set), which naturally zeroes out the intersection via set math. + """ + if not sending_configs: + return default_filter + + sending_configs_levels = [group.get('levels') for group in sending_configs] + + if all(sending_configs_levels): + levels_union: set[str] = set() + for levels in sending_configs_levels: + if levels: + levels_union.update(levels) + query_filter = {**default_filter, 'level': {'$in': sorted(levels_union)}} + else: + query_filter = dict(default_filter) + + # `or []`, not a `.get(..., [])` default: a document with `ignore_models_id` + # explicitly set to `null` (no schema validation on this collection) makes + # `.get()` return None, and `set(None)` would raise TypeError. + sending_configs_ignore_models_id = [ + set(group.get('ignore_models_id') or []) for group in sending_configs + ] + models_intersection = set.intersection(*sending_configs_ignore_models_id) + + if models_intersection: + query_filter['model_id'] = {'$nin': sorted(models_intersection)} + + return query_filter diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 70b74ab..c85c745 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -716,7 +716,7 @@ def test_format_log_report(formatters): [ { 'status': 'sent', - 'timestamp': '2021-01-01', + 'timestamp': '2021-01-01 00:00:00+0000', 'groups': ['test_receiver_group', 'test_receiver_group2'], 'message': 'test_message', 'level': 'test_level', diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py index faeac7f..3266f70 100644 --- a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -32,16 +32,19 @@ async def test_run(workflow_mock, load_notification_package): 'base_data_filter': {'level': 'ERROR'}, } + # Await order is now: last_timestamp -> sending_configs -> notification_package + # (fixed ordering bug: sending_configs must resolve before load_latest_data is + # called, so its levels/ignore_models_id can feed the query-time filter). workflow_mock.start_local_activity_method.side_effect = [ '2023-01-01 12:00:00', [ { - 'id': '1', + 'id_r': '1', } ], [ { - 'id_r': '1', + 'id': '1', } ], ] @@ -158,7 +161,9 @@ async def test_run_has_data(workflow_mock, load_notification_package): 'mail_type': 'test_mail_type', } - workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', ['data'], []] + # sending_configs (awaited 2nd) is empty here on purpose, to exercise the + # early-return branch below regardless of notification_package having data. + workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', [], ['data']] output = await load_notification_package.run(input_data) @@ -169,3 +174,103 @@ async def test_run_has_data(workflow_mock, load_notification_package): } workflow_mock.start_activity_method.assert_not_called() + + +def test_compute_query_time_filter_empty_active_groups(load_notification_package): + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter([], default_filter) + + assert result == default_filter + + +def test_compute_query_time_filter_mixed_groups_no_levels_restriction(load_notification_package): + # One active group restricts levels/ignore_models_id, the other has neither set + # (the common case today) -> that group wants everything, so no levels + # restriction can be applied at query time; falls back to default_filter. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + {'group_name': 'group_b'}, + ] + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': 'ERROR'} + + +def test_compute_query_time_filter_union_replaces_default_level_restriction( + load_notification_package, +): + # default_filter already restricts level (e.g. Alerts' ERROR-only default), + # but every active group specifies levels of its own -> the union REPLACES + # default_filter's level restriction, it does not AND with it. This is + # deliberate per SIENTIAPDE-2011's design doc and ticket text ("Alerts' + # hard-coded level: ERROR should become whatever levels the active groups + # actually care about"), not a bug -- this test locks the behavior in so a + # future "fix" reintroducing AND semantics fails loudly instead of passing + # silently. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['WARNING'], 'ignore_models_id': []}, + ] + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['WARNING']}} + + +def test_compute_query_time_filter_model_ignored_by_only_one_group_not_excluded( + load_notification_package, +): + # model_a is ignored by group_a but group_b still wants it -> must stay in + # the fetch (the per-group filter is what drops it for group_a specifically). + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': []}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['ERROR']}} + assert 'model_id' not in result + + +def test_compute_query_time_filter_union_levels_and_intersection_models( + load_notification_package, +): + # Every active group restricts levels -> union of levels applies. + # model_a is ignored by every active group -> safe to exclude at query time. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a', 'model_b']}, + {'group_name': 'group_b', 'levels': ['WARNING', 'ERROR'], 'ignore_models_id': ['model_a']}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == { + 'level': {'$in': ['ERROR', 'WARNING']}, + 'model_id': {'$nin': ['model_a']}, + } + + +def test_compute_query_time_filter_ignore_models_id_explicit_none_does_not_raise( + load_notification_package, +): + # receiver_groups has no schema validation, so a document can have + # ignore_models_id explicitly set to null (key present, value None) rather + # than omitted. `group.get('ignore_models_id', [])` would still return None + # in that case (the default only applies when the key is missing), and + # `set(None)` raises TypeError -- must not crash the whole activity. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': None}, + {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['ERROR']}} + assert 'model_id' not in result