diff --git a/python-avd/pyavd/_eos_designs/structured_config/network_services/router_adaptive_virtual_topology.py b/python-avd/pyavd/_eos_designs/structured_config/network_services/router_adaptive_virtual_topology.py index 726c181f0b7..72882a05c29 100644 --- a/python-avd/pyavd/_eos_designs/structured_config/network_services/router_adaptive_virtual_topology.py +++ b/python-avd/pyavd/_eos_designs/structured_config/network_services/router_adaptive_virtual_topology.py @@ -106,7 +106,7 @@ def _cv_pathfinder_profiles(self: AvdStructuredConfigNetworkServicesProtocol) -> for match in policy.get("matches", []): profile = { "name": match["avt_profile"], - "load_balance_policy": match["load_balance_policy"]["name"], + "load_balance_policy": match["load_balance_policy"].name, } if (internet_exit_policy_name := match["internet_exit_policy_name"]) is not None and internet_exit_policy_name in [ policy.name for policy, _connections in self._filtered_internet_exit_policies_and_connections @@ -123,7 +123,7 @@ def _cv_pathfinder_profiles(self: AvdStructuredConfigNetworkServicesProtocol) -> if (default_match := policy.get("default_match")) is not None: profile = { "name": default_match["avt_profile"], - "load_balance_policy": default_match["load_balance_policy"]["name"], + "load_balance_policy": default_match["load_balance_policy"].name, } if (internet_exit_policy_name := default_match["internet_exit_policy_name"]) is not None and internet_exit_policy_name in [ policy.name for policy, _connections in self._filtered_internet_exit_policies_and_connections diff --git a/python-avd/pyavd/_eos_designs/structured_config/network_services/router_path_selection.py b/python-avd/pyavd/_eos_designs/structured_config/network_services/router_path_selection.py index 2d4e047ac73..2b1a598324f 100644 --- a/python-avd/pyavd/_eos_designs/structured_config/network_services/router_path_selection.py +++ b/python-avd/pyavd/_eos_designs/structured_config/network_services/router_path_selection.py @@ -3,10 +3,11 @@ # that can be found in the LICENSE file. from __future__ import annotations -from functools import cached_property from typing import TYPE_CHECKING, Protocol -from pyavd._utils import append_if_not_duplicate, get, strip_empties_from_dict +from pyavd._eos_cli_config_gen.schema import EosCliConfigGen +from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor +from pyavd._utils import get if TYPE_CHECKING: from . import AvdStructuredConfigNetworkServicesProtocol @@ -19,70 +20,46 @@ class RouterPathSelectionMixin(Protocol): Class should only be used as Mixin to a AvdStructuredConfig class. """ - @cached_property - def router_path_selection(self: AvdStructuredConfigNetworkServicesProtocol) -> dict | None: + @structured_config_contributor + def router_path_selection(self: AvdStructuredConfigNetworkServicesProtocol) -> None: """Return structured config for router path-selection (DPS).""" if not self.shared_utils.is_wan_router: - return None + return - router_path_selection = { - "load_balance_policies": self._wan_load_balance_policies(), - } + self._set_wan_load_balance_policies() # When running CV Pathfinder, only load balance policies are configured # for AutoVPN, need also vrfs and policies. if self.inputs.wan_mode == "autovpn": - vrfs = [ - {"name": vrf.name, "path_selection_policy": f"{vrf.policy}-WITH-CP" if vrf.name == "default" else vrf.policy} for vrf in self._filtered_wan_vrfs - ] - - router_path_selection.update( - { - "policies": self._autovpn_policies(), - "vrfs": vrfs, - }, - ) + for vrf in self._filtered_wan_vrfs: + self.structured_config.router_path_selection.vrfs.append_new( + name=vrf.name, + path_selection_policy=f"{vrf.policy}-WITH-CP" if vrf.name == "default" else vrf.policy, + ) - return strip_empties_from_dict(router_path_selection) + self._autovpn_policies() - def _wan_load_balance_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> list: - """Return a list of load balance policies.""" - load_balance_policies = [] + def _set_wan_load_balance_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> None: + """Set list of load balance policies.""" for policy in self._filtered_wan_policies: for match in policy.get("matches", []): - append_if_not_duplicate( - list_of_dicts=load_balance_policies, - primary_key="name", - new_dict=match["load_balance_policy"], - context="Router Path-Selection load-balance policies.", - context_keys=["name"], - ) - if (default_match := policy.get("default_match")) is not None: - append_if_not_duplicate( - list_of_dicts=load_balance_policies, - primary_key="name", - new_dict=default_match["load_balance_policy"], - context="Router Path-Selection load-balance policies.", - context_keys=["name"], - ) + if "load_balance_policy" in match: + self.structured_config.router_path_selection.load_balance_policies.append(match["load_balance_policy"]) - return load_balance_policies + if (default_match := policy.get("default_match")) is not None and "load_balance_policy" in default_match: + self.structured_config.router_path_selection.load_balance_policies.append(default_match["load_balance_policy"]) - def _autovpn_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> list: - """Return a list of policies for AutoVPN.""" - policies = [] + def _autovpn_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> None: + """Set list of policies for AutoVPN.""" for policy in self._filtered_wan_policies: - autovpn_policy = {"name": policy["name"], "rules": []} + policy_item = EosCliConfigGen.RouterPathSelection.PoliciesItem() + policy_item.name = policy["name"] for index, match in enumerate(get(policy, "matches", default=[]), start=1): - autovpn_policy["rules"].append( - { - "id": 10 * index, - "application_profile": match["application_profile"], - "load_balance": match["load_balance_policy"]["name"], - }, + policy_item.rules.append_new( + id=10 * index, + application_profile=get(match, "application_profile"), + load_balance=match["load_balance_policy"].name if "load_balance_policy" in match else None, ) - if (default_match := policy.get("default_match")) is not None: - autovpn_policy["default_match"] = {"load_balance": default_match["load_balance_policy"]["name"]} - - policies.append(strip_empties_from_dict(autovpn_policy)) - return policies + if (default_match := policy.get("default_match")) is not None and "load_balance_policy" in default_match: + policy_item.default_match.load_balance = default_match["load_balance_policy"].name + self.structured_config.router_path_selection.policies.append(policy_item) diff --git a/python-avd/pyavd/_eos_designs/structured_config/network_services/utils_wan.py b/python-avd/pyavd/_eos_designs/structured_config/network_services/utils_wan.py index 6754a75bea9..78890146b1b 100644 --- a/python-avd/pyavd/_eos_designs/structured_config/network_services/utils_wan.py +++ b/python-avd/pyavd/_eos_designs/structured_config/network_services/utils_wan.py @@ -6,6 +6,7 @@ from functools import cached_property from typing import TYPE_CHECKING, Literal, Protocol +from pyavd._eos_cli_config_gen.schema import EosCliConfigGen from pyavd._eos_designs.schema import EosDesigns from pyavd._errors import AristaAvdError, AristaAvdInvalidInputsError, AristaAvdMissingVariableError from pyavd._utils import get, get_ip_from_ip_prefix @@ -71,6 +72,7 @@ def _filtered_wan_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> ) raise AristaAvdInvalidInputsError(msg) + # TODO: remove as_dict and fix the rest of utils_wan in another PR vrf_policy = self._wan_virtual_topologies_policies[vrf.policy]._as_dict() vrf_policy["profile_prefix"] = vrf.policy @@ -104,7 +106,7 @@ def _update_policy_match_statements(self: AvdStructuredConfigNetworkServicesProt if ( load_balance_policy := self._generate_wan_load_balance_policy( load_balance_policy_name, - control_plane_virtual_topology._as_dict(), + control_plane_virtual_topology, policy["name"], ) ) is None: @@ -134,7 +136,9 @@ def _update_policy_match_statements(self: AvdStructuredConfigNetworkServicesProt f"wan_virtual_topologies.policies[{policy['profile_prefix']}]." f"application_virtual_topologies[{application_virtual_topology['application_profile']}]" ) - load_balance_policy = self._generate_wan_load_balance_policy(load_balance_policy_name, application_virtual_topology, context_path) + # TODO: Refactor this once we get objects from policy object + avt_object = EosDesigns.WanVirtualTopologies.PoliciesItem.ApplicationVirtualTopologiesItem._from_dict(application_virtual_topology) + load_balance_policy = self._generate_wan_load_balance_policy(load_balance_policy_name, avt_object, context_path) if not load_balance_policy: # Empty load balance policy so skipping # TODO: Add "nodes" or similar under the profile and raise here @@ -171,24 +175,19 @@ def _update_policy_match_statements(self: AvdStructuredConfigNetworkServicesProt required=True, custom_error_msg=f"wan_virtual_topologies.policies[{policy['profile_prefix']}].default_virtual_toplogy.", ) + # TODO: this should be removed in a future refactor + avt_object = EosDesigns.WanVirtualTopologies.PoliciesItem.DefaultVirtualTopology._from_dict(default_virtual_topology) # Separating default_match as it is used differently default_match = None - if not get(default_virtual_topology, "drop_unmatched", default=False): - name = get( - default_virtual_topology, - "name", - default=self._default_profile_name(policy["profile_prefix"], "DEFAULT"), - ) + if not avt_object.drop_unmatched: + name = avt_object.name or self._default_profile_name(policy["profile_prefix"], "DEFAULT") context_path = f"wan_virtual_topologies.policies[{policy['profile_prefix']}].default_virtual_topology" # Verify that path_groups are set or raise - get( - default_virtual_topology, - "path_groups", - required=True, - custom_error_msg=f"Either 'drop_unmatched' or 'path_groups' must be set under '{context_path}'.", - ) + if not avt_object.path_groups: + msg = (f"Either 'drop_unmatched' or 'path_groups' must be set under '{context_path}'.",) + raise AristaAvdInvalidInputsError(msg) load_balance_policy_name = self.shared_utils.generate_lb_policy_name(name) - load_balance_policy = self._generate_wan_load_balance_policy(load_balance_policy_name, default_virtual_topology, context_path) + load_balance_policy = self._generate_wan_load_balance_policy(load_balance_policy_name, avt_object, context_path) if not load_balance_policy: msg = ( f"The `default_virtual_topology` path-groups configuration for `wan_virtual_topologies.policies[{policy['name']}]` produces " @@ -196,10 +195,8 @@ def _update_policy_match_statements(self: AvdStructuredConfigNetworkServicesProt "`default_virtual_topology` path-groups." ) raise AristaAvdError(msg) - application_profile = get(default_virtual_topology, "application_profile", default="default") - default_match = { - "application_profile": application_profile, + "application_profile": "default", "avt_profile": name, "internet_exit_policy_name": get(default_virtual_topology, "internet_exit.policy"), "traffic_class": get(default_virtual_topology, "traffic_class"), @@ -219,7 +216,14 @@ def _update_policy_match_statements(self: AvdStructuredConfigNetworkServicesProt policy["matches"] = matches policy["default_match"] = default_match - def _generate_wan_load_balance_policy(self: AvdStructuredConfigNetworkServicesProtocol, name: str, input_dict: dict, context_path: str) -> dict | None: + def _generate_wan_load_balance_policy( + self: AvdStructuredConfigNetworkServicesProtocol, + name: str, + input_topology: EosDesigns.WanVirtualTopologies.ControlPlaneVirtualTopology + | EosDesigns.WanVirtualTopologies.PoliciesItem.DefaultVirtualTopology + | EosDesigns.WanVirtualTopologies.PoliciesItem.ApplicationVirtualTopologiesItem, + context_path: str, + ) -> EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem | None: """ Generate and return a router path-selection load-balance policy. @@ -233,28 +237,24 @@ def _generate_wan_load_balance_policy(self: AvdStructuredConfigNetworkServicesPr input_dict (dict): The dictionary containing the list of path-groups and their preference. context_path (str): Key used for context for error messages. """ - wan_load_balance_policy = { - "name": name, - "path_groups": [], - **get(input_dict, "constraints", default={}), - } + wan_load_balance_policy = EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem( + name=name, + **input_topology.constraints._as_dict(), + ) if self.inputs.wan_mode == "cv-pathfinder": - wan_load_balance_policy["lowest_hop_count"] = get(input_dict, "lowest_hop_count") - - # An entry is composed of a list of path-groups in `names` and a `priority` - policy_entries = get(input_dict, "path_groups", []) + # TODO: Decide if the default value in schema for lowest_hop_count is correct. + wan_load_balance_policy.lowest_hop_count = input_topology._get_defined_attr("lowest_hop_count") or None # Using this flag while looping through all entries to keep track of any path group present on the remote host any_path_group_on_wan_ha_peer = self.shared_utils.wan_ha - for policy_entry in policy_entries: + for policy_entry in input_topology.path_groups: policy_entry_priority = None - if preference := get(policy_entry, "preference"): - policy_entry_priority = self._path_group_preference_to_eos_priority(preference, f"{context_path}[{policy_entry.get('names')}]") + if policy_entry.preference: + policy_entry_priority = self._path_group_preference_to_eos_priority(policy_entry.preference, f"{context_path}[{policy_entry.names}]") - entry_path_groups = policy_entry.get("names") - for path_group_name in entry_path_groups: + for path_group_name in policy_entry.names: if (priority := policy_entry_priority) is None: # No preference defined at the policy level, need to retrieve the default preference if path_group_name not in self.inputs.wan_path_groups: @@ -270,23 +270,21 @@ def _generate_wan_load_balance_policy(self: AvdStructuredConfigNetworkServicesPr if self.shared_utils.is_wan_client and path_group_name not in self.shared_utils.wan_local_path_group_names: continue - path_group = { - "name": path_group_name, - "priority": priority if priority != 1 else None, - } - - wan_load_balance_policy["path_groups"].append(path_group) + wan_load_balance_policy.path_groups.append_new( + name=path_group_name, + priority=priority if priority != 1 else None, + ) # Updating peer path-groups tracking - any_path_group_on_wan_ha_peer = any_path_group_on_wan_ha_peer and set(self.shared_utils.wan_ha_peer_path_group_names).union(set(entry_path_groups)) + any_path_group_on_wan_ha_peer = any_path_group_on_wan_ha_peer and set(self.shared_utils.wan_ha_peer_path_group_names).union(set(policy_entry.names)) - if len(wan_load_balance_policy["path_groups"]) == 0 and not any_path_group_on_wan_ha_peer: + if len(wan_load_balance_policy.path_groups) == 0 and not any_path_group_on_wan_ha_peer: # The policy is empty, and either the site is not using HA or no path-group in the policy is present on the HA peer return None if self.shared_utils.wan_ha or self.shared_utils.is_cv_pathfinder_server: # Adding HA path-group with priority 1 - wan_load_balance_policy["path_groups"].append({"name": self.inputs.wan_ha.lan_ha_path_group_name}) + wan_load_balance_policy.path_groups.append_new(name=self.inputs.wan_ha.lan_ha_path_group_name) return wan_load_balance_policy diff --git a/python-avd/pyavd/_eos_designs/structured_config/overlay/router_path_selection.py b/python-avd/pyavd/_eos_designs/structured_config/overlay/router_path_selection.py index 0b7a7ff7223..324ba7bdb07 100644 --- a/python-avd/pyavd/_eos_designs/structured_config/overlay/router_path_selection.py +++ b/python-avd/pyavd/_eos_designs/structured_config/overlay/router_path_selection.py @@ -6,8 +6,10 @@ from functools import cached_property from typing import TYPE_CHECKING, Protocol +from pyavd._eos_cli_config_gen.schema import EosCliConfigGen +from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor from pyavd._errors import AristaAvdError -from pyavd._utils import get, get_ip_from_ip_prefix, strip_empties_from_dict +from pyavd._utils import get, get_ip_from_ip_prefix if TYPE_CHECKING: from . import AvdStructuredConfigOverlayProtocol @@ -20,21 +22,17 @@ class RouterPathSelectionMixin(Protocol): Class should only be used as Mixin to a AvdStructuredConfig class. """ - @cached_property - def router_path_selection(self: AvdStructuredConfigOverlayProtocol) -> dict | None: - """Return structured config for router path-selection (DPS).""" + @structured_config_contributor + def router_path_selection(self: AvdStructuredConfigOverlayProtocol) -> None: + """Set structured config for router path-selection (DPS).""" if not self.shared_utils.is_wan_router: - return None + return - router_path_selection = { - "tcp_mss_ceiling": {"ipv4_segment_size": self.shared_utils.node_config.dps_mss_ipv4}, - "path_groups": self._get_path_groups(), - } + self.structured_config.router_path_selection.tcp_mss_ceiling.ipv4_segment_size = self.shared_utils.node_config.dps_mss_ipv4 + self._set_path_groups() if self.shared_utils.is_wan_server: - router_path_selection["peer_dynamic_source"] = "stun" - - return strip_empties_from_dict(router_path_selection) + self.structured_config.router_path_selection.peer_dynamic_source = "stun" @cached_property def _dp_ipsec_profile_name(self: AvdStructuredConfigOverlayProtocol) -> str: @@ -46,87 +44,75 @@ def _dp_ipsec_profile_name(self: AvdStructuredConfigOverlayProtocol) -> str: return self.inputs.wan_ipsec_profiles.data_plane.profile_name return self.inputs.wan_ipsec_profiles.control_plane.profile_name - def _get_path_groups(self: AvdStructuredConfigOverlayProtocol) -> list: - """Generate the required path-groups locally.""" - path_groups = [] - + def _set_path_groups(self: AvdStructuredConfigOverlayProtocol) -> None: + """Set the required path-groups locally.""" # Configure all path-groups on Pathfinders and AutoVPN RRs. Otherwise only configure the local path-groups path_groups_to_configure = self.inputs.wan_path_groups if self.shared_utils.is_wan_server else self.shared_utils.wan_local_path_groups for path_group in path_groups_to_configure: - pg_name = path_group.name - is_local_pg = pg_name in self.shared_utils.wan_local_path_group_names + is_local_pg = path_group.name in self.shared_utils.wan_local_path_group_names disable_dynamic_peer_ipsec = is_local_pg and not path_group.ipsec.dynamic_peers - - path_group_data = { - "name": pg_name, - "id": self._get_path_group_id(pg_name, path_group.id), - "local_interfaces": self._get_local_interfaces_for_path_group(pg_name), - "dynamic_peers": self._get_dynamic_peers(disable_dynamic_peer_ipsec), - "static_peers": self._get_static_peers_for_path_group(pg_name), - } + path_group_item = EosCliConfigGen.RouterPathSelection.PathGroupsItem() + path_group_item._update( + name=path_group.name, + id=self._get_path_group_id(path_group.name, path_group.id), + ) + self._set_local_interfaces_for_path_group(path_group.name, path_group_item) + self._set_dynamic_peers(disable_ipsec=disable_dynamic_peer_ipsec, path_group_item=path_group_item) + self._set_static_peers_for_path_group(path_group.name, path_group_item) if is_local_pg: # On pathfinder IPsec profile is not required for non local path_groups if path_group.ipsec.static_peers: - path_group_data["ipsec_profile"] = self.inputs.wan_ipsec_profiles.control_plane.profile_name + path_group_item.ipsec_profile = self.inputs.wan_ipsec_profiles.control_plane.profile_name # KeepAlive config is not required for non local path_groups if interval := path_group.dps_keepalive.interval: if interval == "auto": - path_group_data["keepalive"] = {"auto": True} + path_group_item.keepalive.auto = True else: if not (interval.isdigit() and 50 <= int(interval) <= 60000): msg = ( f"Invalid value '{interval}' for dps_keepalive.interval - " - f"should be either 'auto', or an integer[50-60000] for wan_path_groups[{pg_name}]" + f"should be either 'auto', or an integer[50-60000] for wan_path_groups[{path_group.name}]" ) raise AristaAvdError(msg) - path_group_data["keepalive"] = { - "interval": int(interval), - "failure_threshold": path_group.dps_keepalive.failure_threshold, - } + path_group_item.keepalive._update(interval=int(interval), failure_threshold=path_group.dps_keepalive.failure_threshold) - path_groups.append(path_group_data) + self.structured_config.router_path_selection.path_groups.append(path_group_item) if self.shared_utils.wan_ha or self.shared_utils.is_cv_pathfinder_server: - path_groups.append(self._generate_ha_path_group()) - - return path_groups + path_group_item = EosCliConfigGen.RouterPathSelection.PathGroupsItem() + self._generate_ha_path_group(path_group_item=path_group_item) + self.structured_config.router_path_selection.path_groups.append(path_group_item) - def _generate_ha_path_group(self: AvdStructuredConfigOverlayProtocol) -> dict: + def _generate_ha_path_group(self: AvdStructuredConfigOverlayProtocol, path_group_item: EosCliConfigGen.RouterPathSelection.PathGroupsItem) -> None: """Called only when self.shared_utils.wan_ha is True or on Pathfinders.""" - ha_path_group = { - "name": self.inputs.wan_ha.lan_ha_path_group_name, - "id": self._get_path_group_id(self.inputs.wan_ha.lan_ha_path_group_name), - "flow_assignment": "lan", - } + path_group_item._update( + name=self.inputs.wan_ha.lan_ha_path_group_name, + id=self._get_path_group_id(self.inputs.wan_ha.lan_ha_path_group_name), + flow_assignment="lan", + ) + if self.shared_utils.is_cv_pathfinder_server: - return ha_path_group + return if self.shared_utils.use_port_channel_for_direct_ha is True: - local_interfaces = [{"name": f"Port-Channel{self.shared_utils.wan_ha_port_channel_id}"}] + path_group_item.local_interfaces.append_new(name=f"Port-Channel{self.shared_utils.wan_ha_port_channel_id}") else: - local_interfaces = [{"name": interface} for interface in self.shared_utils.wan_ha_interfaces] - + for interface in self.shared_utils.wan_ha_interfaces: + path_group_item.local_interfaces.append_new(name=interface) # not a pathfinder device - ha_path_group.update( - { - # This should be the LAN interface over which a DPS tunnel is built - "local_interfaces": local_interfaces, - "static_peers": [ - { - "router_ip": self._wan_ha_peer_vtep_ip(), - "name": self.shared_utils.wan_ha_peer, - "ipv4_addresses": [get_ip_from_ip_prefix(ip_address) for ip_address in self.shared_utils.wan_ha_peer_ip_addresses], - }, - ], - }, + path_group_item.static_peers.append_new( + router_ip=self._wan_ha_peer_vtep_ip(), + name=self.shared_utils.wan_ha_peer, + ipv4_addresses=EosCliConfigGen.RouterPathSelection.PathGroupsItem.StaticPeersItem.Ipv4Addresses( + [get_ip_from_ip_prefix(ip_address) for ip_address in self.shared_utils.wan_ha_peer_ip_addresses] + ), ) - if self.shared_utils.wan_ha_ipsec: - ha_path_group["ipsec_profile"] = self._dp_ipsec_profile_name - return ha_path_group + if self.shared_utils.wan_ha_ipsec: + path_group_item.ipsec_profile = self._dp_ipsec_profile_name def _wan_ha_interfaces(self: AvdStructuredConfigOverlayProtocol) -> list: """Return list of interfaces for HA.""" @@ -149,44 +135,48 @@ def _get_path_group_id(self: AvdStructuredConfigOverlayProtocol, path_group_name return config_id return 500 - def _get_local_interfaces_for_path_group(self: AvdStructuredConfigOverlayProtocol, path_group_name: str) -> list: + def _set_local_interfaces_for_path_group( + self: AvdStructuredConfigOverlayProtocol, path_group_name: str, path_group_item: EosCliConfigGen.RouterPathSelection.PathGroupsItem + ) -> None: """ Generate the router_path_selection.local_interfaces list. For AUTOVPN clients, configure the stun server profiles as appropriate """ - local_interfaces = [] if path_group_name not in self.shared_utils.wan_local_path_groups: - return local_interfaces + return for interface in self.shared_utils.wan_local_path_groups[path_group_name]._internal_data.interfaces: - local_interface = {"name": get(interface, "name", required=True)} + local_interface_item = EosCliConfigGen.RouterPathSelection.PathGroupsItem.LocalInterfacesItem() + local_interface_item.name = get(interface, "name", required=True) if self.shared_utils.is_wan_client and self.shared_utils.should_connect_to_wan_rs([path_group_name]): stun_server_profiles = self._stun_server_profiles.get(path_group_name, []) if stun_server_profiles: - local_interface["stun"] = {"server_profiles": [profile.name for profile in stun_server_profiles]} + for profile in stun_server_profiles: + local_interface_item.stun.server_profiles.append_new(profile.name) - local_interfaces.append(local_interface) + path_group_item.local_interfaces.append(local_interface_item) - return local_interfaces - - def _get_dynamic_peers(self: AvdStructuredConfigOverlayProtocol, disable_ipsec: bool) -> dict | None: + def _set_dynamic_peers( + self: AvdStructuredConfigOverlayProtocol, disable_ipsec: bool, path_group_item: EosCliConfigGen.RouterPathSelection.PathGroupsItem + ) -> None: """TODO: support ip_local ?""" if not self.shared_utils.is_wan_client: - return None + return + + path_group_item.dynamic_peers.enabled = True - dynamic_peers = {"enabled": True} if disable_ipsec: - dynamic_peers["ipsec"] = False - return dynamic_peers + path_group_item.dynamic_peers.ipsec = False - def _get_static_peers_for_path_group(self: AvdStructuredConfigOverlayProtocol, path_group_name: str) -> list | None: + def _set_static_peers_for_path_group( + self: AvdStructuredConfigOverlayProtocol, path_group_name: str, path_group_item: EosCliConfigGen.RouterPathSelection.PathGroupsItem + ) -> None: """Retrieves the static peers to configure for a given path-group based on the connected nodes.""" if not self.shared_utils.is_wan_router: - return None + return - static_peers = [] for wan_route_server in self.shared_utils.filtered_wan_route_servers: if path_group_name not in wan_route_server.path_groups: continue @@ -195,12 +185,8 @@ def _get_static_peers_for_path_group(self: AvdStructuredConfigOverlayProtocol, p get_ip_from_ip_prefix(interface.public_ip) for interface in wan_route_server.path_groups[path_group_name].interfaces if interface.public_ip ] - static_peers.append( - { - "router_ip": wan_route_server.vtep_ip, - "name": wan_route_server.hostname, - "ipv4_addresses": ipv4_addresses, - }, + path_group_item.static_peers.append_new( + router_ip=wan_route_server.vtep_ip, + name=wan_route_server.hostname, + ipv4_addresses=EosCliConfigGen.RouterPathSelection.PathGroupsItem.StaticPeersItem.Ipv4Addresses(ipv4_addresses), ) - - return static_peers