feat: improve bgp incident visibility
This commit is contained in:
@@ -563,3 +563,83 @@ async def test_bgp_incidents_api_returns_incident():
|
||||
assert list_response.json()["total"] == 1
|
||||
assert detail_response.status_code == 200
|
||||
assert detail_response.json()["incident_type"] == "origin_change"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bgp_incident_summary_api_returns_aggregates():
|
||||
class _SummaryResult:
|
||||
def __init__(self, scalar_value=None, rows=None):
|
||||
self._scalar_value = scalar_value
|
||||
self._rows = rows or []
|
||||
|
||||
def scalar(self):
|
||||
return self._scalar_value
|
||||
|
||||
def fetchall(self):
|
||||
return self._rows
|
||||
|
||||
class _SummarySession:
|
||||
def __init__(self):
|
||||
self.calls = 0
|
||||
|
||||
async def execute(self, _stmt):
|
||||
self.calls += 1
|
||||
if self.calls == 1:
|
||||
return _SummaryResult(scalar_value=2)
|
||||
if self.calls == 2:
|
||||
return _SummaryResult(rows=[("origin_change", 2)])
|
||||
if self.calls == 3:
|
||||
return _SummaryResult(rows=[("critical", 1), ("high", 1)])
|
||||
return _SummaryResult(rows=[("active", 2)])
|
||||
|
||||
db = _SummarySession()
|
||||
client = await _bgp_test_client(db)
|
||||
|
||||
try:
|
||||
response = await client.get("/api/v1/bgp/incidents/summary")
|
||||
finally:
|
||||
await client.aclose()
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert payload["total"] == 2
|
||||
assert payload["by_type"]["origin_change"] == 2
|
||||
assert payload["by_severity"]["critical"] == 1
|
||||
assert payload["by_status"]["active"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bgp_event_summary_api_returns_aggregates():
|
||||
observation_one = BGPObservation(
|
||||
id=1,
|
||||
source="ris_live_bgp",
|
||||
collector="rrc00",
|
||||
prefix="203.0.113.0/24",
|
||||
event_type="announcement",
|
||||
observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC),
|
||||
)
|
||||
observation_two = BGPObservation(
|
||||
id=2,
|
||||
source="ris_live_bgp",
|
||||
collector="rrc01",
|
||||
prefix="198.51.100.0/24",
|
||||
event_type="withdrawal",
|
||||
observed_at=datetime(2026, 3, 30, 10, 5, tzinfo=UTC),
|
||||
)
|
||||
db = _FakeAsyncSession([[observation_one, observation_two]])
|
||||
client = await _bgp_test_client(db)
|
||||
|
||||
try:
|
||||
response = await client.get("/api/v1/bgp/events/summary")
|
||||
finally:
|
||||
await client.aclose()
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert payload["total"] == 2
|
||||
assert payload["collector_count"] == 2
|
||||
assert payload["prefix_count"] == 2
|
||||
assert payload["by_type"]["announcement"] == 1
|
||||
assert payload["by_type"]["withdrawal"] == 1
|
||||
|
||||
Reference in New Issue
Block a user