"""Tests for BGP observability helpers.""" from datetime import UTC, datetime, timedelta import pytest from httpx import ASGITransport, AsyncClient from unittest.mock import AsyncMock, patch from app.api.v1.bgp import BGP_SOURCES from app.core.security import get_current_user from app.db.session import get_db from app.main import app from app.services.bgp_detectors import ( detect_mass_withdrawal_anomalies, detect_origin_change_anomalies, detect_path_flap_anomalies, detect_route_leak_anomalies, ) from app.services.collectors.bgp_common import ( create_bgp_anomalies_for_batch, save_bgp_observations_for_batch, ) from app.services.bgp_enrichment import enrich_bgp_events_for_batch, extract_bgp_network_fields from app.services.bgp_incidents import ( create_bgp_incidents_for_anomalies, infer_related_infrastructure, ) from app.services.bgp_collectors import build_bgp_collector_coverage from app.api.v1.visualization import convert_bgp_incidents_to_geojson, build_incident_geography_hints from app.models.bgp_anomaly import BGPAnomaly from app.models.collected_data import CollectedData from app.models.bgp_incident import BGPIncident from app.models.bgp_observation import BGPObservation from app.models.user import User from app.services.collectors.bgp_common import normalize_bgp_event from app.services.collectors.bgpstream import BGPStreamBackfillCollector from app.services.collectors.iptoasn import IPtoASNPrefixGeoCollector from app.services.collectors.opengeofeed import OpenGeoFeedPrefixGeoCollector from app.services.collectors.nro_delegated import NRODelegatedPrefixGeoCollector class _FakeScalarResult: def __init__(self, rows): self._rows = rows def all(self): return self._rows class _FakeResult: def __init__(self, rows): self._rows = rows def scalars(self): return _FakeScalarResult(self._rows) def fetchall(self): return self._rows def fetchone(self): return self._rows[0] if self._rows else None class _FakeAsyncSession: def __init__(self, results, gets=None): self._results = list(results) self._gets = gets or {} self.added = [] self.commits = 0 async def execute(self, _stmt, _params=None): if not self._results: return _FakeResult([]) return _FakeResult(self._results.pop(0)) async def get(self, model, item_id): return self._gets.get((model, item_id)) def add(self, item): self.added.append(item) async def commit(self): self.commits += 1 def test_normalize_bgp_event_from_live_payload(): event = normalize_bgp_event( { "collector": "rrc00", "peer_asn": "3333", "peer_ip": "2001:db8::1", "type": "UPDATE", "event_type": "announcement", "prefix": "203.0.113.0/24", "path": ["3333", "64500", "64496"], "communities": ["3333:100"], "timestamp": "2026-03-26T08:00:00Z", }, project="ris-live", ) assert event["name"] == "203.0.113.0/24" assert event["metadata"]["collector"] == "rrc00" assert event["metadata"]["peer_asn"] == 3333 assert event["metadata"]["origin_asn"] == 64496 assert event["metadata"]["as_path_length"] == 3 assert event["metadata"]["prefix_length"] == 24 assert event["metadata"]["is_more_specific"] is False def test_normalize_bgp_event_uses_peer_and_community_fallbacks(): event = normalize_bgp_event( { "collector": "rrc00", "peer_asn": "3333", "peer": "2405:a640::50", "type": "UPDATE", "prefix": "2401:2260::/32", "path": [3333, 15412, 9304, 151650], "community": [[15412, 603], [3333, 100]], "timestamp": "2026-03-27T06:07:18.470000+00:00", }, project="ris-live", ) assert event["metadata"]["peer_ip"] == "2405:a640::50" assert event["metadata"]["communities"] == [[15412, 603], [3333, 100]] def test_bgpstream_transform_preserves_broker_record(): collector = BGPStreamBackfillCollector() transformed = collector.transform( [ { "project": "routeviews", "collector": "route-views.sg", "filename": "rib.20260326.0800.gz", "startTime": "2026-03-26T08:00:00Z", "prefix": "198.51.100.0/24", "origin_asn": 64512, } ] ) assert len(transformed) == 1 record = transformed[0] assert record["name"] == "rib.20260326.0800.gz" assert record["metadata"]["project"] == "bgpstream" assert record["metadata"]["broker_record"]["filename"] == "rib.20260326.0800.gz" def test_iptoasn_transform_creates_prefix_geography_records(): collector = IPtoASNPrefixGeoCollector() transformed = collector.transform( [ { "range_start": "1.0.0.0", "range_end": "1.0.0.255", "asn": "13335", "country_code": "AU", "as_name": "CLOUDFLARENET", } ] ) assert len(transformed) == 1 record = transformed[0] assert record["name"] == "1.0.0.0/24" assert record["metadata"]["family"] == "ipv4" assert record["metadata"]["country_code"] == "AU" assert record["metadata"]["asn"] == 13335 assert record["metadata"]["source_dataset"] == "iptoasn_combined" def test_iptoasn_build_dataset_urls_from_combined(): urls = IPtoASNPrefixGeoCollector._build_dataset_urls( "https://iptoasn.com/data/ip2asn-combined.tsv.gz" ) assert urls == [ "https://iptoasn.com/data/ip2asn-v4.tsv.gz", "https://iptoasn.com/data/ip2asn-v6.tsv.gz", ] def test_iptoasn_build_dataset_urls_passthrough_non_combined(): url = "https://example.com/custom.tsv.gz" urls = IPtoASNPrefixGeoCollector._build_dataset_urls(url) assert urls == [url] def test_opengeofeed_transform_creates_prefix_geography_records(): collector = OpenGeoFeedPrefixGeoCollector() transformed = collector.transform( [ { "prefix": "203.0.113.0/24", "country_code": "GB", "region": "GB-LND", "city": "London", "postal_code": "EC1A", "extra_columns": ["source:example"], } ] ) assert len(transformed) == 1 record = transformed[0] assert record["name"] == "203.0.113.0/24" assert record["metadata"]["family"] == "ipv4" assert record["metadata"]["range_start"] == "203.0.113.0" assert record["metadata"]["range_end"] == "203.0.113.255" assert record["metadata"]["country_code"] == "GB" assert record["metadata"]["source_dataset"] == "opengeofeed_public" assert record["metadata"]["confidence"] == "geofeed" def test_nro_delegated_transform_creates_prefix_geography_records(): collector = NRODelegatedPrefixGeoCollector() transformed = collector.transform( [ { "rir": "ripencc", "country_code": "DE", "type": "ipv4", "start": "198.51.100.0", "value": "256", "allocated_date": "20250401", "status": "allocated", } ] ) assert len(transformed) == 1 record = transformed[0] assert record["name"] == "198.51.100.0/24" assert record["metadata"]["family"] == "ipv4" assert record["metadata"]["range_start"] == "198.51.100.0" assert record["metadata"]["range_end"] == "198.51.100.255" assert record["metadata"]["country_code"] == "DE" assert record["metadata"]["source_dataset"] == "nro_delegated_stats" assert record["metadata"]["confidence"] == "registry_allocated" def test_bgp_anomaly_to_dict(): anomaly = BGPAnomaly( source="ris_live_bgp", anomaly_type="origin_change", severity="critical", status="active", entity_key="origin_change:203.0.113.0/24:64497", prefix="203.0.113.0/24", origin_asn=64496, new_origin_asn=64497, summary="Origin ASN changed", confidence=0.9, evidence={"previous_origins": [64496], "current_origins": [64497]}, ) data = anomaly.to_dict() assert data["source"] == "ris_live_bgp" assert data["anomaly_type"] == "origin_change" assert data["new_origin_asn"] == 64497 assert data["evidence"]["previous_origins"] == [64496] def test_bgp_observation_to_dict(): observation = BGPObservation( source="ris_live_bgp", ingest_batch_id="ris_live_bgp:1:1", source_event_id="evt-1", collector="rrc00", peer_asn=3333, peer_ip="2001:db8::1", prefix="203.0.113.0/24", event_type="announcement", as_path=[3333, 64500, 64496], origin_asn=64496, next_hop="2001:db8::2", communities=["3333:100"], collector_geo={"city": "Amsterdam", "country": "Netherlands"}, raw_payload={"raw": "deadbeef"}, ) data = observation.to_dict() assert data["source"] == "ris_live_bgp" assert data["collector"] == "rrc00" assert data["event_type"] == "announcement" assert data["as_path"] == [3333, 64500, 64496] assert data["collector_geo"]["city"] == "Amsterdam" def test_extract_bgp_network_fields(): ipv4 = extract_bgp_network_fields("203.0.113.0/24") assert ipv4["prefix_family"] == "ipv4" assert ipv4["prefix_length"] == 24 assert ipv4["prefix_supernet"] == "203.0.0.0/16" assert ipv4["is_more_specific"] is False ipv6 = extract_bgp_network_fields("2001:db8:1::/48") assert ipv6["prefix_family"] == "ipv6" assert ipv6["prefix_length"] == 48 assert ipv6["prefix_supernet"] == "2001:db8::/32" assert ipv6["is_more_specific"] is False def test_detect_mass_withdrawal_anomalies(): events = [ { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64496, "event_type": "withdrawal", } } for _ in range(3) ] anomalies = detect_mass_withdrawal_anomalies( source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert len(anomalies) == 1 assert anomalies[0].anomaly_type == "mass_withdrawal" assert anomalies[0].prefix == "203.0.113.0/24" def test_detect_origin_change_anomalies_creates_conflict_without_baseline(): events = [ { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64496, "collector": "rrc00", "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, } }, { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64497, "collector": "rrc01", "collector_location": { "country": "United Kingdom", "city": "London", "latitude": 51.5072, "longitude": -0.1276, }, } }, ] anomalies = detect_origin_change_anomalies( source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, previous_origin_map={}, ) assert len(anomalies) == 2 assert {item.anomaly_type for item in anomalies} == {"origin_conflict"} assert anomalies[0].peer_scope == ["rrc00", "rrc01"] def test_detect_mass_withdrawal_anomalies_accepts_cross_collector_pair(): events = [ { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64496, "event_type": "withdrawal", "collector": "rrc00", "peer_asn": 3333, "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, } }, { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64496, "event_type": "withdrawal", "collector": "rrc01", "peer_asn": 3334, "collector_location": { "country": "United Kingdom", "city": "London", "latitude": 51.5072, "longitude": -0.1276, }, } }, ] anomalies = detect_mass_withdrawal_anomalies( source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert len(anomalies) == 1 assert anomalies[0].severity == "medium" assert anomalies[0].evidence["collector_count"] == 2 def test_detect_route_leak_anomalies_creates_candidate_for_divergent_long_paths(): events = [ { "metadata": { "collector": "rrc00", "event_type": "announcement", "prefix": "203.0.113.0/24", "origin_asn": 64496, "as_path": [64500, 64496], "collector_location": {"country": "NL", "city": "Amsterdam", "latitude": 52.3, "longitude": 4.9}, "enrichment": {"prefix_scope": {"regions": [{"country": "NL"}]}}, } }, { "metadata": { "collector": "rrc01", "event_type": "announcement", "prefix": "203.0.113.0/24", "origin_asn": 64496, "as_path": [64510, 64520, 64530, 64540, 64496], "collector_location": {"country": "GB", "city": "London", "latitude": 51.5, "longitude": -0.1}, "enrichment": {"prefix_scope": {"regions": [{"country": "GB"}]}}, } }, ] anomalies = detect_route_leak_anomalies( source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert len(anomalies) == 1 assert anomalies[0].anomaly_type == "route_leak_candidate" assert anomalies[0].evidence["max_path_length"] == 5 def test_detect_path_flap_anomalies_creates_signal_for_repeated_state_changes(): base_timestamp = datetime(2026, 3, 27, 0, 0, tzinfo=UTC) events = [] for index, event_type in enumerate(["announcement", "withdrawal", "announcement", "withdrawal"]): events.append( { "metadata": { "collector": "rrc00", "event_type": event_type, "timestamp": (base_timestamp + timedelta(minutes=index)).isoformat(), "prefix": "198.51.100.0/24", "origin_asn": 64512, "as_path": [64500 + index, 64512] if event_type == "announcement" else [], "collector_location": {"country": "NL", "city": "Amsterdam", "latitude": 52.3, "longitude": 4.9}, "enrichment": {"prefix_scope": {"regions": [{"country": "NL"}]}}, } } ) anomalies = detect_path_flap_anomalies( source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert len(anomalies) == 1 assert anomalies[0].anomaly_type == "path_flap" assert anomalies[0].evidence["transitions"] == 3 def test_bgp_incident_to_dict(): incident = BGPIncident( source="ris_live_bgp", incident_key="origin_change:203.0.113.0/24:64497", incident_type="origin_change", title="Origin Change incident on 203.0.113.0/24", summary="Grouped incident summary", severity="critical", status="active", confidence=0.91, affected_prefixes=["203.0.113.0/24"], affected_asns=[64496, 64497], affected_collectors=["rrc00", "rrc01"], affected_regions=[{"country": "Netherlands", "city": "Amsterdam"}], evidence_refs=["origin_change:203.0.113.0/24:64497"], ) data = incident.to_dict() assert data["incident_type"] == "origin_change" assert data["affected_prefixes"] == ["203.0.113.0/24"] assert data["affected_collectors"] == ["rrc00", "rrc01"] def test_convert_bgp_incidents_to_geojson_adds_estimated_geography(): incident = BGPIncident( source="ris_live_bgp", incident_key="origin_change:203.0.113.0/24:64497", incident_type="origin_change", title="Origin Change incident on 203.0.113.0/24", summary="Grouped incident summary", severity="critical", status="active", confidence=0.91, affected_prefixes=["203.0.113.0/24"], affected_collectors=["rrc00", "rrc01"], affected_regions=[ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, { "collector": "rrc01", "country": "United Kingdom", "city": "London", "latitude": 51.5072, "longitude": -0.1276, }, ], ) payload = convert_bgp_incidents_to_geojson([incident]) feature = payload["features"][0] assert feature["properties"]["geography_mode"] == "collector_centroid" assert feature["properties"]["estimated_radius_km"] > 0 assert feature["properties"]["estimated_center"]["latitude"] != 0 def test_convert_bgp_incidents_to_geojson_prefers_prefix_scope_hint(): incident = BGPIncident( source="ris_live_bgp", incident_key="origin_change:203.0.113.0/24:64497", incident_type="origin_change", title="Origin Change incident on 203.0.113.0/24", summary="Grouped incident summary", severity="critical", status="active", confidence=0.91, affected_prefixes=["203.0.113.0/24"], affected_collectors=["rrc00"], affected_regions=[ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, } ], ) payload = convert_bgp_incidents_to_geojson( [incident], { incident.incident_key: { "geography_mode": "prefix_scope", "regions": [ { "country": "Japan", "city": "Tokyo", "latitude": 35.6764, "longitude": 139.65, } ], } }, ) feature = payload["features"][0] assert feature["properties"]["geography_mode"] == "prefix_scope" assert feature["geometry"]["coordinates"] == [139.65, 35.6764] def test_convert_bgp_incidents_to_geojson_prefers_prefix_geography_hint(): incident = BGPIncident( source="ris_live_bgp", incident_key="origin_change:198.51.100.0/24:64512", incident_type="origin_change", title="Origin Change incident on 198.51.100.0/24", summary="Grouped incident summary", severity="critical", status="active", confidence=0.91, affected_prefixes=["198.51.100.0/24"], affected_collectors=["rrc00"], affected_regions=[ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, } ], ) payload = convert_bgp_incidents_to_geojson( [incident], { incident.incident_key: { "geography_mode": "prefix_geography", "regions": [ { "country": "日本", "city": None, "latitude": 35.6764, "longitude": 139.65, } ], } }, ) feature = payload["features"][0] assert feature["properties"]["geography_mode"] == "prefix_geography" assert feature["geometry"]["coordinates"] == [139.65, 35.6764] @pytest.mark.asyncio async def test_enrich_bgp_events_for_batch_adds_profiles_and_prefix_scope(): historical_observation = BGPObservation( source="ris_live_bgp", collector="rrc01", prefix="203.0.113.0/24", origin_asn=64496, observed_at=datetime(2026, 3, 28, 0, 0, tzinfo=UTC), collector_geo={ "country": "United Kingdom", "city": "London", "latitude": 51.5072, "longitude": -0.1276, }, event_type="announcement", ) peeringdb_record = CollectedData( source="peeringdb_network", name="ExampleNet", extra_data={ "asn": 64497, "country": "NL", "city": "Amsterdam", "info_type": "Content", "ix_count": 3, "url": "https://example.net", }, ) peeringdb_record.id = 99 iptoasn_row = { "extra_data": { "family": "ipv4", "range_start": "203.0.113.0", "range_end": "203.0.113.255", "prefix": "203.0.113.0/24", "country": "英国", "country_code": "GB", "asn": 64497, "as_name": "Example ASN", "source_dataset": "iptoasn_combined", } } db = _FakeAsyncSession([[historical_observation], [], [iptoasn_row], [peeringdb_record]]) events = [ { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64497, "new_origin_asn": None, "collector": "rrc00", "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, "as_path": [3333, 64497, 64497], "timestamp": "2026-03-30T10:00:00Z", }, "reference_date": "2026-03-30T10:00:00Z", } ] enriched = await enrich_bgp_events_for_batch(db, source="ris_live_bgp", events=events) enrichment = enriched[0]["metadata"]["enrichment"] assert enrichment["path_prepending"] is True assert enrichment["is_new_origin_for_prefix"] is True assert enrichment["rpki_validation"]["status"] == "unknown" assert enrichment["origin_asn_profile"]["name"] == "ExampleNet" assert enrichment["prefix_geography"]["country"] == "英国" assert enrichment["prefix_geography"]["source"] == "iptoasn_combined" assert enrichment["prefix_scope"]["countries"] == ["United Kingdom"] assert enrichment["prefix_scope"]["cities"] == ["London"] @pytest.mark.asyncio async def test_enrich_bgp_events_for_batch_prefers_opengeofeed_over_iptoasn(): opengeofeed_row = { "extra_data": { "family": "ipv4", "range_start": "203.0.113.0", "range_end": "203.0.113.255", "prefix": "203.0.113.0/24", "country_code": "GB", "region": "GB-LND", "city": "London", "source_dataset": "opengeofeed_public", "confidence": "geofeed", } } db = _FakeAsyncSession([[], [opengeofeed_row], []]) events = [ { "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64497, "collector": "rrc00", "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, "as_path": [3333, 64497], "timestamp": "2026-03-30T10:00:00Z", }, "reference_date": "2026-03-30T10:00:00Z", } ] enriched = await enrich_bgp_events_for_batch(db, source="ris_live_bgp", events=events) geography = enriched[0]["metadata"]["enrichment"]["prefix_geography"] assert geography["source"] == "opengeofeed_public" assert geography["confidence"] == "geofeed" assert geography["city"] == "London" @pytest.mark.asyncio async def test_enrich_bgp_events_for_batch_falls_back_to_nro_delegated(): nro_row = { "extra_data": { "family": "ipv4", "range_start": "198.51.100.0", "range_end": "198.51.100.255", "prefix": "198.51.100.0/24", "country_code": "DE", "rir": "ripencc", "source_dataset": "nro_delegated_stats", "confidence": "registry_allocated", } } db = _FakeAsyncSession([[], [], [nro_row], []]) events = [ { "metadata": { "prefix": "198.51.100.0/24", "origin_asn": 64512, "collector": "rrc00", "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, "as_path": [3333, 64512], "timestamp": "2026-03-30T10:00:00Z", }, "reference_date": "2026-03-30T10:00:00Z", } ] enriched = await enrich_bgp_events_for_batch(db, source="ris_live_bgp", events=events) geography = enriched[0]["metadata"]["enrichment"]["prefix_geography"] assert geography["source"] == "nro_delegated_stats" assert geography["confidence"] == "registry_allocated" assert geography["country"] == "德国" @pytest.mark.asyncio async def test_create_bgp_incidents_for_anomalies_aggregates_regions_and_collectors(): db = _FakeAsyncSession([[]]) anomaly = BGPAnomaly( source="ris_live_bgp", anomaly_type="origin_change", severity="critical", status="active", entity_key="origin_change:203.0.113.0/24:64497", prefix="203.0.113.0/24", origin_asn=64496, new_origin_asn=64497, summary="Origin ASN changed", confidence=0.9, evidence={ "impacted_regions": [ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, } ] }, ) with patch( "app.services.bgp_incidents.infer_related_infrastructure", new=AsyncMock(return_value={"related_cables": [], "related_ixps": []}), ): created = await create_bgp_incidents_for_anomalies( db, source="ris_live_bgp", snapshot_id=1, task_id=2, anomalies=[anomaly], ) assert created == 1 assert db.commits == 1 assert len(db.added) == 1 incident = db.added[0] assert incident.incident_type == "origin_change" assert incident.affected_collectors == ["rrc00"] assert incident.affected_regions[0]["city"] == "Amsterdam" @pytest.mark.asyncio async def test_create_bgp_incidents_for_anomalies_refreshes_existing_incident(): existing = BGPIncident( source="ris_live_bgp", incident_key="origin_change:203.0.113.0/24:64497", incident_type="origin_change", title="Old title", summary="Old summary", severity="medium", status="active", confidence=0.4, affected_prefixes=["203.0.113.0/24"], affected_asns=[64496, 64497], affected_collectors=["rrc00"], affected_regions=[ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, } ], related_cables=[], related_ixps=[], evidence_refs=["old-key"], ) db = _FakeAsyncSession([[existing]]) anomaly = BGPAnomaly( source="ris_live_bgp", anomaly_type="origin_change", severity="critical", status="active", entity_key="origin_change:203.0.113.0/24:64497", prefix="203.0.113.0/24", origin_asn=64496, new_origin_asn=64497, summary="Origin ASN changed", confidence=0.9, evidence={ "impacted_regions": [ { "collector": None, "country": "United States", "city": None, "latitude": 39.8283, "longitude": -98.5795, } ] }, ) with patch( "app.services.bgp_incidents.infer_related_infrastructure", new=AsyncMock(return_value={"related_cables": [{"landing_point": "NYC"}], "related_ixps": []}), ): created = await create_bgp_incidents_for_anomalies( db, source="ris_live_bgp", snapshot_id=1, task_id=2, anomalies=[anomaly], ) assert created == 0 assert db.commits == 1 assert len(db.added) == 0 assert existing.summary != "Old summary" assert existing.severity == "critical" assert existing.confidence == 0.9 assert existing.affected_regions[0]["country"] == "United States" assert existing.related_cables == [{"landing_point": "NYC"}] assert existing.evidence_refs == ["origin_change:203.0.113.0/24:64497"] @pytest.mark.asyncio async def test_build_incident_geography_hints_prefers_evidence_prefix_scope_when_no_cached_prefix_geo(): incident = BGPIncident( source="ris_live_bgp", incident_key="origin_change:93.175.153.0/24:16509", incident_type="origin_change", title="Origin Change incident on 93.175.153.0/24", summary="summary", severity="critical", status="active", affected_prefixes=["93.175.153.0/24"], evidence_refs=["origin_change:93.175.153.0/24:16509"], ) anomaly = BGPAnomaly( source="ris_live_bgp", anomaly_type="origin_change", severity="critical", status="active", entity_key="origin_change:93.175.153.0/24:16509", prefix="93.175.153.0/24", origin_asn=12654, new_origin_asn=16509, summary="summary", confidence=0.8, evidence={ "prefix_scope": { "regions": [ { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, } ] } }, ) db = _FakeAsyncSession([[anomaly]]) hints = await build_incident_geography_hints(db, [incident]) assert hints["origin_change:93.175.153.0/24:16509"]["geography_mode"] == "prefix_scope" assert hints["origin_change:93.175.153.0/24:16509"]["regions"][0]["country"] == "Netherlands" @pytest.mark.asyncio async def test_infer_related_infrastructure_links_nearby_cables(): landing = CollectedData( source="arcgis_landing_points", name="Amsterdam Landing", data_type="landing_point", extra_data={ "city_id": 10, "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, ) relation = CollectedData( source="arcgis_cable_landing_relation", name="rel-1", data_type="landing_relation", extra_data={"city_id": 10, "cable_id": 20}, ) cable = CollectedData( source="arcgis_cables", name="AEConnect-1", data_type="cable", extra_data={"cable_id": 20}, ) db = _FakeAsyncSession([[landing], [relation], [cable]]) result = await infer_related_infrastructure( db, [ { "collector": "rrc00", "country": "Netherlands", "city": "Amsterdam", "latitude": 52.36, "longitude": 4.90, } ], ) assert len(result["related_cables"]) == 1 assert result["related_cables"][0]["landing_point"] == "Amsterdam Landing" assert result["related_cables"][0]["cable_names"] == ["AEConnect-1"] assert result["related_ixps"][0]["name"] == "Amsterdam, Netherlands" @pytest.mark.asyncio async def test_build_bgp_collector_coverage_summarizes_observations(): now = datetime.now(UTC) obs_one = BGPObservation( source="ris_live_bgp", collector="rrc00", prefix="203.0.113.0/24", origin_asn=64496, peer_asn=3333, event_type="announcement", observed_at=now, collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) obs_two = BGPObservation( source="ris_live_bgp", collector="rrc00", prefix="198.51.100.0/24", origin_asn=64497, peer_asn=3334, event_type="withdrawal", observed_at=now + timedelta(minutes=5), collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) db = _FakeAsyncSession([[obs_one, obs_two]]) coverage = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES) first = next(item for item in coverage if item["collector"] == "rrc00") assert first["observation_count"] == 2 assert first["recent_15m_observation_count"] == 2 assert first["recent_24h_observation_count"] == 2 assert first["recent_7d_observation_count"] == 2 assert first["prefix_count"] == 2 assert first["recent_15m_prefix_count"] == 2 assert first["recent_24h_prefix_count"] == 2 assert first["origin_asn_count"] == 2 assert first["latest_event_type"] == "withdrawal" assert first["baseline_scope"]["countries"] == ["Netherlands"] assert first["baseline_scope"]["cities"] == ["Amsterdam"] @pytest.mark.asyncio async def test_save_bgp_observations_for_batch_adds_rows(): db = _FakeAsyncSession([]) events = [ { "source_id": "evt-1", "description": "rrc00 observed announcement for 203.0.113.0/24", "reference_date": "2026-03-30T10:00:00Z", "metadata": { "collector": "rrc00", "peer_asn": 3333, "peer_ip": "2001:db8::1", "prefix": "203.0.113.0/24", "event_type": "announcement", "as_path": [3333, 64500, 64496], "origin_asn": 64496, "next_hop": "2001:db8::2", "communities": ["3333:100"], "timestamp": "2026-03-30T10:00:00Z", "collector_location": {"city": "Amsterdam", "country": "Netherlands"}, "raw_message": {"raw": "deadbeef"}, }, } ] created = await save_bgp_observations_for_batch( db, source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert created == 1 assert db.commits == 1 assert len(db.added) == 1 observation = db.added[0] assert observation.ingest_batch_id == "ris_live_bgp:2:1" assert observation.collector == "rrc00" assert observation.prefix == "203.0.113.0/24" @pytest.mark.asyncio async def test_create_bgp_anomalies_for_batch_calls_incident_aggregation(): previous_record = CollectedData( source="ris_live_bgp", extra_data={"prefix": "203.0.113.0/24", "origin_asn": 64496}, ) db = _FakeAsyncSession([ [], [], [], [previous_record], [], ]) events = [ { "reference_date": "2026-03-30T10:00:00Z", "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64497, "collector": "rrc00", "collector_location": { "country": "Netherlands", "city": "Amsterdam", "latitude": 52.3676, "longitude": 4.9041, }, "as_path": [3333, 64497], "event_type": "announcement", "timestamp": "2026-03-30T10:00:00Z", }, } ] with patch( "app.services.collectors.bgp_common.create_bgp_incidents_for_anomalies", new=AsyncMock(return_value=1), ) as incident_mock: created = await create_bgp_anomalies_for_batch( db, source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert created == 1 assert db.commits == 1 assert len(db.added) == 1 anomaly = db.added[0] assert anomaly.anomaly_type == "origin_change" assert anomaly.prefix == "203.0.113.0/24" incident_mock.assert_awaited_once() @pytest.mark.asyncio async def test_create_bgp_anomalies_for_batch_skips_existing_entity_keys(): previous_record = CollectedData( source="ris_live_bgp", extra_data={"prefix": "203.0.113.0/24", "origin_asn": 64496}, ) existing_key = ("origin_change:203.0.113.0/24:64497",) existing_anomaly = BGPAnomaly( source="ris_live_bgp", anomaly_type="origin_change", severity="critical", status="active", entity_key="origin_change:203.0.113.0/24:64497", prefix="203.0.113.0/24", origin_asn=64496, new_origin_asn=64497, ) db = _FakeAsyncSession([ [], [], [], [previous_record], [existing_key], [existing_anomaly], ]) events = [ { "reference_date": "2026-03-30T10:00:00Z", "metadata": { "prefix": "203.0.113.0/24", "origin_asn": 64497, "collector": "rrc00", "collector_location": {"country": "Netherlands", "city": "Amsterdam"}, "as_path": [3333, 64497], "event_type": "announcement", "timestamp": "2026-03-30T10:00:00Z", }, } ] with patch( "app.services.collectors.bgp_common.create_bgp_incidents_for_anomalies", new=AsyncMock(return_value=0), ) as incident_mock: created = await create_bgp_anomalies_for_batch( db, source="ris_live_bgp", snapshot_id=1, task_id=2, events=events, ) assert created == 0 assert len(db.added) == 0 incident_mock.assert_awaited_once() async def _bgp_test_client(db_session): async def override_get_db(): yield db_session def override_get_current_user(): return User(id=1, username="testuser", email="test@example.com", password_hash="x", role="admin") app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[get_current_user] = override_get_current_user transport = ASGITransport(app=app) client = AsyncClient(transport=transport, base_url="http://test") return client @pytest.mark.asyncio async def test_bgp_events_api_lists_observations(): observation = BGPObservation( id=1, source=BGP_SOURCES[0], collector="rrc00", peer_asn=3333, prefix="203.0.113.0/24", event_type="announcement", as_path=[3333, 64500, 64496], origin_asn=64496, observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), ) db = _FakeAsyncSession([[observation]]) client = await _bgp_test_client(db) try: response = await client.get("/api/v1/bgp/events") finally: await client.aclose() app.dependency_overrides.clear() assert response.status_code == 200 payload = response.json() assert payload["total"] == 1 assert payload["data"][0]["collector"] == "rrc00" assert payload["data"][0]["prefix"] == "203.0.113.0/24" @pytest.mark.asyncio async def test_bgp_incidents_api_returns_incident(): incident = BGPIncident( id=7, source="ris_live_bgp", incident_key="origin_change:203.0.113.0/24:64497", incident_type="origin_change", title="Origin Change incident on 203.0.113.0/24", summary="Grouped incident summary", severity="critical", status="active", confidence=0.91, affected_prefixes=["203.0.113.0/24"], affected_collectors=["rrc00"], ) db = _FakeAsyncSession( [[incident]], gets={(BGPIncident, 7): incident}, ) client = await _bgp_test_client(db) try: list_response = await client.get("/api/v1/bgp/incidents") detail_response = await client.get("/api/v1/bgp/incidents/7") finally: await client.aclose() app.dependency_overrides.clear() assert list_response.status_code == 200 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 @pytest.mark.asyncio async def test_bgp_collectors_api_returns_coverage(): now = datetime.now(UTC) observation = BGPObservation( id=1, source="ris_live_bgp", collector="rrc00", peer_asn=3333, prefix="203.0.113.0/24", event_type="announcement", origin_asn=64496, observed_at=now, collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) db = _FakeAsyncSession([[observation], [observation]]) client = await _bgp_test_client(db) try: list_response = await client.get("/api/v1/bgp/collectors") summary_response = await client.get("/api/v1/bgp/collectors/summary") finally: await client.aclose() app.dependency_overrides.clear() assert list_response.status_code == 200 list_payload = list_response.json() assert list_payload["total"] >= 1 target = next(item for item in list_payload["data"] if item["collector"] == "rrc00") assert target["observation_count"] == 1 assert target["prefix_count"] == 1 assert summary_response.status_code == 200 summary_payload = summary_response.json() assert summary_payload["active_collectors"] >= 1 assert summary_payload["observed_prefixes"] >= 1 assert summary_payload["recent_24h_events"] >= 1 assert summary_payload["recent_7d_events"] >= 1