zwitschi
c17f41aaf8
- Introduced new tables for audit events and runtime state snapshots in the database schema. - Created data classes for AuditRecord and RuntimeStateRecord to represent the new entities. - Implemented AuditRepository and RuntimeStateRepository for inserting and retrieving records. - Enhanced the dashboard to include an audit trail section, displaying recent audit events. - Added tests for the new audit repository and runtime lifecycle functionalities. - Updated settings validation to ensure proper configuration for alerting features. - Integrated alert notifications across various components, including execution sequencer and loss limits.
132 lines
3.4 KiB
Python
132 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
import pytest
|
|
|
|
from arbitrade.alerting.notifier import AlertEvent, AlertNotifier
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class _FakeChannel:
|
|
events: list[AlertEvent] = field(default_factory=list)
|
|
fail: bool = False
|
|
|
|
async def send(self, event: AlertEvent) -> None:
|
|
if self.fail:
|
|
raise RuntimeError("channel send failed")
|
|
self.events.append(event)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_notifier_sends_event_when_enabled() -> None:
|
|
channel = _FakeChannel()
|
|
notifier = AlertNotifier([channel], enabled=True, min_severity="info")
|
|
|
|
sent = await notifier.notify(
|
|
category="trade",
|
|
severity="info",
|
|
title="Trade complete",
|
|
message="Completed all legs.",
|
|
)
|
|
|
|
assert sent is True
|
|
assert len(channel.events) == 1
|
|
assert channel.events[0].category == "trade"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_notifier_respects_severity_and_category_filters() -> None:
|
|
channel = _FakeChannel()
|
|
notifier = AlertNotifier(
|
|
[channel],
|
|
enabled=True,
|
|
min_severity="error",
|
|
category_flags={"trade": False, "error": True},
|
|
)
|
|
|
|
low = await notifier.notify(
|
|
category="error",
|
|
severity="warning",
|
|
title="Low",
|
|
message="Ignored by severity.",
|
|
)
|
|
filtered = await notifier.notify(
|
|
category="trade",
|
|
severity="critical",
|
|
title="Trade",
|
|
message="Ignored by category.",
|
|
)
|
|
high = await notifier.notify(
|
|
category="error",
|
|
severity="critical",
|
|
title="High",
|
|
message="Delivered.",
|
|
)
|
|
|
|
assert low is False
|
|
assert filtered is False
|
|
assert high is True
|
|
assert len(channel.events) == 1
|
|
assert channel.events[0].title == "High"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_notifier_applies_dedup_window() -> None:
|
|
channel = _FakeChannel()
|
|
notifier = AlertNotifier([channel], dedup_seconds=60.0)
|
|
|
|
first = await notifier.notify(
|
|
category="error",
|
|
severity="error",
|
|
title="Burst",
|
|
message="Same message",
|
|
)
|
|
second = await notifier.notify(
|
|
category="error",
|
|
severity="error",
|
|
title="Burst",
|
|
message="Same message",
|
|
)
|
|
|
|
assert first is True
|
|
assert second is False
|
|
assert len(channel.events) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_notifier_returns_false_when_all_channels_fail() -> None:
|
|
notifier = AlertNotifier([_FakeChannel(fail=True), _FakeChannel(fail=True)])
|
|
|
|
sent = await notifier.notify(
|
|
category="error",
|
|
severity="critical",
|
|
title="Failure",
|
|
message="Both channels fail.",
|
|
)
|
|
|
|
assert sent is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_notifier_exposes_status_snapshot_for_dashboard() -> None:
|
|
channel = _FakeChannel()
|
|
notifier = AlertNotifier([channel], enabled=True, min_severity="info", dedup_seconds=30.0)
|
|
|
|
await notifier.notify(
|
|
category="system",
|
|
severity="warning",
|
|
title="Reconnect",
|
|
message="Socket restored.",
|
|
)
|
|
|
|
status = notifier.status_snapshot()
|
|
|
|
assert status["enabled"] is True
|
|
assert status["has_channels"] is True
|
|
assert status["configured_channels"] == ["_FakeChannel"]
|
|
assert status["last_result"] == "success"
|
|
assert status["last_attempted_at"] is not None
|
|
assert status["last_success_at"] is not None
|
|
assert status["last_event"] is not None
|