Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 23 additions & 13 deletions backend/tests/e2e/db/repositories/test_dlq_repository.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import uuid
from collections.abc import AsyncIterator
from datetime import datetime, timezone

import pytest
Expand All @@ -12,8 +13,9 @@
pytestmark = pytest.mark.e2e


async def insert_test_dlq_docs() -> list[str]:
"""Insert test DLQ documents using Beanie and return their event ids."""
@pytest.fixture
async def dlq_docs() -> AsyncIterator[list[str]]:
"""Three DLQ rows on topics t1 and t2; they go when the test ends, so the DLQ size job stops counting them."""
now = datetime.now(timezone.utc)
event_ids = [f"dlq-{uuid.uuid4().hex}" for _ in range(3)]

Expand Down Expand Up @@ -67,28 +69,36 @@ async def insert_test_dlq_docs() -> list[str]:

for doc in docs:
await doc.insert()
return event_ids
yield event_ids
await DLQMessageDocument.find({"event.event_id": {"$in": event_ids}}).delete()


async def test_list_get_and_updates(dlq_repository: DLQRepository) -> None:
await insert_test_dlq_docs()
@pytest.fixture
async def logged_in_event() -> AsyncIterator[UserLoggedInEvent]:
"""The event the test dead-letters for two groups; its rows go when the test ends."""
event = UserLoggedInEvent(
user_id="u1",
login_method=LoginMethod.PASSWORD,
metadata=EventMetadata(service_name="svc", service_version="1", user_id="u1"),
)
yield event
await DLQMessageDocument.find({"event.event_id": event.event_id}).delete()


async def test_list_get_and_updates(dlq_repository: DLQRepository, dlq_docs: list[str]) -> None:
res = await dlq_repository.get_messages(limit=2)
assert res.total >= 3 and len(res.messages) <= 2
topics = await dlq_repository.get_topics_summary()
assert any(t.topic == "t1" for t in topics)


async def test_each_consumer_group_keeps_its_own_row(dlq_repository: DLQRepository) -> None:
event = UserLoggedInEvent(
user_id="u1",
login_method=LoginMethod.PASSWORD,
metadata=EventMetadata(service_name="svc", service_version="1", user_id="u1"),
)
async def test_each_consumer_group_keeps_its_own_row(
dlq_repository: DLQRepository, logged_in_event: UserLoggedInEvent
) -> None:
for group, error in [("group-a", "a failed"), ("group-b", "b failed"), ("group-a", "a failed again")]:
await dlq_repository.save_message(
DLQMessage(event=event, original_topic="t1", consumer_group=group, error=error)
DLQMessage(event=logged_in_event, original_topic="t1", consumer_group=group, error=error)
)

docs = await DLQMessageDocument.find({"event.event_id": event.event_id}).sort("consumer_group").to_list()
docs = await DLQMessageDocument.find({"event.event_id": logged_in_event.event_id}).sort("consumer_group").to_list()
assert [(d.consumer_group, d.error) for d in docs] == [("group-a", "a failed again"), ("group-b", "b failed")]
Loading