From 9df151d29de755350f6b583c84731af076c79a9b Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 13:46:03 -0300 Subject: [PATCH 1/8] SIENTIAPDE-2012: per-group levels/ignore_models_id filtering Extends filter_notification_alerts (slot_manager.py) and filter_notification_reports (formatters.py) to check receiver_group.get('levels') and receiver_group.get('ignore_models_id'), mirroring the existing contents/ignore pattern exactly. Closes the gap SIENTIAPDE-2011 deliberately left open: the query-time union/intersection filter widens what gets fetched for the union of all active groups, but until now nothing narrowed each group back down to its own configured slice. Both checks short-circuit via `not levels_list or ...` / `not ignore_models_id_list or ...` rather than accessing notification['level']/ ['model_id'] directly, so groups that don't configure either field (today's common case, and every existing unit test fixture) never evaluate the right-hand side and can't raise KeyError on notifications missing those keys. ignore_models_id reads via `.get(...) or []`, defensively handling an explicit `null` in a receiver_groups document from the start (same fix SIENTIAPDE-2011 needed a follow-up commit for). Tests: - 8 new unit tests (4 per file): levels restriction, ignore_models_id restriction, neither-configured passthrough, explicit-None safety. - New e2e test (test_scenario_a_2_1_per_group_levels_and_ignore_models_id) against real Mongo/Redis/Temporal: two active groups configured differently: SIENTIAPDE-2011's query-time filter widens the fetch to the union of both, and this subtask's per-group filter narrows each group back to its own levels/ignore_models_id, including the case where a model is ignored by only one of the active groups. - Deduplicated the two e2e suites' identical _extract_html_body helper into e2e/helpers.py (extract_html_body), now shared by both. Co-Authored-By: Claude Sonnet 5 --- e2e/helpers.py | 38 +++++ e2e/scenarios.md | 12 +- e2e/test_alerts_main_workflow.py | 90 ++++++++++ e2e/test_reports_main_workflow.py | 41 +---- orchestrator/activities/formatters.py | 9 + orchestrator/activities/slot_manager.py | 9 + .../activities/test_formatters.py | 140 ++++++++++++++++ .../activities/test_slot_manager.py | 156 ++++++++++++++++++ 8 files changed, 451 insertions(+), 44 deletions(-) diff --git a/e2e/helpers.py b/e2e/helpers.py index cafe1ce..ebc27d6 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -7,6 +7,7 @@ import json import re import uuid from datetime import UTC, datetime, timedelta +from email.message import EmailMessage, Message from pathlib import Path from typing import Any @@ -356,3 +357,40 @@ def default_receiver_group( 'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'], 'ignore_models_id': ignore_models_id or [], } + + +def extract_html_body(message: EmailMessage | Message) -> str: + """ + Return the text/html portion of an email message as a decoded string. + + Walks every part looking for the first text/html payload, decoding it + according to the part's transfer encoding and charset. Falls back to + the message's own ``get_content``/raw payload when no HTML part is + present so callers can still inspect plain-text reports. Shared between + the Alerts and Reports e2e suites, both of which send mail through the + same ``build_email_html``/``send_email`` activities. + + Args: + message: Captured email message returned by the test SMTP server. + + Return: + str: HTML body content, or an empty string when nothing decodable + is found. + """ + if message.is_multipart(): + for part in message.walk(): + if part.get_content_type() != 'text/html': + continue + payload = part.get_payload(decode=True) + if payload is None: + continue + charset = part.get_content_charset() or 'utf-8' + return payload.decode(charset, errors='replace') + payload = message.get_payload(decode=True) + if payload is not None: + charset = message.get_content_charset() or 'utf-8' + return payload.decode(charset, errors='replace') + try: + return message.get_content() + except (AttributeError, KeyError): + return str(message.get_payload() or '') diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 1bf1eb5..14f1b5f 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -71,12 +71,12 @@ Alert past `notification_ttl` with cache cleared → new email sent. ### A.2.1 Group filtering -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. +Receiver group `levels` / `ignore_models_id` honored, both halves now covered: +the query-time union/intersection across active groups (SIENTIAPDE-2011, see +`LoadNotificationPackage` below) and the per-group precise slice +(SIENTIAPDE-2012) — a group only receives notifications matching its own +`levels`/`ignore_models_id`, even when the query-time filter widened the fetch +to satisfy another active group's broader configuration. ### A.3.1 Empty queue diff --git a/e2e/test_alerts_main_workflow.py b/e2e/test_alerts_main_workflow.py index 638b5aa..27c3ead 100644 --- a/e2e/test_alerts_main_workflow.py +++ b/e2e/test_alerts_main_workflow.py @@ -12,6 +12,7 @@ from e2e.helpers import ( count_log_report_rows, default_notification, default_receiver_group, + extract_html_body, fetch_log_report, load_scenario_input, make_workflow_id, @@ -92,6 +93,95 @@ async def test_scenario_a_1_2_duplicate_suppressed( assert count_log_report_rows(postgres_engine, 'Alerts') == 0 +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_scenario_a_2_1_per_group_levels_and_ignore_models_id( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + smtp_server, +): + """ + A.2.1 (SIENTIAPDE-2012): each receiver group gets only its own precise + slice of notifications, on top of SIENTIAPDE-2011's query-time fetch. + + Two active groups widen the query-time union beyond Alerts' ERROR-only + default: group-errors (levels=['ERROR']) and group-warnings + (levels=['WARNING'], ignore_models_id=['model-x']). Because the query-time + filter only unions/intersects across ALL active groups, both ERROR and + WARNING notifications get fetched for every group, and model-x is not + excluded at the query level either (group-errors has no ignore_models_id + restriction, so the intersection is empty). The per-group filter added in + this subtask is what actually narrows each group back down to what it + configured. + """ + seed_receiver_groups( + mongo_uri, + E2E_DATABASE, + [ + default_receiver_group( + group_name='group-errors', contents=['core_alerts'], levels=['ERROR'] + ), + default_receiver_group( + group_name='group-warnings', + contents=['core_alerts'], + levels=['WARNING'], + ignore_models_id=['model-x'], + ), + ], + ) + seed_notifications( + mongo_uri, + E2E_DATABASE, + [ + default_notification( + notification_id='err-a', level='ERROR', model_name='model-a', model_id='model-a' + ), + default_notification( + notification_id='warn-a', + level='WARNING', + model_name='model-a', + model_id='model-a', + timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC), + ), + default_notification( + notification_id='warn-x', + level='WARNING', + model_name='model-x', + model_id='model-x', + timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC), + ), + ], + ) + + await start_and_await_workflow( + temporal_env.client, + Alerts.run, + load_scenario_input('alerts_happy_path.json'), + make_workflow_id('alerts-2012'), + ) + + # One email per group -- each group ended up with exactly one notification. + assert len(smtp_server.messages) == 2 + bodies = [extract_html_body(message) for message in smtp_server.messages] + combined = '\n'.join(bodies) + + # group-errors (levels=['ERROR']): received err-a, never warn-a/warn-x + # despite both being fetched by the widened query-time union. + errors_body = next(b for b in bodies if 'Errors detected:' in b) + assert 'Warnings detected:' not in errors_body + + # group-warnings (levels=['WARNING'], ignore_models_id=['model-x']): + # received warn-a only -- warn-x was fetched (query-time intersection is + # empty, since group-errors doesn't ignore any model) but dropped here by + # the per-group ignore_models_id check. + warnings_body = next(b for b in bodies if 'Warnings detected:' in b) + assert 'Errors detected:' not in warnings_body + assert 'model-a' in combined + assert 'model-x' not in combined + + @pytest.mark.e2e @pytest.mark.asyncio async def test_scenario_a_3_1_empty_queue( diff --git a/e2e/test_reports_main_workflow.py b/e2e/test_reports_main_workflow.py index 84d31d8..e524ef1 100644 --- a/e2e/test_reports_main_workflow.py +++ b/e2e/test_reports_main_workflow.py @@ -1,7 +1,6 @@ """E2E tests for the Reports main workflow.""" from datetime import UTC, datetime -from email.message import EmailMessage, Message import pytest from temporalio.testing import WorkflowEnvironment @@ -11,6 +10,7 @@ from e2e.helpers import ( count_log_report_rows, default_notification, default_receiver_group, + extract_html_body, load_scenario_input, make_workflow_id, seed_notifications, @@ -20,41 +20,6 @@ from e2e.helpers import ( from orchestrator.workflows.reports import Reports -def _extract_html_body(message: EmailMessage | Message) -> str: - """ - Return the text/html portion of an email message as a decoded string. - - Walks every part looking for the first text/html payload, decoding it - according to the part's transfer encoding and charset. Falls back to - the message's own ``get_content``/raw payload when no HTML part is - present so callers can still inspect plain-text reports. - - Args: - message: Captured email message returned by the test SMTP server. - - Return: - str: HTML body content, or an empty string when nothing decodable - is found. - """ - if message.is_multipart(): - for part in message.walk(): - if part.get_content_type() != 'text/html': - continue - payload = part.get_payload(decode=True) - if payload is None: - continue - charset = part.get_content_charset() or 'utf-8' - return payload.decode(charset, errors='replace') - payload = message.get_payload(decode=True) - if payload is not None: - charset = message.get_content_charset() or 'utf-8' - return payload.decode(charset, errors='replace') - try: - return message.get_content() - except (AttributeError, KeyError): - return str(message.get_payload() or '') - - @pytest.mark.e2e @pytest.mark.asyncio async def test_scenario_r_1_1_happy_path( @@ -99,7 +64,7 @@ async def test_scenario_r_1_1_happy_path( ) assert len(smtp_server.messages) == 1 - body = _extract_html_body(smtp_server.messages[0]) + body = extract_html_body(smtp_server.messages[0]) assert 'Errors detected:' in body assert 'Warnings detected:' in body assert 'Infos detected:' in body @@ -138,7 +103,7 @@ async def test_scenario_r_1_2_error_section_only( ) assert smtp_server.messages, 'Expected at least one report email' - body = _extract_html_body(smtp_server.messages[0]) + body = extract_html_body(smtp_server.messages[0]) assert 'Errors detected:' in body assert 'Warnings detected:' not in body assert 'Infos detected:' not in body diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 79a2701..66c69af 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -824,6 +824,8 @@ class Formatters(SientiaMonitoring): - group_name (str): Name of the receiver group - contents (list): Content types to include (must contain "reports") - ignore (list, optional): Notification IDs to exclude + - levels (list, optional): Notification levels to include + - ignore_models_id (list, optional): Model IDs to exclude Returns: dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name. @@ -847,6 +849,8 @@ class Formatters(SientiaMonitoring): already_added_keys = [] ignore_list = receiver_group.get('ignore', []) + levels_list = receiver_group.get('levels') + ignore_models_id_list = receiver_group.get('ignore_models_id') or [] for notification in notification_package: alert_type = 'reports' @@ -859,6 +863,11 @@ class Formatters(SientiaMonitoring): alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys + and (not levels_list or notification.get('level') in levels_list) + and ( + not ignore_models_id_list + or notification.get('model_id') not in ignore_models_id_list + ) ): receiver_groups[group_name]['notifications'].append(notification) already_added_keys.append(key) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index bbb9ea6..4de2b73 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -396,6 +396,8 @@ class SlotManager(SientiaMonitoring): - group_name (str): Name of the receiver group - contents (list): Alert types to include (core_alerts, persistent_alerts) - ignore (list, optional): Notification IDs to exclude + - levels (list, optional): Notification levels to include + - ignore_models_id (list, optional): Model IDs to exclude - notification_ttl (int): Seconds before considering notification persistent Returns: @@ -421,6 +423,8 @@ class SlotManager(SientiaMonitoring): already_added_keys = [] ignore_list = receiver_group.get('ignore', []) + levels_list = receiver_group.get('levels') + ignore_models_id_list = receiver_group.get('ignore_models_id') or [] for notification in notification_package: alert_type = 'do_nothing' @@ -445,6 +449,11 @@ class SlotManager(SientiaMonitoring): alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys + and (not levels_list or notification.get('level') in levels_list) + and ( + not ignore_models_id_list + or notification.get('model_id') not in ignore_models_id_list + ) ): receiver_groups[group_name]['notifications'].append(notification) already_added_keys.append(key) diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index c85c745..578d335 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -766,3 +766,143 @@ def test_filter_notification_reports(formatters): ], } } + + +def test_filter_notification_reports_levels_restriction(formatters): + # SIENTIAPDE-2012: a group with `levels` set only receives notifications + # matching one of those levels. + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'level': 'INFO', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['reports'], 'levels': ['ERROR']}, + ], + } + + response = formatters.filter_notification_reports(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['reports'], + 'levels': ['ERROR'], + 'notifications': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + }, + ], + }, + } + + +def test_filter_notification_reports_ignore_models_id_restriction(formatters): + # SIENTIAPDE-2012: a group with `ignore_models_id` set does not receive + # notifications from those models, but still receives everything else. + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'model_id': 'model_x', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'model_id': 'model_y', + }, + ], + 'sending_configs': [ + { + 'group_name': 'test_group_1', + 'contents': ['reports'], + 'ignore_models_id': ['model_x'], + }, + ], + } + + response = formatters.filter_notification_reports(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['reports'], + 'ignore_models_id': ['model_x'], + 'notifications': [ + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'model_id': 'model_y', + }, + ], + }, + } + + +def test_filter_notification_reports_no_levels_or_ignore_models_id_passes_everything( + formatters, +): + # SIENTIAPDE-2012: a group configuring neither field is unaffected -- same + # behavior as before this subtask (only `contents`/`ignore` apply). + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + 'model_id': 'model_x', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'level': 'INFO', + 'model_id': 'model_y', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['reports']}, + ], + } + + response = formatters.filter_notification_reports(input_data) + + assert response['test_group_1']['notifications'] == input_data['notification_package'] + + +def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_raise( + formatters, +): + # receiver_groups has no schema validation -- ignore_models_id can be + # explicitly null (key present, value None) instead of omitted. Must not + # crash (see the equivalent SIENTIAPDE-2011 fix in load_notification_package). + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'model_id': 'model_x', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['reports'], 'ignore_models_id': None}, + ], + } + + response = formatters.filter_notification_reports(input_data) + + assert response['test_group_1']['notifications'] == input_data['notification_package'] diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index 12873ca..25aa9b5 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -389,6 +389,162 @@ def test_filter_notification_alerts(slot_manager): } +def test_filter_notification_alerts_levels_restriction(slot_manager): + # SIENTIAPDE-2012: a group with `levels` set only receives notifications + # matching one of those levels. + slot_manager.redis_repository.get = MagicMock(side_effect=[None, None]) + + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'level': 'WARNING', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['core_alerts'], 'levels': ['ERROR']}, + ], + 'notification_ttl': 300, + 'mail_type': 'test_mail_type', + } + + response = slot_manager.filter_notification_alerts(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['core_alerts'], + 'levels': ['ERROR'], + 'notifications': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + }, + ], + }, + } + + +def test_filter_notification_alerts_ignore_models_id_restriction(slot_manager): + # SIENTIAPDE-2012: a group with `ignore_models_id` set does not receive + # notifications from those models, but still receives everything else. + slot_manager.redis_repository.get = MagicMock(side_effect=[None, None]) + + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'model_id': 'model_x', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'model_id': 'model_y', + }, + ], + 'sending_configs': [ + { + 'group_name': 'test_group_1', + 'contents': ['core_alerts'], + 'ignore_models_id': ['model_x'], + }, + ], + 'notification_ttl': 300, + 'mail_type': 'test_mail_type', + } + + response = slot_manager.filter_notification_alerts(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['core_alerts'], + 'ignore_models_id': ['model_x'], + 'notifications': [ + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'model_id': 'model_y', + }, + ], + }, + } + + +def test_filter_notification_alerts_no_levels_or_ignore_models_id_passes_everything( + slot_manager, +): + # SIENTIAPDE-2012: a group configuring neither field is unaffected -- same + # behavior as before this subtask (only `contents`/`ignore` apply). + slot_manager.redis_repository.get = MagicMock(side_effect=[None, None]) + + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'level': 'ERROR', + 'model_id': 'model_x', + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2', + 'level': 'INFO', + 'model_id': 'model_y', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['core_alerts']}, + ], + 'notification_ttl': 300, + 'mail_type': 'test_mail_type', + } + + response = slot_manager.filter_notification_alerts(input_data) + + assert response['test_group_1']['notifications'] == input_data['notification_package'] + + +def test_filter_notification_alerts_ignore_models_id_explicit_none_does_not_raise( + slot_manager, +): + # receiver_groups has no schema validation -- ignore_models_id can be + # explicitly null (key present, value None) instead of omitted. Must not + # crash (see the equivalent SIENTIAPDE-2011 fix in load_notification_package). + slot_manager.redis_repository.get = MagicMock(side_effect=[None]) + + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1', + 'model_id': 'model_x', + }, + ], + 'sending_configs': [ + {'group_name': 'test_group_1', 'contents': ['core_alerts'], 'ignore_models_id': None}, + ], + 'notification_ttl': 300, + 'mail_type': 'test_mail_type', + } + + response = slot_manager.filter_notification_alerts(input_data) + + assert response['test_group_1']['notifications'] == input_data['notification_package'] + + def test_store_notification_cache(slot_manager): """Test store_notification_cache""" test_data = { From 86962c2cfc983c6374a4e7141f396d2b7c856699 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 14:35:22 -0300 Subject: [PATCH 2/8] Refactorizing function to reduce its Cognitive Complexity --- orchestrator/activities/slot_manager.py | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 4de2b73..dcbd536 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -449,11 +449,8 @@ class SlotManager(SientiaMonitoring): alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys - and (not levels_list or notification.get('level') in levels_list) - and ( - not ignore_models_id_list - or notification.get('model_id') not in ignore_models_id_list - ) + and self._check_levels_and_ignore_models_id( + notification, levels_list, ignore_models_id_list) ): receiver_groups[group_name]['notifications'].append(notification) already_added_keys.append(key) @@ -467,6 +464,23 @@ class SlotManager(SientiaMonitoring): return receiver_groups + @staticmethod + def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list): + """ + Check if the notification's level is in the levels_list and its model_id is not in the ignore_models_id_list. + + Args: + notification (dict): The notification to check. + levels_list (list, optional): List of levels to include. If None, all levels are included. + ignore_models_id_list (list, optional): List of model_ids to exclude. If None, no models are excluded. + + Returns: + bool: True if the notification passes the checks, False otherwise. + """ + level_check = not levels_list or notification.get('level') in levels_list + model_id_check = not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list + return level_check and model_id_check + @activity.defn(name='store_notification_cache') def store_notification_cache(self, input_data: dict[str, Any]) -> None: """ From 6a95b8d2df56654a9bbf753587ecee2ebe47d162 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 14:39:59 -0300 Subject: [PATCH 3/8] fix: apply ruff formatting to _check_levels_and_ignore_models_id Fixes CI formatting failure introduced by the manual Sonar cognitive- complexity fix (extracting the levels/ignore_models_id check into its own helper). Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index dcbd536..08b6ba8 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -450,7 +450,8 @@ class SlotManager(SientiaMonitoring): and notification_id not in ignore_list and key not in already_added_keys and self._check_levels_and_ignore_models_id( - notification, levels_list, ignore_models_id_list) + notification, levels_list, ignore_models_id_list + ) ): receiver_groups[group_name]['notifications'].append(notification) already_added_keys.append(key) @@ -474,11 +475,13 @@ class SlotManager(SientiaMonitoring): levels_list (list, optional): List of levels to include. If None, all levels are included. ignore_models_id_list (list, optional): List of model_ids to exclude. If None, no models are excluded. - Returns: - bool: True if the notification passes the checks, False otherwise. + Returns: + bool: True if the notification passes the checks, False otherwise. """ level_check = not levels_list or notification.get('level') in levels_list - model_id_check = not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list + model_id_check = ( + not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list + ) return level_check and model_id_check @activity.defn(name='store_notification_cache') From a59cc1c691f9f3b867e80f5ba94829489e74e0ed Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 14:48:45 -0300 Subject: [PATCH 4/8] fix: apply ruff formatting to _check_keys_and_notif_id Fixes CI formatting failure introduced by the manual Sonar cognitive- complexity fix (extracting the ignore-list/dedup check into its own helper). Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 08b6ba8..683813c 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -447,8 +447,9 @@ class SlotManager(SientiaMonitoring): # Check if this group must be notified if ( alert_type in receiver_group['contents'] - and notification_id not in ignore_list - and key not in already_added_keys + and self._check_keys_and_notif_id( + notification_id, ignore_list, key, already_added_keys + ) and self._check_levels_and_ignore_models_id( notification, levels_list, ignore_models_id_list ) @@ -484,6 +485,23 @@ class SlotManager(SientiaMonitoring): ) return level_check and model_id_check + @staticmethod + def _check_keys_and_notif_id(notification_id, ignore_list, key, already_added_keys) -> bool: + """ + Check if the key has already been added to the list of already added keys. + + Args: + notification_id (str): The ID of the notification. + ignore_list (list): List of notification IDs to ignore. + key (str): The key to check. + already_added_keys (list[str]): List of keys that have already been added. + + Returns: + bool: True if the key has not been added yet, False otherwise. + + """ + return notification_id not in ignore_list and key not in already_added_keys + @activity.defn(name='store_notification_cache') def store_notification_cache(self, input_data: dict[str, Any]) -> None: """ From 081298a883b914a6a5c00450a733e2b2b7acb24a Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 14:51:51 -0300 Subject: [PATCH 5/8] fix: annotate already_added_keys to satisfy mypy The parameter-level annotation on _check_keys_and_notif_id alone wasn't enough -- mypy needs the type at the empty list's declaration site inside filter_notification_alerts. Also reformatted the helper's now-longer signature to satisfy ruff. Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 683813c..395f094 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -420,7 +420,7 @@ class SlotManager(SientiaMonitoring): receiver_groups[group_name] = {**receiver_group, 'notifications': []} receiver_groups[group_name]['notifications'] = [] - already_added_keys = [] + already_added_keys: list[str] = [] ignore_list = receiver_group.get('ignore', []) levels_list = receiver_group.get('levels') @@ -486,7 +486,9 @@ class SlotManager(SientiaMonitoring): return level_check and model_id_check @staticmethod - def _check_keys_and_notif_id(notification_id, ignore_list, key, already_added_keys) -> bool: + def _check_keys_and_notif_id( + notification_id, ignore_list, key, already_added_keys: list[str] + ) -> bool: """ Check if the key has already been added to the list of already added keys. From b7325938f5dcc2dbacd76947503025acbe819a49 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 15:05:59 -0300 Subject: [PATCH 6/8] refactor: extract per-group filtering to reduce cognitive complexity filter_notification_alerts tripped Sonar's cognitive-complexity gate (16 > 15 allowed) from its nested loop/if structure. Split into _filter_group_notifications (per-group notification selection) and _determine_alert_type (Redis TTL classification), leaving the public method as a simple loop over receiver groups. Behavior and call order are unchanged -- verified against the existing filter_notification_alerts unit tests by inspection (sientia_do isn't installable in this local environment, so the real suite can't run here; CI will confirm). Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 108 +++++++++++++++--------- 1 file changed, 67 insertions(+), 41 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 395f094..7182551 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -417,54 +417,80 @@ class SlotManager(SientiaMonitoring): for receiver_group in sending_configs: group_name = receiver_group['group_name'] - receiver_groups[group_name] = {**receiver_group, 'notifications': []} - receiver_groups[group_name]['notifications'] = [] - - already_added_keys: list[str] = [] - - ignore_list = receiver_group.get('ignore', []) - levels_list = receiver_group.get('levels') - ignore_models_id_list = receiver_group.get('ignore_models_id') or [] - - for notification in notification_package: - alert_type = 'do_nothing' - notification_id = notification['notification_id'] - # Check if notification was recently sent - key = f'{notification["trigger"]}:{notification_id}' - - last_sent = self.redis_repository.get(key) - - if last_sent is None: - alert_type = 'core_alerts' - - else: - last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ) - - # Check if "notification_ttl" seconds has passed since last sent - if (now() - last_sent) > timedelta(seconds=notification_ttl): - alert_type = 'persistent_alerts' - - # Check if this group must be notified - if ( - alert_type in receiver_group['contents'] - and self._check_keys_and_notif_id( - notification_id, ignore_list, key, already_added_keys - ) - and self._check_levels_and_ignore_models_id( - notification, levels_list, ignore_models_id_list - ) - ): - receiver_groups[group_name]['notifications'].append(notification) - already_added_keys.append(key) + notifications = self._filter_group_notifications( + receiver_group, notification_package, notification_ttl + ) + receiver_groups[group_name] = {**receiver_group, 'notifications': notifications} # Remove groups with no notifications - receiver_groups = { + return { group_name: group for group_name, group in receiver_groups.items() if group['notifications'] } - return receiver_groups + def _filter_group_notifications( + self, + receiver_group: dict[str, Any], + notification_package: list[dict[str, Any]], + notification_ttl: int, + ) -> list[dict[str, Any]]: + """ + Select the notifications this receiver group should get. + + Args: + receiver_group (dict[str, Any]): The receiver group configuration. + notification_package (list[dict[str, Any]]): Candidate notifications. + notification_ttl (int): Seconds before considering a notification persistent. + + Returns: + list[dict[str, Any]]: Notifications matching this group's filters. + """ + already_added_keys: list[str] = [] + ignore_list = receiver_group.get('ignore', []) + levels_list = receiver_group.get('levels') + ignore_models_id_list = receiver_group.get('ignore_models_id') or [] + + notifications = [] + for notification in notification_package: + notification_id = notification['notification_id'] + key = f'{notification["trigger"]}:{notification_id}' + alert_type = self._determine_alert_type(key, notification_ttl) + + if ( + alert_type in receiver_group['contents'] + and self._check_keys_and_notif_id( + notification_id, ignore_list, key, already_added_keys + ) + and self._check_levels_and_ignore_models_id( + notification, levels_list, ignore_models_id_list + ) + ): + notifications.append(notification) + already_added_keys.append(key) + + return notifications + + def _determine_alert_type(self, key: str, notification_ttl: int) -> str: + """ + Classify a notification as new, persistent, or not-yet-due. + + Args: + key (str): The Redis cache key for this notification. + notification_ttl (int): Seconds before considering a notification persistent. + + Returns: + str: 'core_alerts', 'persistent_alerts', or 'do_nothing'. + """ + last_sent = self.redis_repository.get(key) + if last_sent is None: + return 'core_alerts' + + last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ) + if (now() - last_sent) > timedelta(seconds=notification_ttl): + return 'persistent_alerts' + + return 'do_nothing' @staticmethod def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list): From eaf0ca786371e8a2469b913a044e459d5c23329b Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Thu, 17 Sep 2026 09:52:38 -0300 Subject: [PATCH 7/8] docs: make e2e/scenarios.md a strict 1:1 mirror of actual e2e tests Removed scenarios documented with no backing test (1.2.1, 1.3.2, 1.4.1, A.1.3) and corrected three descriptions that claimed assertions the tests don't actually make (1.1.1's "Redis slots written", 1.3.1's "timestamps inserted", 1.5.1's "orphan schedules removed") to match what's really verified. A.2.1 stays as-is here -- it's already accurate for this branch's real per-group filtering test. Co-Authored-By: Claude Sonnet 5 --- e2e/scenarios.md | 30 ++++++++++-------------------- 1 file changed, 10 insertions(+), 20 deletions(-) diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 14f1b5f..5277223 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -6,6 +6,11 @@ Functional reference for orchestrator E2E scenarios. Tests live under `e2e/`, us pytest e2e/ --override-ini testpaths=e2e -m e2e ``` +**Every section below maps 1:1 to an actual test function** (named in each section, or as a bullet +under a `Source:` heading) — this file describes what the suite verifies today, not planned or +aspirational coverage. Adding a scenario here without a matching test (or vice versa) is a doc bug; +fix the mismatch instead of leaving it for later. + ## Execution context - MongoDB, Redis, PostgreSQL: testcontainers (session-scoped). @@ -21,31 +26,20 @@ Source: `e2e/test_orchestrator_main_workflow.py` ### 1.1.1 Happy path -Pipelines in Mongo → schedules created in correct namespace/task queue, Redis slots written, `orchestrated_schedules` updated. - -### 1.2.1 No-op tick - -Mongo, Redis, and Temporal already match desired state → no new schedules or slot writes. +Pipelines in Mongo → Temporal schedule created in the `laborious` namespace, `orchestrated_schedules` entries written. ### 1.3.1 Create-only -New pipeline only → schedules created, timestamps inserted. - -### 1.3.2 Update-only - -Existing pipeline with newer `updated_at` → schedule updated in Temporal. +New pipeline only → Temporal schedule created for it. ### 1.3.3 Delete-only -Pipeline removed from Mongo → schedule deleted from Temporal. - -### 1.4.1 Conflict ordering - -Pipeline update and slot delete on same OPC server → slot insert before delete (production ordering). +Pipeline removed from Mongo → previously-created Temporal schedule deleted. ### 1.5.1 Empty pipelines -No active pipelines → orphan schedules removed, no new orchestration writes. +No active pipelines → no `orchestrated_schedules` documents written, even with a pre-existing +unrelated Temporal schedule present. ### 1.6.1 TTL index bootstrap @@ -65,10 +59,6 @@ ERROR notification → one SMTP message, one `log_report` row, Redis cache key. Second run with same data and cache seeded → no extra email or log row. -### A.1.3 Persistent escalation - -Alert past `notification_ttl` with cache cleared → new email sent. - ### A.2.1 Group filtering Receiver group `levels` / `ignore_models_id` honored, both halves now covered: From 5cc8015fc7aa86318358c34e3f87e525f07e77a9 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 17 Sep 2026 14:36:26 +0000 Subject: [PATCH 8/8] chore: bump version to 0.6.5 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 658479e..5313360 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "orchestrator" -version = "0.6.4" +version = "0.6.5" description = "Sientia DataOps Orchestrator - ML Model Orchestration System" readme = "README.md" requires-python = ">=3.11"