From c14d4e81ffe9a4896fe24c4e9f7e16c489f69bda Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 22 Sep 2026 11:56:01 +0800 Subject: [PATCH] Point VRF BGP neighbor aux at config_bgp_peer (running-config bgp). Discover and aux both use zte.config_bgp_peer; enrich remote_as/activate/route-maps from intent. Guard with tests so aux templates cannot introduce extra placeholders. Co-authored-by: Cursor --- netx_api/biz_state/profiles.py | 61 ++++++++++++++++++++---------- tests/test_biz_state_profiles.py | 49 +++++++++++++++++++++++- tests/test_zte_extended_parsers.py | 41 +++++++++++++++----- 3 files changed, 120 insertions(+), 31 deletions(-) diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index 3482788..818e73e 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -434,6 +434,11 @@ _BGP_ROUTE_FIELDS: list[FieldDef] = [ FieldDef("as_num", length=16, role="state", display_name="AS"), FieldDef("state", length=64, role="state", display_name="PeerState"), FieldDef("pfx_rcd", length=32, role="meta", display_name="PfxRcd"), + # Intent enrich from config_bgp_peer (VRF neighbor in/out aux) + FieldDef("remote_as", length=16, role="state", display_name="Remote AS"), + FieldDef("activate", length=16, role="state", display_name="Activate"), + FieldDef("route_map_in", length=128, role="meta", display_name="RM In"), + FieldDef("route_map_out", length=128, role="meta", display_name="RM Out"), ] _IP_ROUTE_FIELDS: list[FieldDef] = [ @@ -1070,19 +1075,23 @@ def _zte_status_profiles() -> list[ParseProfile]: r"neighbor\s+(?Pin)\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" ), textfsm_command="show bgp vpnv4 unicast neighbor in", - description="Per-VRF CE peer routes; discover (vrf,neighbor) from BGP peer intent.", + description=( + "Per-VRF CE peer routes; discover (vrf,neighbor) from BGP peer intent; " + "aux: show running-config bgp (no placeholders)." + ), placeholders=[_BGP_VRF_PEER_IPV4, _BGP_NEIGHBOR_IPV4_VRF], fields=list(_BGP_ROUTE_FIELDS), tags=["bgp", "vpnv4", "vrf", "route"], sort_order=430, enabled=True, kind="collect", - aux_commands=[AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv4_vrf_summary")], + aux_commands=[AuxCommand(key="config_bgp_peer", profile_id="zte.config_bgp_peer")], enrich_joins=[ EnrichJoin( - from_aux="bgp_summary", - on="neighbor", - take=("as_num", "state", "pfx_rcd"), + from_aux="config_bgp_peer", + left_on="neighbor", + right_on="neighbor", + take=("remote_as", "activate", "route_map_in", "route_map_out"), ), ], ), @@ -1098,7 +1107,10 @@ def _zte_status_profiles() -> list[ParseProfile]: r"neighbor\s+(?Pout)\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" ), textfsm_command="show bgp vpnv4 unicast neighbor out", - description="Per-VRF CE peer advertised routes; discover pairs from BGP peer intent.", + description=( + "Per-VRF CE advertised routes; discover pairs from BGP peer intent; " + "aux: show running-config bgp." + ), placeholders=[_BGP_VRF_PEER_IPV4, _BGP_NEIGHBOR_IPV4_VRF], fields=list(_BGP_ROUTE_FIELDS), tags=["bgp", "vpnv4", "vrf", "route"], @@ -1106,12 +1118,13 @@ def _zte_status_profiles() -> list[ParseProfile]: enabled=True, kind="collect", collect_lane="heavy", - aux_commands=[AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv4_vrf_summary")], + aux_commands=[AuxCommand(key="config_bgp_peer", profile_id="zte.config_bgp_peer")], enrich_joins=[ EnrichJoin( - from_aux="bgp_summary", - on="neighbor", - take=("as_num", "state", "pfx_rcd"), + from_aux="config_bgp_peer", + left_on="neighbor", + right_on="neighbor", + take=("remote_as", "activate", "route_map_in", "route_map_out"), ), ], ), @@ -1127,19 +1140,23 @@ def _zte_status_profiles() -> list[ParseProfile]: r"neighbor\s+(?Pin)\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" ), textfsm_command="show bgp vpnv6 unicast neighbor in", - description="Per-VRF IPv6 CE peer routes; discover pairs from BGP peer intent.", + description=( + "Per-VRF IPv6 CE peer routes; discover pairs from BGP peer intent; " + "aux: show running-config bgp." + ), placeholders=[_BGP_VRF_PEER_IPV6, _BGP_NEIGHBOR_IPV6_VRF], fields=list(_BGP_ROUTE_FIELDS), tags=["bgp", "vpnv6", "vrf", "route"], sort_order=440, enabled=True, kind="collect", - aux_commands=[AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv6_vrf_summary")], + aux_commands=[AuxCommand(key="config_bgp_peer", profile_id="zte.config_bgp_peer")], enrich_joins=[ EnrichJoin( - from_aux="bgp_summary", - on="neighbor", - take=("as_num", "state", "pfx_rcd"), + from_aux="config_bgp_peer", + left_on="neighbor", + right_on="neighbor", + take=("remote_as", "activate", "route_map_in", "route_map_out"), ), ], ), @@ -1155,7 +1172,10 @@ def _zte_status_profiles() -> list[ParseProfile]: r"neighbor\s+(?Pout)\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" ), textfsm_command="show bgp vpnv6 unicast neighbor out", - description="Per-VRF IPv6 CE advertised routes; discover pairs from BGP peer intent.", + description=( + "Per-VRF IPv6 CE advertised routes; discover pairs from BGP peer intent; " + "aux: show running-config bgp." + ), placeholders=[_BGP_VRF_PEER_IPV6, _BGP_NEIGHBOR_IPV6_VRF], fields=list(_BGP_ROUTE_FIELDS), tags=["bgp", "vpnv6", "vrf", "route"], @@ -1163,12 +1183,13 @@ def _zte_status_profiles() -> list[ParseProfile]: enabled=True, kind="collect", collect_lane="heavy", - aux_commands=[AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv6_vrf_summary")], + aux_commands=[AuxCommand(key="config_bgp_peer", profile_id="zte.config_bgp_peer")], enrich_joins=[ EnrichJoin( - from_aux="bgp_summary", - on="neighbor", - take=("as_num", "state", "pfx_rcd"), + from_aux="config_bgp_peer", + left_on="neighbor", + right_on="neighbor", + take=("remote_as", "activate", "route_map_in", "route_map_out"), ), ], ), diff --git a/tests/test_biz_state_profiles.py b/tests/test_biz_state_profiles.py index 6a904e2..96c9345 100644 --- a/tests/test_biz_state_profiles.py +++ b/tests/test_biz_state_profiles.py @@ -5,9 +5,15 @@ from __future__ import annotations import unittest from netx_api.biz_state.command_match import expand_from_bindings, match_command, preview_task_item -from netx_api.biz_state.profiles import all_profiles, get_profile, profiles_for_vendor +from netx_api.biz_state.profiles import ( + all_profiles, + get_profile, + profile_to_public_dict, + profiles_for_vendor, +) from netx_api.biz_state.parsers import normalize_lldp_neighbors from netx_api.lldp_shared import parse_neighbor_output +import re class BizStateProfileTests(unittest.TestCase): @@ -58,6 +64,47 @@ class BizStateProfileTests(unittest.TestCase): hits = parse_neighbor_output("", vendor="zte", device_type="zte_zxros") self.assertEqual(hits, []) + def test_vrf_neighbor_discover_and_aux_both_config_bgp(self) -> None: + for pid in ( + "zte.bgp_vpnv4_vrf_neighbor_in", + "zte.bgp_vpnv4_vrf_neighbor_out", + "zte.bgp_vpnv6_vrf_neighbor_in", + "zte.bgp_vpnv6_vrf_neighbor_out", + ): + p = get_profile(pid) + assert p is not None + for ph in p.placeholders: + self.assertEqual(ph.discover_profile_id, "zte.config_bgp_peer") + self.assertEqual([a.profile_id for a in p.aux_commands], ["zte.config_bgp_peer"]) + pub = profile_to_public_dict(p) + aux = pub.get("aux_commands") or [] + self.assertEqual(len(aux), 1) + self.assertIn("running-config bgp", aux[0]["command_template"]) + + def test_aux_templates_placeholders_subset_of_primary(self) -> None: + """Aux CLI must not introduce placeholders outside primary params. + + Prevents cases like VRF neighbor aux pointing at summary with + while the UI shows both as if they were independent unbound vars. + """ + ph_re = re.compile(r"<([^>]+)>") + for p in all_profiles(): + if not p.enabled or p.kind != "collect": + continue + primary_ph = {ph.name for ph in (p.placeholders or [])} + for a in p.aux_commands or []: + ap = get_profile(a.profile_id) + if ap is None: + continue + tmpl = str(ap.command_template or "") + aux_ph = set(ph_re.findall(tmpl)) + extra = aux_ph - primary_ph + self.assertFalse( + extra, + f"{p.profile_id} aux {a.key} ({a.profile_id}) has " + f"placeholders {extra} not in primary {primary_ph}", + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index 16284bf..1788fb4 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -375,24 +375,43 @@ class ZteExtendedParserTests(unittest.TestCase): self.assertEqual(routes[0]["network"], "10.1.0.0/24") self.assertEqual(routes[0]["direction"], "in") - peers = normalize_bgp_peer( - raw_text=_BGP_V4_SUMMARY, command="show bgp vpnv4 unicast vrf CUST_A summary" - ) from netx_api.biz_state.enrich import EnrichJoin + from netx_api.biz_state.parsers.zte.config_bgp_peer import normalize_config_bgp_peer + intent = normalize_config_bgp_peer( + raw_text=""" +! + neighbor 10.0.0.1 remote-as 65001 + address-family ipv4 vrf CUST_A + neighbor 10.0.0.1 activate + neighbor 10.0.0.1 route-map RM_IN in + $ +$ +""", + command="show running-config bgp", + ) apply_enrich_joins( routes, - {"bgp_summary": peers}, - [EnrichJoin(from_aux="bgp_summary", on="neighbor", take=("as_num", "state", "pfx_rcd"))], + {"config_bgp_peer": intent}, + [ + EnrichJoin( + from_aux="config_bgp_peer", + on="neighbor", + take=("remote_as", "activate", "route_map_in", "route_map_out"), + ) + ], ) - self.assertEqual(routes[0]["as_num"], "65001") - self.assertEqual(routes[0]["state"], "Established") + self.assertEqual(routes[0]["remote_as"], "65001") + self.assertEqual(routes[0]["activate"], "enable") ra = resolve_aux_command( - AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv4_vrf_summary"), - params={"vrf": "CUST_A"}, + AuxCommand(key="config_bgp_peer", profile_id="zte.config_bgp_peer"), + params={"vrf": "CUST_A", "neighbor": "10.0.0.1"}, ) - self.assertEqual(ra.command, "show bgp vpnv4 unicast vrf CUST_A summary | one-line") + self.assertEqual(ra.command, "show running-config bgp | one-line") + # Aux has no placeholders — params must not leak into the CLI + self.assertNotIn("<", ra.command) + self.assertNotIn("CUST_A", ra.command) def test_bgp_route_empty_total_skips_header(self) -> None: empty_in = normalize_bgp_route( @@ -545,6 +564,8 @@ class ZteExtendedParserTests(unittest.TestCase): p = get_profile("zte.bgp_vpnv4_vrf_neighbor_in") assert p is not None + self.assertEqual([a.key for a in p.aux_commands], ["config_bgp_peer"]) + self.assertEqual(p.aux_commands[0].profile_id, "zte.config_bgp_peer") pairs = expand_from_bindings( profile=p, bindings=[{"vrf": "CUST_A", "neighbor": "10.0.0.1"}],