141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
from contextlib import asynccontextmanager
|
|
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.domain.domain_errors import DomainError
|
|
from app.domain.schemas.task_alert import TaskAlertEvent
|
|
from app.services.task_alerts import TaskAlertService
|
|
|
|
|
|
NOW = datetime(2026, 9, 18, 2, 30, tzinfo=timezone.utc)
|
|
USER_ID = uuid4()
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
class Repository:
|
|
def __init__(self):
|
|
self.events = {}
|
|
self.alerts = {}
|
|
|
|
@asynccontextmanager
|
|
async def transaction(self):
|
|
yield self
|
|
|
|
async def lock_alert(self, _alert_id):
|
|
return None
|
|
|
|
async def get_event(self, producer_id, event_id):
|
|
return deepcopy(self.events.get((producer_id, event_id)))
|
|
|
|
async def get_alert(self, alert_id):
|
|
value = self.alerts.get(alert_id)
|
|
return (value["revision"], deepcopy(value["document"])) if value else None
|
|
|
|
async def save_event(self, event, resulting_state):
|
|
self.events[(event["producer_id"], event["event_id"])] = deepcopy(
|
|
event | {"resulting_state": resulting_state}
|
|
)
|
|
|
|
async def save_alert(self, state):
|
|
self.alerts[state["alert_id"]] = {
|
|
"revision": state["revision"],
|
|
"document": deepcopy({key: value for key, value in state.items() if key != "revision"}),
|
|
}
|
|
|
|
|
|
def command(
|
|
event_type="raised",
|
|
*,
|
|
event_id=None,
|
|
title="Low pressure",
|
|
producer_id=f"principal:{USER_ID}",
|
|
):
|
|
return TaskAlertEvent(
|
|
event_id=event_id or uuid4(),
|
|
producer_id=producer_id,
|
|
alert_id="schedule-1:2026-09-18T02:30:00Z",
|
|
event_type=event_type,
|
|
occurred_at=NOW,
|
|
condition_id=None,
|
|
source={
|
|
"task_type": "scheduled_task",
|
|
"task_id": "schedule-1",
|
|
"run_id": "occurrence-1",
|
|
},
|
|
title=title,
|
|
description="Scheduled task reported a failure.",
|
|
severity="warning",
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_alert_events_are_idempotent_and_transitions_are_monotonic():
|
|
repository = Repository()
|
|
service = TaskAlertService(
|
|
repository,
|
|
SimpleNamespace(project_role="member", user_id=USER_ID),
|
|
clock=lambda: NOW,
|
|
)
|
|
raised_command = command()
|
|
raised = await service.push(raised_command)
|
|
repeated = await service.push(raised_command)
|
|
assert raised.status == repeated.status == "active"
|
|
assert raised.revision == repeated.revision == 1
|
|
|
|
acknowledged = await service.push(command("acknowledged"))
|
|
resolved = await service.push(command("resolved"))
|
|
assert acknowledged.status == "acknowledged" and acknowledged.revision == 2
|
|
assert resolved.status == "resolved" and resolved.revision == 3
|
|
|
|
with pytest.raises(DomainError) as invalid:
|
|
await service.push(command("acknowledged"))
|
|
assert invalid.value.code == "INVALID_TRANSITION"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_alert_event_reuse_and_fact_rewrites_are_rejected():
|
|
repository = Repository()
|
|
service = TaskAlertService(
|
|
repository,
|
|
SimpleNamespace(project_role="member", user_id=USER_ID),
|
|
clock=lambda: NOW,
|
|
)
|
|
event_id = uuid4()
|
|
await service.push(command(event_id=event_id))
|
|
|
|
with pytest.raises(DomainError) as reused:
|
|
await service.push(command(event_id=event_id, title="Changed"))
|
|
assert reused.value.code == "IDEMPOTENCY_CONFLICT"
|
|
|
|
with pytest.raises(DomainError) as rewritten:
|
|
await service.push(command("resolved", title="Changed"))
|
|
assert rewritten.value.code == "IDEMPOTENCY_CONFLICT"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_internal_alert_producer_is_bound_without_user_membership():
|
|
repository = Repository()
|
|
service = TaskAlertService(
|
|
repository,
|
|
None,
|
|
clock=lambda: NOW,
|
|
internal_producer_id="scheduler:demo",
|
|
)
|
|
|
|
state = await service.push(command(producer_id="scheduler:demo"))
|
|
assert state.status == "active"
|
|
|
|
mismatched = command(producer_id="scheduler:demo")
|
|
mismatched.producer_id = "another-task"
|
|
with pytest.raises(DomainError) as denied:
|
|
await service.push(mismatched)
|
|
assert denied.value.code == "PERMISSION_DENIED"
|