gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

SIENTIAPDE-2012: per-group levels/ignore_models_id filtering

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-15 13:46:03 -03:00
parent d250383f78
commit 9df151d29d
8 changed files with 451 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
e2e/helpers.py
View File

@@ -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 '')

12
e2e/scenarios.md
View File

@@ -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

90
e2e/test_alerts_main_workflow.py
View File

@@ -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(

41
e2e/test_reports_main_workflow.py
View File

@@ -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

9
orchestrator/activities/formatters.py
View File

@@ -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)

9
orchestrator/activities/slot_manager.py
View File

@@ -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)

140
tests/orchestrator/activities/test_formatters.py
View File

@@ -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']

156
tests/orchestrator/activities/test_slot_manager.py
View File

@@ -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 = {