diff --git a/osism/tasks/conductor/sonic/config_generator.py b/osism/tasks/conductor/sonic/config_generator.py index 3f707563e..76d72e432 100644 --- a/osism/tasks/conductor/sonic/config_generator.py +++ b/osism/tasks/conductor/sonic/config_generator.py @@ -36,9 +36,16 @@ get_connected_interfaces, get_connected_device_for_sonic_interface, get_connected_interface_ipv4_address, + get_connected_interface_ip_addresses, + is_numbered_neighbor_address, ) from .cache import get_cached_device_interfaces -from .constants import BGP_AF_L2VPN_EVPN_TAG, DEFAULT_SONIC_ROLES +from .constants import ( + BGP_AF_IPV4_UNICAST, + BGP_AF_IPV6_UNICAST, + BGP_AF_L2VPN_EVPN_TAG, + DEFAULT_SONIC_ROLES, +) # Global cache for metalbox IPs per device to avoid duplicate lookups _metalbox_ip_cache: dict[int, Optional[str]] = {} @@ -1090,12 +1097,12 @@ def get_vrf_for_interface(interface_name): neighbor_id = port_name vrf_name = get_vrf_for_interface(port_name) - ipv4_key = f"{vrf_name}|{neighbor_id}|ipv4_unicast" + ipv4_key = f"{vrf_name}|{neighbor_id}|{BGP_AF_IPV4_UNICAST}" config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": "true"} # Only add ipv6_unicast if v6only would be true (no transfer role IPv4) if not has_transfer_ipv4: - ipv6_key = f"{vrf_name}|{neighbor_id}|ipv6_unicast" + ipv6_key = f"{vrf_name}|{neighbor_id}|{BGP_AF_IPV6_UNICAST}" config["BGP_NEIGHBOR_AF"][ipv6_key] = {"admin_status": "true"} logger.debug( f"Added BGP_NEIGHBOR_AF with ipv4_unicast and ipv6_unicast for interface {port_name} (no direct IPv4)" @@ -1156,8 +1163,8 @@ def get_vrf_for_interface(interface_name): neighbor_id = pc_name vrf_name = get_vrf_for_interface(pc_name) - ipv4_key = f"{vrf_name}|{neighbor_id}|ipv4_unicast" - ipv6_key = f"{vrf_name}|{neighbor_id}|ipv6_unicast" + ipv4_key = f"{vrf_name}|{neighbor_id}|{BGP_AF_IPV4_UNICAST}" + ipv6_key = f"{vrf_name}|{neighbor_id}|{BGP_AF_IPV6_UNICAST}" config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": "true"} config["BGP_NEIGHBOR_AF"][ipv6_key] = {"admin_status": "true"} @@ -1346,6 +1353,25 @@ def get_vrf_for_interface(interface_name): if not addresses: continue + # Address families the local SVI can source. A numbered BGP + # session only comes up when the SVI itself carries a routable + # address in that family, so an IPv4-only SVI must not emit an + # IPv6 numbered neighbor (and vice versa); link-local-only + # addresses do not count (see is_numbered_neighbor_address). + local_address_families = { + version + for version in (4, 6) + if any( + is_numbered_neighbor_address(addr, version) for addr in addresses + ) + } + if not local_address_families: + logger.debug( + f"VLAN {vid} SVI has no routable IP address, " + f"skipping BGP configuration" + ) + continue + # Find untagged member interfaces for this VLAN # Only untagged members are relevant for VLAN BGP neighbors if vid not in vlan_info["vlan_members"]: @@ -1387,35 +1413,50 @@ def get_vrf_for_interface(interface_name): ) continue - # Get peer IP address using the existing FHRP VIP detection logic - peer_ipv4 = None + # Look up both peer address families in a single NetBox + # round-trip (direct IP first, then FHRP VIP), then keep only + # the families the local SVI can actually source. A dual-stack + # peer yields one neighbor per address family. + peer_ip_afs = [] if netbox: - peer_ipv4 = get_connected_interface_ipv4_address( + peer_ipv4, peer_ipv6 = get_connected_interface_ip_addresses( device, sonic_iface_name, netbox ) + if peer_ipv4 and 4 in local_address_families: + peer_ip_afs.append((peer_ipv4, BGP_AF_IPV4_UNICAST)) + if peer_ipv6 and 6 in local_address_families: + peer_ip_afs.append((peer_ipv6, BGP_AF_IPV6_UNICAST)) + + if not peer_ip_afs: + logger.debug( + f"No peer IP address found for interface {sonic_iface_name} " + f"(NetBox: {netbox_iface_name}) in VLAN {vid}" + ) + continue - if peer_ipv4: + for peer_ip, address_family in peer_ip_afs: # Avoid duplicate peer IPs across multiple untagged members - if peer_ipv4 in peer_ips_found: + if peer_ip in peer_ips_found: logger.debug( - f"Peer IP {peer_ipv4} already configured for VLAN {vid}, " + f"Peer IP {peer_ip} already configured for VLAN {vid}, " f"skipping duplicate from interface {sonic_iface_name}" ) continue - peer_ips_found.add(peer_ipv4) + peer_ips_found.add(peer_ip) # Get VRF for the VLAN interface (e.g., Vlan100) vlan_interface_name = f"Vlan{vid}" vrf_name = get_vrf_for_interface(vlan_interface_name) # Create BGP neighbor with peer IP address (FHRP VIP or direct IP) - neighbor_key = f"{vrf_name}|{peer_ipv4}" + neighbor_key = f"{vrf_name}|{peer_ip}" # Determine peer_type - for VLAN interfaces, default to external peer_type = "external" - # Set v6only=false for IPv4 BGP neighbor (only for default VRF) + # The neighbor is reached over a numbered (global) address, so + # v6only does not apply; set it to false for the default VRF. bgp_neighbor_config = { "peer_type": peer_type, } @@ -1424,18 +1465,14 @@ def get_vrf_for_interface(interface_name): config["BGP_NEIGHBOR"][neighbor_key] = bgp_neighbor_config - # Add BGP_NEIGHBOR_AF for IPv4 unicast - ipv4_af_key = f"{vrf_name}|{peer_ipv4}|ipv4_unicast" - config["BGP_NEIGHBOR_AF"][ipv4_af_key] = {"admin_status": "true"} + # Add BGP_NEIGHBOR_AF for the matching address family + af_key = f"{vrf_name}|{peer_ip}|{address_family}" + config["BGP_NEIGHBOR_AF"][af_key] = {"admin_status": "true"} logger.info( - f"Added BGP neighbor configuration for VLAN {vid} using peer IP {peer_ipv4} " - f"from connected interface {sonic_iface_name} (NetBox: {netbox_iface_name})" - ) - else: - logger.debug( - f"No peer IPv4 address found for interface {sonic_iface_name} " - f"(NetBox: {netbox_iface_name}) in VLAN {vid}" + f"Added BGP neighbor configuration for VLAN {vid} using peer IP {peer_ip} " + f"({address_family}) from connected interface {sonic_iface_name} " + f"(NetBox: {netbox_iface_name})" ) if not peer_ips_found: diff --git a/osism/tasks/conductor/sonic/connections.py b/osism/tasks/conductor/sonic/connections.py index 2c6cdfd38..5e2f22b07 100644 --- a/osism/tasks/conductor/sonic/connections.py +++ b/osism/tasks/conductor/sonic/connections.py @@ -6,6 +6,8 @@ using the NetBox connected_endpoints API, replacing legacy cable-based detection. """ +import ipaddress + from loguru import logger from typing import Optional, Set, Tuple, List, Any, DefaultDict from collections import defaultdict @@ -423,10 +425,41 @@ def get_device_bgp_neighbors_via_loopback( return bgp_neighbors -def get_connected_interface_ipv4_address(device, sonic_port_name, netbox): +def is_numbered_neighbor_address(address, ip_version): + """Return True if ``address`` can be used as a *numbered* BGP neighbor of + ``ip_version``. + + ``address`` may be bare (``10.0.0.1``) or carry a prefix length + (``10.0.0.1/24``); it is parsed with :mod:`ipaddress`, which is more + robust than inspecting the string for ``.`` versus ``:``. + + Link-local, loopback and multicast addresses are rejected: a numbered + session needs a routable address on both ends, so e.g. a + ``BGP_NEIGHBOR[...|fe80::...]`` entry would never come up -- link-local + peers require the unnumbered/``v6only`` form instead. An unparseable + address matches no version. """ - Get the IPv4 address(es) of the connected endpoint interface for a given SONiC port. - Checks for direct IP addresses first, then for FHRP VIP addresses. + try: + ip = ipaddress.ip_interface(address).ip + except ValueError: + return False + if ip.version != ip_version: + return False + return not (ip.is_link_local or ip.is_loopback or ip.is_multicast) + + +def get_connected_interface_ip_addresses(device, sonic_port_name, netbox): + """Return ``(ipv4, ipv6)`` peer addresses for the interface connected to + ``sonic_port_name`` using a single NetBox round-trip. + + The connected interface, its directly-assigned IP addresses and its FHRP + VIPs are each fetched once and scanned for both address families, so a + caller that needs a dual-stack peer pays one set of lookups instead of + two. For each family a directly-assigned address takes precedence over an + FHRP VIP -- this precedence is inherited from the original IPv4-only + helper and is deliberate (do not "fix" it toward VIP peering). Only + addresses usable as numbered neighbors are returned (see + :func:`is_numbered_neighbor_address`). Args: device: The SONiC device @@ -434,26 +467,25 @@ def get_connected_interface_ipv4_address(device, sonic_port_name, netbox): netbox: The NetBox API client Returns: - - For direct IP: The IPv4 address string of the connected interface - - For FHRP VIP: The first VIP address found (if multiple VIPs exist, logs all but returns first) - - None if no addresses found + A ``(ipv4, ipv6)`` tuple of address strings; either element is + ``None`` when no usable address of that family was found. """ try: interface = netbox.dcim.interfaces.get( device_id=device.id, name=sonic_port_name ) if not interface: - return None + return None, None # Check if interface has connected_endpoints using the modern API if not ( hasattr(interface, "connected_endpoints") and interface.connected_endpoints ): - return None + return None, None # Ensure connected_endpoints_reachable is True if not getattr(interface, "connected_endpoints_reachable", False): - return None + return None, None # Process each connected endpoint to find the first valid interface connected_interface = None @@ -463,27 +495,35 @@ def get_connected_interface_ipv4_address(device, sonic_port_name, netbox): break if not connected_interface: - return None + return None, None - # First, try to get direct IPv4 addresses assigned to the connected interface + # First, collect directly-assigned addresses of the connected + # interface. A directly-assigned address wins over an FHRP VIP of the + # same family -- this precedence is deliberate, do not "fix" it toward + # VIP peering. + direct = {4: None, 6: None} ip_addresses = netbox.ipam.ip_addresses.filter( assigned_object_id=connected_interface.id, ) - for ip_address in ip_addresses: - # Check if it's an IPv4 address - if "/" in str(ip_address.address): - address = str(ip_address.address).split("/")[0] - if "." in address: # IPv4 address + address = str(ip_address.address).split("/")[0] + for version in (4, 6): + if direct[version] is None and is_numbered_neighbor_address( + address, version + ): + direct[version] = address logger.debug( - f"Found direct IPv4 address {address} on connected interface " - f"{connected_interface.name} for port {sonic_port_name}" + f"Found direct IPv{version} address {address} on connected " + f"interface {connected_interface.name} for port {sonic_port_name}" ) - return address - # If no direct IP found, check for FHRP group membership and VIP addresses + # Only query FHRP when at least one family still lacks a direct address. + if direct[4] is not None and direct[6] is not None: + return direct[4], direct[6] + logger.debug( - f"No direct IPv4 found on {connected_interface.name}, checking for FHRP VIP addresses" + f"Missing a direct address for at least one family on " + f"{connected_interface.name}, checking for FHRP VIP addresses" ) # Get FHRP group assignments for the connected interface @@ -503,49 +543,60 @@ def get_connected_interface_ipv4_address(device, sonic_port_name, netbox): logger.debug(f"Could not query VIP addresses: {vip_e}") all_vip_addresses = [] - # Collect all VIP IPv4 addresses from all FHRP groups this interface belongs to - vip_addresses_found = [] - + # Collect the first matching VIP per family across all FHRP groups this + # interface belongs to. + vip = {4: None, 6: None} for assignment in fhrp_assignments: if not assignment.group: continue # Find VIP addresses assigned to this specific FHRP group - for vip in all_vip_addresses: + for candidate in all_vip_addresses: # Check if this VIP is assigned to the current FHRP group - if ( - hasattr(vip, "assigned_object_type") - and vip.assigned_object_type == "ipam.fhrpgroup" - and hasattr(vip, "assigned_object_id") - and vip.assigned_object_id == assignment.group.id + if not ( + hasattr(candidate, "assigned_object_type") + and candidate.assigned_object_type == "ipam.fhrpgroup" + and hasattr(candidate, "assigned_object_id") + and candidate.assigned_object_id == assignment.group.id ): - # Check if it's an IPv4 address - if "/" in str(vip.address): - address = str(vip.address).split("/")[0] - if "." in address: # IPv4 address - vip_addresses_found.append(address) - logger.debug( - f"Found FHRP VIP address {address} for connected interface " - f"{connected_interface.name} (FHRP group: {assignment.group.name or assignment.group.id}) " - f"for port {sonic_port_name}" - ) - - # Return the first VIP address found (for BGP neighbor compatibility) - if vip_addresses_found: - logger.debug( - f"Found {len(vip_addresses_found)} VIP addresses for port {sonic_port_name}: {vip_addresses_found}" - ) - logger.debug(f"Returning first VIP address: {vip_addresses_found[0]}") - return vip_addresses_found[0] + continue + + address = str(candidate.address).split("/")[0] + for version in (4, 6): + if vip[version] is None and is_numbered_neighbor_address( + address, version + ): + vip[version] = address + logger.debug( + f"Found FHRP VIP IPv{version} address {address} for " + f"connected interface {connected_interface.name} " + f"(FHRP group: {assignment.group.name or assignment.group.id}) " + f"for port {sonic_port_name}" + ) - logger.debug( - f"No IPv4 address (direct or FHRP VIP) found on connected interface " - f"{connected_interface.name} for port {sonic_port_name}" - ) - return None + return (direct[4] or vip[4]), (direct[6] or vip[6]) except Exception as e: logger.warning( - f"Could not get connected interface IPv4 for port {sonic_port_name}: {e}" + f"Could not get connected interface IP addresses for port " + f"{sonic_port_name}: {e}" ) - return None + return None, None + + +def get_connected_interface_ipv4_address(device, sonic_port_name, netbox): + """Get the IPv4 address of the connected endpoint interface for a SONiC port. + + Thin wrapper around :func:`get_connected_interface_ip_addresses`; see that + function for the lookup details (direct IP first, then FHRP VIP). + """ + return get_connected_interface_ip_addresses(device, sonic_port_name, netbox)[0] + + +def get_connected_interface_ipv6_address(device, sonic_port_name, netbox): + """Get the IPv6 address of the connected endpoint interface for a SONiC port. + + Thin wrapper around :func:`get_connected_interface_ip_addresses`; see that + function for the lookup details (direct IP first, then FHRP VIP). + """ + return get_connected_interface_ip_addresses(device, sonic_port_name, netbox)[1] diff --git a/osism/tasks/conductor/sonic/constants.py b/osism/tasks/conductor/sonic/constants.py index e8f5f00f6..4f34b8a17 100644 --- a/osism/tasks/conductor/sonic/constants.py +++ b/osism/tasks/conductor/sonic/constants.py @@ -5,6 +5,10 @@ # Tag to add AF L2VPN EVPN to BGP neighbor BGP_AF_L2VPN_EVPN_TAG = "bgp-af-l2vpn-evpn" +# BGP address-family (afi-safi) identifiers used in BGP_NEIGHBOR_AF keys +BGP_AF_IPV4_UNICAST = "ipv4_unicast" +BGP_AF_IPV6_UNICAST = "ipv6_unicast" + # Default AS prefix for local ASN calculation DEFAULT_LOCAL_AS_PREFIX = 4200 diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_bgp_vlan_vrf.py b/tests/unit/tasks/conductor/sonic/test_config_generator_bgp_vlan_vrf.py index 555845768..88136cc90 100644 --- a/tests/unit/tasks/conductor/sonic/test_config_generator_bgp_vlan_vrf.py +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_bgp_vlan_vrf.py @@ -60,10 +60,12 @@ def bgp_config(): @pytest.fixture def patch_bgp(mocker): - """Patch the two connection helpers ``_add_bgp_configurations`` calls. + """Patch the connection helpers ``_add_bgp_configurations`` calls. - ``connected_device`` returns ``None`` and ``peer_ipv4`` returns ``None`` - by default; tests override ``.return_value`` / ``.side_effect``. + ``connected_device`` returns ``None``; the connected-interface/port-channel + peer lookup ``peer_ipv4`` returns ``None`` and the combined VLAN peer + lookup ``peer_ips`` returns ``(None, None)`` by default. Tests override + ``.return_value`` / ``.side_effect``. """ return SimpleNamespace( connected_device=mocker.patch.object( @@ -76,6 +78,11 @@ def patch_bgp(mocker): "get_connected_interface_ipv4_address", return_value=None, ), + peer_ips=mocker.patch.object( + config_generator, + "get_connected_interface_ip_addresses", + return_value=(None, None), + ), ) @@ -346,7 +353,7 @@ def _vlan_info(self, members, addresses=None): } def test_untagged_member_with_peer_ip(self, bgp_config, patch_bgp): - patch_bgp.peer_ipv4.return_value = "192.0.2.20" + patch_bgp.peer_ips.return_value = ("192.0.2.20", None) _call_bgp( bgp_config, netbox=object(), @@ -361,8 +368,104 @@ def test_untagged_member_with_peer_ip(self, bgp_config, patch_bgp): "admin_status": "true" } + def test_untagged_member_with_peer_ipv6(self, bgp_config, patch_bgp): + patch_bgp.peer_ips.return_value = (None, "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + # IPv6 SVI so the IPv6 peer is sourceable (see address-family gate). + vlan_info=self._vlan_info( + {"eth0": "untagged"}, addresses=["2001:db8::1/64"] + ), + ) + assert bgp_config["BGP_NEIGHBOR"]["default|2001:db8::20"] == { + "peer_type": "external", + "v6only": "false", + } + assert bgp_config["BGP_NEIGHBOR_AF"]["default|2001:db8::20|ipv6_unicast"] == { + "admin_status": "true" + } + assert "default|2001:db8::20|ipv4_unicast" not in bgp_config["BGP_NEIGHBOR_AF"] + + def test_dual_stack_peer_adds_both_neighbors(self, bgp_config, patch_bgp): + patch_bgp.peer_ips.return_value = ("192.0.2.20", "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + vlan_info=self._vlan_info( + {"eth0": "untagged"}, + addresses=["10.0.0.1/24", "2001:db8::1/64"], + ), + ) + assert set(bgp_config["BGP_NEIGHBOR"]) == { + "default|192.0.2.20", + "default|2001:db8::20", + } + assert bgp_config["BGP_NEIGHBOR_AF"]["default|192.0.2.20|ipv4_unicast"] == { + "admin_status": "true" + } + assert bgp_config["BGP_NEIGHBOR_AF"]["default|2001:db8::20|ipv6_unicast"] == { + "admin_status": "true" + } + + def test_ipv4_only_svi_skips_ipv6_peer(self, bgp_config, patch_bgp): + # The SVI sources only IPv4, so an IPv6 peer must not be emitted as a + # numbered neighbor -- it could never come up without a local v6 route. + patch_bgp.peer_ips.return_value = ("192.0.2.20", "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + vlan_info=self._vlan_info({"eth0": "untagged"}, addresses=["10.0.0.1/24"]), + ) + assert set(bgp_config["BGP_NEIGHBOR"]) == {"default|192.0.2.20"} + assert "default|2001:db8::20" not in bgp_config["BGP_NEIGHBOR"] + assert "default|2001:db8::20|ipv6_unicast" not in bgp_config["BGP_NEIGHBOR_AF"] + + def test_ipv6_only_svi_skips_ipv4_peer(self, bgp_config, patch_bgp): + # Mirror of the above: an IPv6-only SVI must not emit an IPv4 neighbor. + patch_bgp.peer_ips.return_value = ("192.0.2.20", "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + vlan_info=self._vlan_info( + {"eth0": "untagged"}, addresses=["2001:db8::1/64"] + ), + ) + assert set(bgp_config["BGP_NEIGHBOR"]) == {"default|2001:db8::20"} + assert "default|192.0.2.20" not in bgp_config["BGP_NEIGHBOR"] + + def test_link_local_only_svi_skipped(self, bgp_config, patch_bgp): + # A link-local-only SVI cannot source a numbered session in any family. + patch_bgp.peer_ips.return_value = ("192.0.2.20", "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + vlan_info=self._vlan_info({"eth0": "untagged"}, addresses=["fe80::1/64"]), + ) + assert bgp_config["BGP_NEIGHBOR"] == {} + + def test_ipv6_peer_non_default_vrf_no_v6only(self, bgp_config, patch_bgp): + patch_bgp.peer_ips.return_value = (None, "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={"Ethernet0": _nbif("eth0")}, + vlan_info=self._vlan_info( + {"eth0": "untagged"}, addresses=["2001:db8::1/64"] + ), + vrf_info={"interface_vrf_mapping": {"Vlan100": "Vrf42"}}, + ) + entry = bgp_config["BGP_NEIGHBOR"]["Vrf42|2001:db8::20"] + assert "v6only" not in entry + assert "Vrf42|2001:db8::20|ipv6_unicast" in bgp_config["BGP_NEIGHBOR_AF"] + def test_duplicate_peer_ip_deduped(self, bgp_config, patch_bgp): - patch_bgp.peer_ipv4.return_value = "192.0.2.20" + patch_bgp.peer_ips.return_value = ("192.0.2.20", None) _call_bgp( bgp_config, netbox=object(), @@ -374,6 +477,22 @@ def test_duplicate_peer_ip_deduped(self, bgp_config, patch_bgp): ) assert list(bgp_config["BGP_NEIGHBOR"]) == ["default|192.0.2.20"] + def test_duplicate_peer_ipv6_deduped(self, bgp_config, patch_bgp): + patch_bgp.peer_ips.return_value = (None, "2001:db8::20") + _call_bgp( + bgp_config, + netbox=object(), + netbox_interfaces={ + "Ethernet0": _nbif("eth0"), + "Ethernet1": _nbif("eth1"), + }, + vlan_info=self._vlan_info( + {"eth0": "untagged", "eth1": "untagged"}, + addresses=["2001:db8::1/64"], + ), + ) + assert list(bgp_config["BGP_NEIGHBOR"]) == ["default|2001:db8::20"] + @pytest.mark.parametrize( "vlan_info", [ @@ -393,7 +512,7 @@ def test_duplicate_peer_ip_deduped(self, bgp_config, patch_bgp): ], ) def test_skipped_cases(self, bgp_config, patch_bgp, vlan_info): - patch_bgp.peer_ipv4.return_value = "192.0.2.20" + patch_bgp.peer_ips.return_value = ("192.0.2.20", None) _call_bgp( bgp_config, netbox=object(), @@ -403,7 +522,7 @@ def test_skipped_cases(self, bgp_config, patch_bgp, vlan_info): assert bgp_config["BGP_NEIGHBOR"] == {} def test_no_peer_ip_warns(self, bgp_config, patch_bgp, loguru_logs): - patch_bgp.peer_ipv4.return_value = None + patch_bgp.peer_ips.return_value = (None, None) _call_bgp( bgp_config, netbox=object(), @@ -418,7 +537,7 @@ def test_no_peer_ip_warns(self, bgp_config, patch_bgp, loguru_logs): ) def test_vlan_interface_name_used_for_vrf_lookup(self, bgp_config, patch_bgp): - patch_bgp.peer_ipv4.return_value = "192.0.2.20" + patch_bgp.peer_ips.return_value = ("192.0.2.20", None) _call_bgp( bgp_config, netbox=object(), @@ -433,7 +552,7 @@ def test_vlan_interface_name_used_for_vrf_lookup(self, bgp_config, patch_bgp): def test_member_not_in_netbox_interfaces_skipped( self, bgp_config, patch_bgp, loguru_logs ): - patch_bgp.peer_ipv4.return_value = "192.0.2.20" + patch_bgp.peer_ips.return_value = ("192.0.2.20", None) _call_bgp( bgp_config, netbox=object(), diff --git a/tests/unit/tasks/conductor/sonic/test_connections.py b/tests/unit/tasks/conductor/sonic/test_connections.py index e486bc1fd..2acf0fa9f 100644 --- a/tests/unit/tasks/conductor/sonic/test_connections.py +++ b/tests/unit/tasks/conductor/sonic/test_connections.py @@ -824,3 +824,162 @@ def test_load_vip_addresses_cache_overwrites_existing(mocker, reset_vip_cache): connections.load_vip_addresses_cache() assert connections._vip_addresses_cache == [] + + +# --------------------------------------------------------------------------- +# is_numbered_neighbor_address / get_connected_interface_ip{,v4,v6}_address(es) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address, ip_version, expected", + [ + ("192.0.2.1", 4, True), + ("192.0.2.1", 6, False), + ("2001:db8::1", 6, True), + ("2001:db8::1", 4, False), + # A prefix length is tolerated (parsed via ip_interface). + ("10.0.0.1/24", 4, True), + ("2001:db8::1/64", 6, True), + # Unparseable / non-address strings match no version. + ("not-an-ip", 4, False), + ("not-an-ip", 6, False), + # Link-local, loopback and multicast are not numbered neighbors. + ("169.254.0.1", 4, False), + ("fe80::1", 6, False), + ("127.0.0.1", 4, False), + ("::1", 6, False), + ("224.0.0.1", 4, False), + ("ff02::1", 6, False), + ], +) +def test_is_numbered_neighbor_address(address, ip_version, expected): + assert connections.is_numbered_neighbor_address(address, ip_version) is expected + + +def _netbox_with_direct_ips(mocker, addresses): + """Build a NetBox stub whose connected interface carries ``addresses``.""" + connected_iface = SimpleNamespace(id=2, name="peer-eth0") + interface = SimpleNamespace( + connected_endpoints=[connected_iface], + connected_endpoints_reachable=True, + ) + nb = mocker.Mock() + nb.dcim.interfaces.get.return_value = interface + nb.ipam.ip_addresses.filter.return_value = [ + SimpleNamespace(address=a) for a in addresses + ] + nb.ipam.fhrp_group_assignments.filter.return_value = [] + return nb + + +def test_get_connected_interface_ipv4_returns_direct_ipv4(mocker, reset_vip_cache): + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["2001:db8::1/64", "192.0.2.1/24"]) + assert ( + connections.get_connected_interface_ipv4_address(device, "Ethernet0", nb) + == "192.0.2.1" + ) + + +def test_get_connected_interface_ipv6_returns_direct_ipv6(mocker, reset_vip_cache): + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["192.0.2.1/24", "2001:db8::1/64"]) + assert ( + connections.get_connected_interface_ipv6_address(device, "Ethernet0", nb) + == "2001:db8::1" + ) + + +def test_get_connected_interface_ipv6_none_when_only_ipv4(mocker, reset_vip_cache): + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["192.0.2.1/24"]) + assert ( + connections.get_connected_interface_ipv6_address(device, "Ethernet0", nb) + is None + ) + + +def test_get_connected_interface_ipv6_returns_fhrp_vip(mocker, reset_vip_cache): + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["192.0.2.1/24"]) + group = SimpleNamespace(id=7, name="vrrp-7") + nb.ipam.fhrp_group_assignments.filter.return_value = [SimpleNamespace(group=group)] + connections._vip_addresses_cache = [ + SimpleNamespace( + address="2001:db8::ffff/64", + assigned_object_type="ipam.fhrpgroup", + assigned_object_id=7, + ), + SimpleNamespace( + address="192.0.2.254/24", + assigned_object_type="ipam.fhrpgroup", + assigned_object_id=7, + ), + ] + assert ( + connections.get_connected_interface_ipv6_address(device, "Ethernet0", nb) + == "2001:db8::ffff" + ) + + +def test_get_connected_interface_ipv6_direct_wins_over_vip(mocker, reset_vip_cache): + # A directly-assigned IPv6 wins over an FHRP IPv6 VIP even when the FHRP + # lookup actually runs (here it runs because IPv4 has no direct address). + # This precedence is deliberate -- see get_connected_interface_ip_addresses. + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["2001:db8::1/64"]) + group = SimpleNamespace(id=7, name="vrrp-7") + nb.ipam.fhrp_group_assignments.filter.return_value = [SimpleNamespace(group=group)] + connections._vip_addresses_cache = [ + SimpleNamespace( + address="2001:db8::ffff/64", + assigned_object_type="ipam.fhrpgroup", + assigned_object_id=7, + ), + SimpleNamespace( + address="192.0.2.254/24", + assigned_object_type="ipam.fhrpgroup", + assigned_object_id=7, + ), + ] + ipv4, ipv6 = connections.get_connected_interface_ip_addresses( + device, "Ethernet0", nb + ) + assert ipv6 == "2001:db8::1" # direct beats the 2001:db8::ffff VIP + assert ipv4 == "192.0.2.254" # no direct v4 -> FHRP VIP used + + +def test_get_connected_interface_ipv6_skips_link_local(mocker, reset_vip_cache): + # A link-local IPv6 is not a valid numbered neighbor; the global address + # on the same interface is used instead. + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["fe80::1/64", "2001:db8::1/64"]) + assert ( + connections.get_connected_interface_ipv6_address(device, "Ethernet0", nb) + == "2001:db8::1" + ) + + +def test_get_connected_interface_ipv6_link_local_only_returns_none( + mocker, reset_vip_cache +): + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["fe80::1/64"]) + assert ( + connections.get_connected_interface_ipv6_address(device, "Ethernet0", nb) + is None + ) + + +def test_get_connected_interface_ip_addresses_dual_stack_single_pass( + mocker, reset_vip_cache +): + # Both families resolved from direct addresses in one call; the FHRP + # lookup is skipped entirely once both are satisfied. + device = SimpleNamespace(id=1) + nb = _netbox_with_direct_ips(mocker, ["192.0.2.1/24", "2001:db8::1/64"]) + assert connections.get_connected_interface_ip_addresses( + device, "Ethernet0", nb + ) == ("192.0.2.1", "2001:db8::1") + nb.ipam.fhrp_group_assignments.filter.assert_not_called()