23 lines
609 B
Python
23 lines
609 B
Python
|
import pytest
|
||
|
|
|
||
|
|
from onguard24.domain import Alert, AlertReceived, InMemoryEventBus, Severity
|
||
|
|
|
||
|
|
|
||
|
|
def test_alert_model() -> None:
|
||
|
|
a = Alert(source="grafana", severity=Severity.CRITICAL, title="x")
|
||
|
|
assert a.source == "grafana"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_event_bus_alert_received() -> None:
|
||
|
|
seen: list[str] = []
|
||
|
|
|
||
|
|
async def h(ev: AlertReceived) -> None:
|
||
|
|
seen.append(ev.name)
|
||
|
|
|
||
|
|
bus = InMemoryEventBus()
|
||
|
|
bus.subscribe("alert.received", h) # type: ignore[arg-type]
|
||
|
|
a = Alert(source="grafana")
|
||
|
|
await bus.publish_alert_received(a)
|
||
|
|
assert seen == ["alert.received"]
|