From 7be65c6ed60c55bea4e85bba56518aaed88c24b2 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Tue, 5 May 2026 07:29:31 +0200 Subject: [PATCH] sonic/config_generator: use "up" for admin_status, add constants _add_bgp_configurations wrote "admin_status": "true" for every BGP_NEIGHBOR_AF row it generated: ipv4_unicast, ipv6_unicast and l2vpn_evpn keys for both interfaces and port channels, plus the VLAN peer ipv4_unicast row. The schema enforced by the new Pydantic validator (BgpNeighborAfListRow.admin_status) is Literal["up", "down"], so validation rejected every generated row with "Input should be 'up' or 'down'". Same fix pattern as #2260. Replace the literal "true" with "up" at all seven BGP call sites, and centralise the schema-allowed values as ADMIN_STATUS_UP / ADMIN_STATUS_DOWN constants in sonic/constants.py so future drift fails at import time rather than at validation. Switch the remaining admin_status producers (MGMT_INTERFACE, INTERFACE incl. breakout ports, VLAN, VLAN_INTERFACE, LOOPBACK, and the PORTCHANNEL row built in sonic/interface.py) to the same constants for consistency. AI-assisted: Claude Code Signed-off-by: Christian Berendt --- .../tasks/conductor/sonic/config_generator.py | 45 ++++++++++++------- osism/tasks/conductor/sonic/constants.py | 6 +++ osism/tasks/conductor/sonic/interface.py | 9 +++- 3 files changed, 41 insertions(+), 19 deletions(-) diff --git a/osism/tasks/conductor/sonic/config_generator.py b/osism/tasks/conductor/sonic/config_generator.py index 4da36140..8567cb6c 100644 --- a/osism/tasks/conductor/sonic/config_generator.py +++ b/osism/tasks/conductor/sonic/config_generator.py @@ -38,7 +38,12 @@ get_connected_interface_ipv4_address, ) from .cache import get_cached_device_interfaces -from .constants import BGP_AF_L2VPN_EVPN_TAG, DEFAULT_SONIC_ROLES +from .constants import ( + ADMIN_STATUS_DOWN, + ADMIN_STATUS_UP, + 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]] = {} @@ -297,7 +302,7 @@ def generate_sonic_config(device, hwsku, device_as_mapping=None, config_version= # Add management interface configuration if oob_ip_result: oob_ip, prefix_len = oob_ip_result - config["MGMT_INTERFACE"]["eth0"] = {"admin_status": "up"} + config["MGMT_INTERFACE"]["eth0"] = {"admin_status": ADMIN_STATUS_UP} config["MGMT_INTERFACE"][f"eth0|{oob_ip}/{prefix_len}"] = {} metalbox_ip = _get_metalbox_ip_for_device(device) # Write into the existing STATIC_ROUTE dict so any pre-existing @@ -376,12 +381,12 @@ def _add_port_configurations( # Set admin_status to "up" if port is connected or is a port channel member, otherwise "down" admin_status = ( - "up" + ADMIN_STATUS_UP if ( port_name in connected_interfaces or port_name in portchannel_info["member_mapping"] ) - else "down" + else ADMIN_STATUS_DOWN ) # Check if this port is a breakout port and adjust speed and lanes accordingly @@ -651,12 +656,12 @@ def _add_missing_breakout_ports( # Set admin_status based on connection or port channel membership admin_status = ( - "up" + ADMIN_STATUS_UP if ( port_name in connected_interfaces or port_name in portchannel_info["member_mapping"] ) - else "down" + else ADMIN_STATUS_DOWN ) # Generate correct alias (breakout port always gets subport notation) @@ -1023,12 +1028,14 @@ def get_vrf_for_interface(interface_name): vrf_name = get_vrf_for_interface(port_name) ipv4_key = f"{vrf_name}|{neighbor_id}|ipv4_unicast" - config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": "true"} + config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": ADMIN_STATUS_UP} # 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" - config["BGP_NEIGHBOR_AF"][ipv6_key] = {"admin_status": "true"} + config["BGP_NEIGHBOR_AF"][ipv6_key] = { + "admin_status": ADMIN_STATUS_UP + } logger.debug( f"Added BGP_NEIGHBOR_AF with ipv4_unicast and ipv6_unicast for interface {port_name} (no direct IPv4)" ) @@ -1057,7 +1064,9 @@ def get_vrf_for_interface(interface_name): if is_switch_connection or has_l2vpn_tag: l2vpn_key = f"{vrf_name}|{neighbor_id}|l2vpn_evpn" - config["BGP_NEIGHBOR_AF"][l2vpn_key] = {"admin_status": "true"} + config["BGP_NEIGHBOR_AF"][l2vpn_key] = { + "admin_status": ADMIN_STATUS_UP + } logger.debug( f"Added BGP_NEIGHBOR_AF l2vpn_evpn for interface {port_name} " f"(connected to {connected_device.name})" @@ -1090,8 +1099,8 @@ def get_vrf_for_interface(interface_name): ipv4_key = f"{vrf_name}|{neighbor_id}|ipv4_unicast" ipv6_key = f"{vrf_name}|{neighbor_id}|ipv6_unicast" - config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": "true"} - config["BGP_NEIGHBOR_AF"][ipv6_key] = {"admin_status": "true"} + config["BGP_NEIGHBOR_AF"][ipv4_key] = {"admin_status": ADMIN_STATUS_UP} + config["BGP_NEIGHBOR_AF"][ipv6_key] = {"admin_status": ADMIN_STATUS_UP} # Add l2vpn_evpn only for switch-to-switch connections (default VRF) if vrf_name == "default": @@ -1104,7 +1113,7 @@ def get_vrf_for_interface(interface_name): ) if is_switch_connection: l2vpn_key = f"{vrf_name}|{neighbor_id}|l2vpn_evpn" - config["BGP_NEIGHBOR_AF"][l2vpn_key] = {"admin_status": "true"} + config["BGP_NEIGHBOR_AF"][l2vpn_key] = {"admin_status": ADMIN_STATUS_UP} logger.debug( f"Added BGP_NEIGHBOR_AF l2vpn_evpn for port channel {pc_name} " f"(connected to switch {connected_device.name})" @@ -1358,7 +1367,9 @@ def get_vrf_for_interface(interface_name): # 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"} + config["BGP_NEIGHBOR_AF"][ipv4_af_key] = { + "admin_status": ADMIN_STATUS_UP + } logger.info( f"Added BGP neighbor configuration for VLAN {vid} using peer IP {peer_ipv4} " @@ -1695,7 +1706,7 @@ def _add_vlan_configuration(config, vlan_info, netbox_interfaces, device): members.append(sonic_interface_name) config["VLAN"][vlan_name] = { - "admin_status": "up", + "admin_status": ADMIN_STATUS_UP, "autostate": "enable", "members": members, "vlanid": str(vid), @@ -1721,7 +1732,7 @@ def _add_vlan_configuration(config, vlan_info, netbox_interfaces, device): vlan_name = f"Vlan{vid}" if "addresses" in interface_data and interface_data["addresses"]: # Add the VLAN interface - config["VLAN_INTERFACE"][vlan_name] = {"admin_status": "up"} + config["VLAN_INTERFACE"][vlan_name] = {"admin_status": ADMIN_STATUS_UP} # Add IP configuration for each address (IPv4 and IPv6) for address in interface_data["addresses"]: @@ -1733,7 +1744,7 @@ def _add_loopback_configuration(config, loopback_info): """Add Loopback configuration from NetBox.""" for loopback_name, loopback_data in loopback_info["loopbacks"].items(): # Add the Loopback interface - config["LOOPBACK"][loopback_name] = {"admin_status": "up"} + config["LOOPBACK"][loopback_name] = {"admin_status": ADMIN_STATUS_UP} # Add base Loopback interface entry config["LOOPBACK_INTERFACE"][loopback_name] = {} @@ -1934,7 +1945,7 @@ def _add_vrf_configuration(config, vrf_info, netbox_interfaces): # Create VLAN with the same ID as VNI vlan_name = f"Vlan{vni}" config["VLAN"][vlan_name] = { - "admin_status": "up", + "admin_status": ADMIN_STATUS_UP, "autostate": "enable", "vlanid": str(vni), } diff --git a/osism/tasks/conductor/sonic/constants.py b/osism/tasks/conductor/sonic/constants.py index 7a0c2bbd..3aceb5fb 100644 --- a/osism/tasks/conductor/sonic/constants.py +++ b/osism/tasks/conductor/sonic/constants.py @@ -2,6 +2,12 @@ """Constants and mappings for SONiC configuration.""" +# admin_status values accepted by the generated SONiC schema +# (Literal["up", "down"] on INTERFACE, PORTCHANNEL, VLAN, VLAN_INTERFACE, +# LOOPBACK, MGMT_INTERFACE, BGP_NEIGHBOR, BGP_NEIGHBOR_AF, ...). +ADMIN_STATUS_UP = "up" +ADMIN_STATUS_DOWN = "down" + # Tag to add AF L2VPN EVPN to BGP neighbor BGP_AF_L2VPN_EVPN_TAG = "bgp-af-l2vpn-evpn" diff --git a/osism/tasks/conductor/sonic/interface.py b/osism/tasks/conductor/sonic/interface.py index aab218f0..39d6f562 100644 --- a/osism/tasks/conductor/sonic/interface.py +++ b/osism/tasks/conductor/sonic/interface.py @@ -7,7 +7,12 @@ import re from loguru import logger -from .constants import PORT_TYPE_TO_SPEED_MAP, HIGH_SPEED_PORTS, PORT_CONFIG_PATH +from .constants import ( + ADMIN_STATUS_UP, + HIGH_SPEED_PORTS, + PORT_CONFIG_PATH, + PORT_TYPE_TO_SPEED_MAP, +) from .cache import get_cached_device_interfaces # Global cache for port configurations to avoid repeated file reads @@ -1052,7 +1057,7 @@ def detect_port_channels(device): if portchannel_name not in portchannels: portchannels[portchannel_name] = { "members": [], - "admin_status": "up", + "admin_status": ADMIN_STATUS_UP, "fast_rate": "true", "min_links": "1", "mtu": "9100",