Merge branch 'fix/SIENTIAPDE-2013' of https://github.com/Aignosi/sientia-dataops-orchestrator_temporal into fix/SIENTIAPDE-2013
This commit is contained in:
@@ -7,6 +7,7 @@ import json
|
|||||||
import re
|
import re
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import UTC, datetime, timedelta
|
from datetime import UTC, datetime, timedelta
|
||||||
|
from email.message import EmailMessage, Message
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
@@ -356,3 +357,40 @@ def default_receiver_group(
|
|||||||
'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'],
|
'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'],
|
||||||
'ignore_models_id': ignore_models_id or [],
|
'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 '')
|
||||||
|
|||||||
@@ -6,6 +6,11 @@ Functional reference for orchestrator E2E scenarios. Tests live under `e2e/`, us
|
|||||||
pytest e2e/ --override-ini testpaths=e2e -m e2e
|
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
|
## Execution context
|
||||||
|
|
||||||
- MongoDB, Redis, PostgreSQL: testcontainers (session-scoped).
|
- MongoDB, Redis, PostgreSQL: testcontainers (session-scoped).
|
||||||
@@ -21,31 +26,20 @@ Source: `e2e/test_orchestrator_main_workflow.py`
|
|||||||
|
|
||||||
### 1.1.1 Happy path
|
### 1.1.1 Happy path
|
||||||
|
|
||||||
Pipelines in Mongo → schedules created in correct namespace/task queue, Redis slots written, `orchestrated_schedules` updated.
|
Pipelines in Mongo → Temporal schedule created in the `laborious` namespace, `orchestrated_schedules` entries written.
|
||||||
|
|
||||||
### 1.2.1 No-op tick
|
|
||||||
|
|
||||||
Mongo, Redis, and Temporal already match desired state → no new schedules or slot writes.
|
|
||||||
|
|
||||||
### 1.3.1 Create-only
|
### 1.3.1 Create-only
|
||||||
|
|
||||||
New pipeline only → schedules created, timestamps inserted.
|
New pipeline only → Temporal schedule created for it.
|
||||||
|
|
||||||
### 1.3.2 Update-only
|
|
||||||
|
|
||||||
Existing pipeline with newer `updated_at` → schedule updated in Temporal.
|
|
||||||
|
|
||||||
### 1.3.3 Delete-only
|
### 1.3.3 Delete-only
|
||||||
|
|
||||||
Pipeline removed from Mongo → schedule deleted from Temporal.
|
Pipeline removed from Mongo → previously-created Temporal schedule deleted.
|
||||||
|
|
||||||
### 1.4.1 Conflict ordering
|
|
||||||
|
|
||||||
Pipeline update and slot delete on same OPC server → slot insert before delete (production ordering).
|
|
||||||
|
|
||||||
### 1.5.1 Empty pipelines
|
### 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
|
### 1.6.1 TTL index bootstrap
|
||||||
|
|
||||||
@@ -65,18 +59,14 @@ 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.
|
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
|
### A.2.1 Group filtering
|
||||||
|
|
||||||
Receiver group `levels` / `ignore_models_id` honored. The query-time half
|
Receiver group `levels` / `ignore_models_id` honored, both halves now covered:
|
||||||
(union of `levels`, intersection of `ignore_models_id` across active groups)
|
the query-time union/intersection across active groups (SIENTIAPDE-2011, see
|
||||||
is covered at the subworkflow level — see `LoadNotificationPackage` below.
|
`LoadNotificationPackage` below) and the per-group precise slice
|
||||||
The per-group half (each group only receiving its own precise slice) is not
|
(SIENTIAPDE-2012) — a group only receives notifications matching its own
|
||||||
yet implemented in production code (SIENTIAPDE-2012) and so is not yet
|
`levels`/`ignore_models_id`, even when the query-time filter widened the fetch
|
||||||
exercised here.
|
to satisfy another active group's broader configuration.
|
||||||
|
|
||||||
### A.3.1 Empty queue
|
### A.3.1 Empty queue
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ from e2e.helpers import (
|
|||||||
count_log_report_rows,
|
count_log_report_rows,
|
||||||
default_notification,
|
default_notification,
|
||||||
default_receiver_group,
|
default_receiver_group,
|
||||||
|
extract_html_body,
|
||||||
fetch_log_report,
|
fetch_log_report,
|
||||||
load_scenario_input,
|
load_scenario_input,
|
||||||
make_workflow_id,
|
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
|
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.e2e
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_scenario_a_3_1_empty_queue(
|
async def test_scenario_a_3_1_empty_queue(
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
"""E2E tests for the Reports main workflow."""
|
"""E2E tests for the Reports main workflow."""
|
||||||
|
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from email.message import EmailMessage, Message
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from temporalio.testing import WorkflowEnvironment
|
from temporalio.testing import WorkflowEnvironment
|
||||||
@@ -11,6 +10,7 @@ from e2e.helpers import (
|
|||||||
count_log_report_rows,
|
count_log_report_rows,
|
||||||
default_notification,
|
default_notification,
|
||||||
default_receiver_group,
|
default_receiver_group,
|
||||||
|
extract_html_body,
|
||||||
load_scenario_input,
|
load_scenario_input,
|
||||||
make_workflow_id,
|
make_workflow_id,
|
||||||
seed_notifications,
|
seed_notifications,
|
||||||
@@ -20,41 +20,6 @@ from e2e.helpers import (
|
|||||||
from orchestrator.workflows.reports import Reports
|
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.e2e
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_scenario_r_1_1_happy_path(
|
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
|
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 'Errors detected:' in body
|
||||||
assert 'Warnings detected:' in body
|
assert 'Warnings detected:' in body
|
||||||
assert 'Infos 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'
|
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 'Errors detected:' in body
|
||||||
assert 'Warnings detected:' not in body
|
assert 'Warnings detected:' not in body
|
||||||
assert 'Infos detected:' not in body
|
assert 'Infos detected:' not in body
|
||||||
|
|||||||
@@ -824,6 +824,8 @@ class Formatters(SientiaMonitoring):
|
|||||||
- group_name (str): Name of the receiver group
|
- group_name (str): Name of the receiver group
|
||||||
- contents (list): Content types to include (must contain "reports")
|
- contents (list): Content types to include (must contain "reports")
|
||||||
- ignore (list, optional): Notification IDs to exclude
|
- ignore (list, optional): Notification IDs to exclude
|
||||||
|
- levels (list, optional): Notification levels to include
|
||||||
|
- ignore_models_id (list, optional): Model IDs to exclude
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
|
dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
|
||||||
@@ -847,6 +849,8 @@ class Formatters(SientiaMonitoring):
|
|||||||
already_added_keys = []
|
already_added_keys = []
|
||||||
|
|
||||||
ignore_list = receiver_group.get('ignore', [])
|
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:
|
for notification in notification_package:
|
||||||
alert_type = 'reports'
|
alert_type = 'reports'
|
||||||
@@ -859,6 +863,11 @@ class Formatters(SientiaMonitoring):
|
|||||||
alert_type in receiver_group['contents']
|
alert_type in receiver_group['contents']
|
||||||
and notification_id not in ignore_list
|
and notification_id not in ignore_list
|
||||||
and key not in already_added_keys
|
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)
|
receiver_groups[group_name]['notifications'].append(notification)
|
||||||
already_added_keys.append(key)
|
already_added_keys.append(key)
|
||||||
|
|||||||
@@ -396,6 +396,8 @@ class SlotManager(SientiaMonitoring):
|
|||||||
- group_name (str): Name of the receiver group
|
- group_name (str): Name of the receiver group
|
||||||
- contents (list): Alert types to include (core_alerts, persistent_alerts)
|
- contents (list): Alert types to include (core_alerts, persistent_alerts)
|
||||||
- ignore (list, optional): Notification IDs to exclude
|
- 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
|
- notification_ttl (int): Seconds before considering notification persistent
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
@@ -415,48 +417,118 @@ class SlotManager(SientiaMonitoring):
|
|||||||
|
|
||||||
for receiver_group in sending_configs:
|
for receiver_group in sending_configs:
|
||||||
group_name = receiver_group['group_name']
|
group_name = receiver_group['group_name']
|
||||||
receiver_groups[group_name] = {**receiver_group, 'notifications': []}
|
notifications = self._filter_group_notifications(
|
||||||
receiver_groups[group_name]['notifications'] = []
|
receiver_group, notification_package, notification_ttl
|
||||||
|
)
|
||||||
already_added_keys = []
|
receiver_groups[group_name] = {**receiver_group, 'notifications': notifications}
|
||||||
|
|
||||||
ignore_list = receiver_group.get('ignore', [])
|
|
||||||
|
|
||||||
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 notification_id not in ignore_list
|
|
||||||
and key not in already_added_keys
|
|
||||||
):
|
|
||||||
receiver_groups[group_name]['notifications'].append(notification)
|
|
||||||
already_added_keys.append(key)
|
|
||||||
|
|
||||||
# Remove groups with no notifications
|
# Remove groups with no notifications
|
||||||
receiver_groups = {
|
return {
|
||||||
group_name: group
|
group_name: group
|
||||||
for group_name, group in receiver_groups.items()
|
for group_name, group in receiver_groups.items()
|
||||||
if group['notifications']
|
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):
|
||||||
|
"""
|
||||||
|
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
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
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.
|
||||||
|
|
||||||
|
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')
|
@activity.defn(name='store_notification_cache')
|
||||||
def store_notification_cache(self, input_data: dict[str, Any]) -> None:
|
def store_notification_cache(self, input_data: dict[str, Any]) -> None:
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "orchestrator"
|
name = "orchestrator"
|
||||||
version = "0.6.4"
|
version = "0.6.5"
|
||||||
description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
|
description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
|
|||||||
@@ -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']
|
||||||
|
|||||||
@@ -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):
|
def test_store_notification_cache(slot_manager):
|
||||||
"""Test store_notification_cache"""
|
"""Test store_notification_cache"""
|
||||||
test_data = {
|
test_data = {
|
||||||
|
|||||||
Reference in New Issue
Block a user