extract BgpContext dataclass for build_bgp_intel

src/netsleuth/bgp.py

@@ -108,6 +108,7 @@ import json
 import os
 import time
 from pathlib import Path
+from dataclasses import dataclass
 from typing import Awaitable, Callable
 
 from netsleuth.models import IxpPresence
@@ -177,25 +178,28 @@ async def cached_json(
 from netsleuth.models import BgpIntel
 
 
-def build_bgp_intel(
-    asn: str,
-    overview: dict,
-    neighbours: dict,
-    prefixes: dict,
-    updates: dict,
-    asrank: dict | None,
-    pdb_net: dict | None,
-    pdb_ixlan: dict | None,
-    timeframe_days: int,
-) -> BgpIntel:
-    info = parse_as_overview(overview)
-    upstreams, peers, downstreams = parse_asn_neighbours(neighbours)
-    announced, v4, v6 = parse_announced_prefixes(prefixes)
-    events = parse_bgp_updates(updates)
-    rank, cone_asns, cone_prefixes = parse_asrank(asrank) if asrank else (None, None, None)
-    info_type, traffic, _net_id = parse_peeringdb_net(pdb_net) if pdb_net else (None, None, None)
+@dataclass
+class BgpContext:
+    asn: str
+    overview: dict
+    neighbours: dict
+    prefixes: dict
+    updates: dict
+    asrank: dict | None
+    pdb_net: dict | None
+    pdb_ixlan: dict | None
+    timeframe_days: int
+
+
+def build_bgp_intel(ctx: BgpContext) -> BgpIntel:
+    info = parse_as_overview(ctx.overview)
+    upstreams, peers, downstreams = parse_asn_neighbours(ctx.neighbours)
+    announced, v4, v6 = parse_announced_prefixes(ctx.prefixes)
+    events = parse_bgp_updates(ctx.updates)
+    rank, cone_asns, cone_prefixes = parse_asrank(ctx.asrank) if ctx.asrank else (None, None, None)
+    info_type, traffic, _net_id = parse_peeringdb_net(ctx.pdb_net) if ctx.pdb_net else (None, None, None)
     return BgpIntel(
-        asn=asn,
+        asn=ctx.asn,
         holder=info["holder"],
         registry=info["registry"],
         allocated_at=info["allocated_at"],
@@ -206,8 +210,8 @@ def build_bgp_intel(
         prefix_count_v4=v4,
         prefix_count_v6=v6,
         flaps=events,
-        stability=classify_stability(events, timeframe_days),
-        ixps=parse_peeringdb_netixlan(pdb_ixlan) if pdb_ixlan else [],
+        stability=classify_stability(events, ctx.timeframe_days),
+        ixps=parse_peeringdb_netixlan(ctx.pdb_ixlan) if ctx.pdb_ixlan else [],
         pdb_info_type=info_type,
         pdb_traffic=traffic,
         asrank=rank,
@@ -284,15 +288,17 @@ async def collect_bgp(
                 pdb_ixlan = None
 
     intel = build_bgp_intel(
-        asn=asn,
-        overview=overview or {},
-        neighbours=neighbours or {},
-        prefixes=prefixes or {},
-        updates=updates or {},
-        asrank=asrank,
-        pdb_net=pdb_net,
-        pdb_ixlan=pdb_ixlan,
-        timeframe_days=providers.ripestat_timeframe_days,
+        BgpContext(
+            asn=asn,
+            overview=overview or {},
+            neighbours=neighbours or {},
+            prefixes=prefixes or {},
+            updates=updates or {},
+            asrank=asrank,
+            pdb_net=pdb_net,
+            pdb_ixlan=pdb_ixlan,
+            timeframe_days=providers.ripestat_timeframe_days,
+        )
     )
     raw = {
         "ripestat-as-overview": overview,

tests/test_bgp.py

@@ -211,20 +211,22 @@ async def test_cached_json_survives_a_corrupt_cache_file(tmp_path: Path):
     assert await cached_json(tmp_path, "k", ttl_hours=24, fetch=fetch) == {"value": "fresh"}
 
 
-from netsleuth.bgp import build_bgp_intel
+from netsleuth.bgp import BgpContext, build_bgp_intel
 
 
 def test_build_bgp_intel_merges_every_source(api_fixture):
     intel = build_bgp_intel(
-        asn="AS64500",
-        overview=api_fixture("ripestat_as_overview.json"),
-        neighbours=api_fixture("ripestat_asn_neighbours.json"),
-        prefixes=api_fixture("ripestat_announced_prefixes.json"),
-        updates=api_fixture("ripestat_bgp_updates.json"),
-        asrank=api_fixture("asrank.json"),
-        pdb_net=api_fixture("peeringdb_net.json"),
-        pdb_ixlan=api_fixture("peeringdb_netixlan.json"),
-        timeframe_days=14,
+        BgpContext(
+            asn="AS64500",
+            overview=api_fixture("ripestat_as_overview.json"),
+            neighbours=api_fixture("ripestat_asn_neighbours.json"),
+            prefixes=api_fixture("ripestat_announced_prefixes.json"),
+            updates=api_fixture("ripestat_bgp_updates.json"),
+            asrank=api_fixture("asrank.json"),
+            pdb_net=api_fixture("peeringdb_net.json"),
+            pdb_ixlan=api_fixture("peeringdb_netixlan.json"),
+            timeframe_days=14,
+        )
     )
     assert intel.asn == "AS64500"
     assert intel.holder == "EXAMPLE-AS Example Telecom BV"
@@ -241,15 +243,17 @@ def test_build_bgp_intel_merges_every_source(api_fixture):
 
 def test_build_bgp_intel_tolerates_every_optional_source_being_absent(api_fixture):
     intel = build_bgp_intel(
-        asn="AS64500",
-        overview=api_fixture("ripestat_as_overview.json"),
-        neighbours={},
-        prefixes={},
-        updates={},
-        asrank=None,
-        pdb_net=None,
-        pdb_ixlan=None,
-        timeframe_days=14,
+        BgpContext(
+            asn="AS64500",
+            overview=api_fixture("ripestat_as_overview.json"),
+            neighbours={},
+            prefixes={},
+            updates={},
+            asrank=None,
+            pdb_net=None,
+            pdb_ixlan=None,
+            timeframe_days=14,
+        )
     )
     assert intel.asn == "AS64500"
     assert intel.upstreams == []