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

SIENTIAPDE-2011: add e2e coverage for query-time filter, fix stale e2e fixtures

Browse Source 
e2e fixtures (default_notification, seed_last_timestamp, default_receiver_group)
still assumed the pre-SIENTIAPDE-2007 string-timestamp representation and the
old ignore_models field name, so the load_notification_package e2e suite was
failing independently of this ticket's changes:

- default_notification now seeds `timestamp` as a native datetime (matches
  production since SIENTIAPDE-2007's Notification.timestamp = now()).
- seed_last_timestamp now uses DATETIME_FORMAT_MS_WITH_TZ (matches what
  put_last_data_timestamp actually writes to Redis today).
- default_receiver_group's ignore_models renamed to ignore_models_id, per
  this ticket's decision.
- test_reports_main_workflow.py's explicit timestamp= call sites updated to
  match.

Also fixes a real production bug in put_last_data_timestamp (slot_manager.py)
surfaced by running the e2e suite against a real Temporal environment: Temporal's
default data converter serializes datetime values to ISO-8601 strings across
the activity boundary, and pymongo returns naive datetimes for BSON Date
fields, so `data['timestamp'].max().strftime(...)` crashed on a plain str.
Unit tests never caught this because they mock the activity boundary and
never cross real (de)serialization. Fixed by normalizing through
`pandas.to_datetime(..., utc=True)`, which handles native datetime/Timestamp
and offset-aware/naive ISO strings uniformly.

Adds test_load_package_query_time_filter_union_and_intersection, a new e2e
test exercising the actual SIENTIAPDE-2011 behavior end-to-end: union of
active groups' levels, intersection of their ignore_models_id, and the
requirement that a model ignored by only one of several active groups must
not be excluded from the fetch.

Note: test_reports_main_workflow.py / test_alerts_main_workflow.py still have
a separate, pre-existing failure in process_notifications.py (same family of
bug: a notification's timestamp field parsed with a fixed strptime format
after crossing the same Temporal serialization boundary) — confirmed present
on the unmodified baseline too, unrelated to this ticket, tracked separately.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-14 12:15:23 -03:00
parent 1711f3b0d1
commit 14b854148a
5 changed files with 143 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
e2e/helpers.py
View File

@@ -29,7 +29,6 @@ MONGO_COLLECTIONS = (
)
DATETIME_FORMAT_MS_WITH_TZ = '%Y-%m-%d %H:%M:%S.%f%z'
DATETIME_FORMAT_WITH_TZ = '%Y-%m-%d %H:%M:%S%z'
_TIMESTAMP_MARKER_PATTERN = re.compile(r'^@now(?:([+-])(\d+)([smhd]))?$')
_TIMESTAMP_UNIT_TO_KWARG = {'s': 'seconds', 'm': 'minutes', 'h': 'hours', 'd': 'days'}
@@ -222,17 +221,19 @@ def seed_last_timestamp(redis_client: Redis, mail_type: str, value: str) -> None
Set notification_last_timestamp for a mail type, JSON-encoded.
Values must be JSON-encoded so ``redis_repository.get`` (which calls
``json.loads`` on the raw payload) can deserialize them. The value
must follow the exact format ``sientia_do.notifications.models.Notification``
writes into ``notification_queue.timestamp``: ``DATETIME_FORMAT_WITH_TZ``
(e.g. ``"2026-05-22 16:47:02+0000"``) — no microseconds and no colon in
the timezone offset.
``json.loads`` on the raw payload) can deserialize them. The value must
follow the exact format ``put_last_data_timestamp``
(``orchestrator/activities/slot_manager.py``) writes into Redis today:
``DATETIME_FORMAT_MS_WITH_TZ`` (e.g. ``"2026-05-22 16:47:02.000000+0000"``)
— microseconds included, no colon in the timezone offset. ``load_latest_data``
parses this same format back into a ``datetime`` for the ``$gt`` comparison,
so the two sides must agree exactly.
Args:
redis_client: Redis client connected to the test instance.
mail_type: Mail type identifier (e.g. ``"Alerts"``, ``"Reports"``).
value: Timestamp string in ``DATETIME_FORMAT_WITH_TZ``
(e.g. ``"2024-06-01 10:30:00+0000"``).
value: Timestamp string in ``DATETIME_FORMAT_MS_WITH_TZ``
(e.g. ``"2024-06-01 10:30:00.000000+0000"``).
"""
redis_client.set(f'notification_last_timestamp:{mail_type}', json.dumps(value))
@@ -295,34 +296,35 @@ def default_notification(
*,
notification_id: str,
level: str = 'ERROR',
timestamp: str | None = None,
timestamp: datetime | None = None,
model_name: str = 'model-a',
model_id: str = '1',
) -> dict[str, Any]:
"""
Build a minimal notification_queue document mirroring production layout.
The ``timestamp`` field is stored as a string in ``DATETIME_FORMAT_WITH_TZ``
because that is exactly what ``sientia_do.notifications.models.Notification``
writes into ``notification_queue`` in production (``now().strftime(
DATETIME_FORMAT_WITH_TZ)``). Tests intentionally use this same format so we
The ``timestamp`` field is stored as a native, timezone-aware ``datetime``
(BSON Date) because that is exactly what
``sientia_do.notifications.models.Notification`` writes into
``notification_queue`` in production since SIENTIAPDE-2007
(``self.timestamp = now()`` — a real ``datetime``, no more string
formatting). Tests intentionally use this same representation so we
surface, rather than hide, real production behavior in downstream
activities.
activities (e.g. ``put_last_data_timestamp``'s ``DataFrame(...)['timestamp']
.max().strftime(...)``, which requires a real ``datetime``, not a string).
Args:
notification_id: Unique identifier for the notification.
level: Notification level (e.g. ``"ERROR"``, ``"WARNING"``).
timestamp: Optional production-format timestamp string. ``None`` falls
back to a fixed sample value.
timestamp: Optional timezone-aware datetime. ``None`` falls back to a
fixed sample value.
model_name: Model name attached to the notification.
model_id: Model id attached to the notification.
Return:
dict[str, Any]: A notification document ready for insertion.
"""
ts = timestamp if timestamp is not None else datetime(
2024, 6, 1, 12, 0, 0, tzinfo=UTC
).strftime(DATETIME_FORMAT_WITH_TZ)
ts = timestamp if timestamp is not None else datetime(2024, 6, 1, 12, 0, 0, tzinfo=UTC)
return {
'notification_id': notification_id,
'level': level,
@@ -343,7 +345,7 @@ def default_receiver_group(
members: list[str] | None = None,
levels: list[str] | None = None,
contents: list[str] | None = None,
ignore_models: list[str] | None = None,
ignore_models_id: list[str] | None = None,
) -> dict[str, Any]:
"""Minimal active receiver_groups document."""
return {
@@ -352,5 +354,5 @@ def default_receiver_group(
'members': members or ['admin@example.com'],
'levels': levels or ['ERROR', 'WARNING', 'INFO'],
'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'],
'ignore_models': ignore_models or [],
'ignore_models_id': ignore_models_id or [],
}

8
e2e/scenarios.md
View File

@@ -71,7 +71,12 @@ Alert past `notification_ttl` with cache cleared → new email sent.
### A.2.1 Group filtering
Receiver group `levels` / `ignore_models` honored.
Receiver group `levels` / `ignore_models_id` honored. The query-time half
(union of `levels`, intersection of `ignore_models_id` across active groups)
is covered at the subworkflow level — see `LoadNotificationPackage` below.
The per-group half (each group only receiving its own precise slice) is not
yet implemented in production code (SIENTIAPDE-2012) and so is not yet
exercised here.
### A.3.1 Empty queue
@@ -103,6 +108,7 @@ No notifications → no SMTP, no Postgres row.
- No prior Redis timestamp → all matching notifications returned, max timestamp stored.
- Prior timestamp → only newer notifications returned.
- Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch.
- Empty Mongo → no Redis timestamp write.
### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py`

5
e2e/test_reports_main_workflow.py
View File

@@ -1,5 +1,6 @@
"""E2E tests for the Reports main workflow."""
from datetime import UTC, datetime
from email.message import EmailMessage, Message
import pytest
@@ -79,13 +80,13 @@ async def test_scenario_r_1_1_happy_path(
notification_id='r-warn',
level='WARNING',
model_name='m2',
timestamp='2024-06-01 12:01:00+0000',
timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC),
),
default_notification(
notification_id='r-info',
level='INFO',
model_name='m3',
timestamp='2024-06-01 12:02:00+0000',
timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC),
),
],
)

110
e2e/test_subworkflow_load_notification_package.py
View File

@@ -1,5 +1,7 @@
"""E2E tests for LoadNotificationPackage subworkflow."""
from datetime import UTC, datetime
import pytest
from redis import Redis
from temporalio.testing import WorkflowEnvironment
@@ -42,10 +44,12 @@ async def test_load_package_without_prior_timestamp(
"""
No Redis timestamp → returns notifications and stores max timestamp.
Notifications are seeded with the exact production format produced by
``sientia_do.notifications.models.Notification`` (string in
``DATETIME_FORMAT_WITH_TZ``, e.g. ``"2024-06-01 11:00:00+0000"``). The
cached "last timestamp" must mirror that representation.
Notifications are seeded with the exact production representation written
by ``sientia_do.notifications.models.Notification`` since SIENTIAPDE-2007:
a native ``datetime`` (BSON Date), not a formatted string. The cached
"last timestamp" written back to Redis by ``put_last_data_timestamp`` is a
string formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` (microseconds
included) — that is the format this test asserts against.
"""
seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
seed_notifications(
@@ -54,11 +58,11 @@ async def test_load_package_without_prior_timestamp(
[
default_notification(
notification_id='n1',
timestamp='2024-06-01 10:00:00+0000',
timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC),
),
default_notification(
notification_id='n2',
timestamp='2024-06-01 11:00:00+0000',
timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC),
),
],
)
@@ -75,7 +79,7 @@ async def test_load_package_without_prior_timestamp(
assert len(result['notification_package']) == 2
stored = redis_client.get('notification_last_timestamp:Alerts') or ''
assert stored.strip('"') == '2024-06-01 11:00:00+0000'
assert stored.strip('"') == '2024-06-01 11:00:00.000000+0000'
@pytest.mark.e2e
@@ -90,27 +94,27 @@ async def test_load_package_with_prior_timestamp(
"""
Existing timestamp → only newer notifications returned.
Both the seeded "last timestamp" (Redis) and the notification timestamps
(Mongo) follow the production format used by
``sientia_do.notifications.models.Notification`` (string in
``DATETIME_FORMAT_WITH_TZ``). ``load_latest_data`` will translate the
Redis value into a Python ``datetime`` and apply ``{$gt: <Date>}``
against the Mongo string timestamps; this exercise surfaces the real
BSON comparison semantics rather than a synthetic ideal.
The seeded "last timestamp" (Redis) follows the format
``put_last_data_timestamp`` actually writes (``DATETIME_FORMAT_MS_WITH_TZ``,
microseconds included); the notification timestamps (Mongo) are native
``datetime`` values, matching current production behavior since
SIENTIAPDE-2007. ``load_latest_data`` parses the Redis string back into a
``datetime`` and applies ``{$gt: <Date>}`` against the Mongo BSON dates —
this exercises the real comparison semantics rather than a synthetic ideal.
"""
seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00+0000')
seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00.000000+0000')
seed_notifications(
mongo_uri,
E2E_DATABASE,
[
default_notification(
notification_id='old',
timestamp='2024-06-01 10:00:00+0000',
timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC),
),
default_notification(
notification_id='new',
timestamp='2024-06-01 11:00:00+0000',
timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC),
),
],
)
@@ -129,6 +133,78 @@ async def test_load_package_with_prior_timestamp(
assert ids == {'new'}
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_load_package_query_time_filter_union_and_intersection(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
redis_client: Redis,
):
"""
SIENTIAPDE-2011: query-time filter unions active groups' levels and
intersects their ignore_models_id, instead of fetching everything.
Two active receiver groups narrow the fetch differently:
- group_a: levels=['ERROR'], ignore_models_id=['model-x']
- group_b: levels=['WARNING'], ignore_models_id=['model-x', 'model-y']
The Mongo query must return exactly what the UNION of levels allows
(ERROR or WARNING, never INFO) and must exclude only models in the
INTERSECTION of every active group's ignore list (model-x, ignored by
both). A model ignored by only one group (model-y) must still be
fetched, since the other group still wants it — per-group correctness
for that case is SIENTIAPDE-2012's job, not this query-time filter's.
"""
seed_receiver_groups(
mongo_uri,
E2E_DATABASE,
[
default_receiver_group(
group_name='group_a', levels=['ERROR'], ignore_models_id=['model-x']
),
default_receiver_group(
group_name='group_b',
levels=['WARNING'],
ignore_models_id=['model-x', 'model-y'],
),
],
)
seed_notifications(
mongo_uri,
E2E_DATABASE,
[
default_notification(
notification_id='error-included', level='ERROR', model_id='model-a'
),
default_notification(
notification_id='warning-included', level='WARNING', model_id='model-a'
),
default_notification(notification_id='info-excluded', level='INFO', model_id='model-a'),
default_notification(
notification_id='modelx-excluded', level='ERROR', model_id='model-x'
),
default_notification(
notification_id='modely-included', level='WARNING', model_id='model-y'
),
],
)
input_data = load_scenario_input('subworkflow_load_notification_package.json')
_build_metadata(input_data)
result = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-filter'),
)
ids = {n['notification_id'] for n in result['notification_package']}
assert ids == {'error-included', 'warning-included', 'modely-included'}
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_load_package_empty_mongo_no_redis_write(

23
orchestrator/activities/slot_manager.py
View File

@@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through():
from logging import Logger
from typing import Any
from pandas import DataFrame
from pandas import DataFrame, to_datetime
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
@@ -331,14 +331,25 @@ class SlotManager(SientiaMonitoring):
self.warning('No data to insert', metadata=metadata)
return None
last_data_timestamp = data['timestamp'].max()
# ``timestamp`` started out as a native ``datetime``/BSON Date in Mongo,
# but by the time it gets here it has crossed a Temporal activity
# boundary (this activity's ``input_data`` is whatever
# ``load_latest_data`` returned): Temporal's default data converter
# serializes ``datetime`` values to plain ISO-8601 strings and does not
# restore them on decode, so ``data['timestamp']`` is a ``str`` in a
# real workflow run even though unit tests mocking the activity in
# isolation never see that conversion. pymongo also returns naive
# datetimes for BSON Date fields (no tzinfo), so the ISO string may
# carry no UTC offset at all. ``utc=True`` normalizes every case
# (native datetime/Timestamp, offset-aware or naive ISO string) into a
# UTC-aware ``Timestamp`` — naive values are BSON Dates, always UTC.
last_data_timestamp = to_datetime(data['timestamp'], utc=True).max()
self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
# ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from
# Mongo), which the Redis repository's plain ``json.dumps`` cannot
# serialize. Store it as a string formatted per
# ``DATETIME_FORMAT_MS_WITH_TZ`` instead.
# The Redis repository's plain ``json.dumps`` cannot serialize a
# ``Timestamp``/``datetime`` directly, so store it as a string
# formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead.
last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ)
try: