Merge branch 'devel' into ip-access-list

.github/workflows/pull-request-management.yml

@@ -1,7 +1,9 @@
 ---
 name: "Collection code testing"
 
-"on": pull_request
+"on":
+  pull_request:
+  merge_group:
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref }}

.pre-commit-config.yaml

@@ -75,7 +75,7 @@ repos:
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.9.4
+    rev: v0.9.6
     hooks:
       # Run the linter.
       - id: ruff

python-avd/pyavd/_eos_designs/structured_config/base/__init__.py

@@ -155,29 +155,22 @@ def static_routes(self) -> list | None:
             }
         ]
 
-    @cached_property
-    def ipv6_static_routes(self) -> list | None:
+    @structured_config_contributor
+    def ipv6_static_routes(self) -> None:
         """ipv6_static_routes set based on ipv6_mgmt_gateway, ipv6_mgmt_destination_networks and mgmt_interface_vrf."""
         if self.shared_utils.ipv6_mgmt_gateway is None or self.shared_utils.node_config.ipv6_mgmt_ip is None:
-            return None
+            return
 
         if self.inputs.ipv6_mgmt_destination_networks:
-            return [
-                {
-                    "vrf": self.inputs.mgmt_interface_vrf,
-                    "destination_address_prefix": mgmt_destination_network,
-                    "gateway": self.shared_utils.ipv6_mgmt_gateway,
-                }
-                for mgmt_destination_network in self.inputs.ipv6_mgmt_destination_networks
-            ]
+            for mgmt_destination_network in self.inputs.ipv6_mgmt_destination_networks:
+                self.structured_config.ipv6_static_routes.append_new(
+                    vrf=self.inputs.mgmt_interface_vrf, destination_address_prefix=mgmt_destination_network, gateway=self.shared_utils.ipv6_mgmt_gateway
+                )
+            return
 
-        return [
-            {
-                "vrf": self.inputs.mgmt_interface_vrf,
-                "destination_address_prefix": "::/0",
-                "gateway": self.shared_utils.ipv6_mgmt_gateway,
-            },
-        ]
+        self.structured_config.ipv6_static_routes.append_new(
+            vrf=self.inputs.mgmt_interface_vrf, destination_address_prefix="::/0", gateway=self.shared_utils.ipv6_mgmt_gateway
+        )
 
     @cached_property
     def service_routing_protocols_model(self) -> str:

python-avd/pyavd/_eos_designs/structured_config/inband_management/__init__.py

@@ -73,20 +73,14 @@ def static_routes(self) -> list | None:
             ),
         ]
 
-    @cached_property
-    def ipv6_static_routes(self) -> list | None:
+    @structured_config_contributor
+    def ipv6_static_routes(self) -> None:
         if not self.shared_utils.configure_inband_mgmt_ipv6 or self.shared_utils.inband_mgmt_ipv6_gateway is None:
-            return None
+            return
 
-        return [
-            strip_empties_from_dict(
-                {
-                    "destination_address_prefix": "::/0",
-                    "gateway": self.shared_utils.inband_mgmt_ipv6_gateway,
-                    "vrf": self.shared_utils.inband_mgmt_vrf,
-                },
-            ),
-        ]
+        self.structured_config.ipv6_static_routes.append_new(
+            destination_address_prefix="::/0", gateway=self.shared_utils.inband_mgmt_ipv6_gateway, vrf=self.shared_utils.inband_mgmt_vrf
+        )
 
     @structured_config_contributor
     def vrfs(self) -> None:

python-avd/pyavd/_eos_designs/structured_config/network_services/ipv6_static_routes.py

@@ -3,9 +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._eos_cli_config_gen.schema import EosCliConfigGen
+from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor
+
 if TYPE_CHECKING:
     from . import AvdStructuredConfigNetworkServicesProtocol
 
@@ -17,31 +19,21 @@ class Ipv6StaticRoutesMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @cached_property
-    def ipv6_static_routes(self: AvdStructuredConfigNetworkServicesProtocol) -> list[dict] | None:
+    @structured_config_contributor
+    def ipv6_static_routes(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
         """
-        Returns structured config for ipv6_static_routes.
+        Set the structured config for ipv6_static_routes.
 
         Consist of
         - ipv6 static_routes defined under the vrfs
         - static routes added automatically for VARPv6 with prefixes
         """
         if not self.shared_utils.network_services_l3:
-            return None
+            return
 
-        ipv6_static_routes = []
         for tenant in self.shared_utils.filtered_tenants:
             for vrf in tenant.vrfs:
                 for static_route in vrf.ipv6_static_routes:
-                    ipv6_static_route = static_route._as_dict()
-                    ipv6_static_route["vrf"] = vrf.name
-                    ipv6_static_route.pop("nodes", None)
-
-                    # Ignore duplicate items in case of duplicate VRF definitions across multiple tenants.
-                    if static_route not in ipv6_static_routes:
-                        ipv6_static_routes.append(ipv6_static_route)
-
-        if ipv6_static_routes:
-            return ipv6_static_routes
-
-        return None
+                    static_route_item = static_route._cast_as(EosCliConfigGen.Ipv6StaticRoutesItem, ignore_extra_keys=True)
+                    static_route_item.vrf = vrf.name
+                    self.structured_config.ipv6_static_routes.append_unique(static_route_item)

python-avd/pyavd/_eos_designs/structured_config/network_services/router_ospf.py

@@ -3,13 +3,16 @@
 # 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._eos_cli_config_gen.schema import EosCliConfigGen
+from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor
 from pyavd._errors import AristaAvdInvalidInputsError
-from pyavd._utils import append_if_not_duplicate, default
+from pyavd._utils import default
 
 if TYPE_CHECKING:
+    from pyavd._eos_designs.schema import EosDesigns
+
     from . import AvdStructuredConfigNetworkServicesProtocol
 
 
@@ -20,85 +23,90 @@ class RouterOspfMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @cached_property
-    def router_ospf(self: AvdStructuredConfigNetworkServicesProtocol) -> dict | None:
+    @structured_config_contributor
+    def router_ospf(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
         """
-        Return structured config for router_ospf.
+        Set structured config for router_ospf.
 
         If we have static_routes in default VRF and not EPVN, and underlay is OSPF
         Then add redistribute static to the underlay OSPF process.
         """
         if not self.shared_utils.network_services_l3:
-            return None
+            return
 
-        ospf_processes = []
         for tenant in self.shared_utils.filtered_tenants:
             for vrf in tenant.vrfs:
-                if not vrf.ospf.enabled:
-                    continue
-
-                if vrf.ospf.nodes and self.shared_utils.hostname not in vrf.ospf.nodes:
+                if not vrf.ospf.enabled or (vrf.ospf.nodes and self.shared_utils.hostname not in vrf.ospf.nodes):
                     continue
 
-                ospf_interfaces = []
-                for l3_interface in vrf.l3_interfaces:
-                    if l3_interface.ospf.enabled:
-                        for node_index, node in enumerate(l3_interface.nodes):
-                            if node != self.shared_utils.hostname:
-                                continue
-
-                            ospf_interfaces.append(l3_interface.interfaces[node_index])
-
-                for svi in vrf.svis:
-                    if svi.ospf.enabled:
-                        interface_name = f"Vlan{svi.id}"
-                        ospf_interfaces.append(interface_name)
-
                 process_id = default(vrf.ospf.process_id, vrf.vrf_id)
                 if not process_id:
                     msg = f"Missing or invalid 'ospf.process_id' or 'vrf_id' under vrf '{vrf.name}"
                     raise AristaAvdInvalidInputsError(msg)
-
-                process = {
-                    "id": process_id,
-                    "vrf": vrf.name if vrf.name != "default" else None,
-                    "passive_interface_default": True,
-                    "router_id": self.get_vrf_router_id(vrf, tenant, vrf.ospf.router_id),
-                    "no_passive_interfaces": ospf_interfaces,
-                    "bfd_enable": vrf.ospf.bfd or None,  # Historic behavior is to only output if True.
-                    "max_lsa": vrf.ospf.max_lsa,
-                }
-
-                process_redistribute = {}
-
-                if vrf.ospf.redistribute_bgp.enabled:
-                    process_redistribute["bgp"] = {"enabled": True}
-                    if route_map := vrf.ospf.redistribute_bgp.route_map:
-                        process_redistribute["bgp"]["route_map"] = route_map
-
-                if vrf.ospf.redistribute_connected.enabled:
-                    process_redistribute["connected"] = {"enabled": True}
-                    if route_map := vrf.ospf.redistribute_connected.route_map:
-                        process_redistribute["connected"]["route_map"] = route_map
-
-                process["redistribute"] = process_redistribute or None
-
-                # Strip None values from process before adding to list
-                process = {key: value for key, value in process.items() if value is not None}
-
-                append_if_not_duplicate(
-                    list_of_dicts=ospf_processes,
-                    primary_key="id",
-                    new_dict=process,
-                    context="OSPF Processes defined under network services",
-                    context_keys=["id"],
+                process = EosCliConfigGen.RouterOspf.ProcessIdsItem(
+                    id=process_id, passive_interface_default=True, max_lsa=vrf.ospf.max_lsa, router_id=self.get_vrf_router_id(vrf, tenant, vrf.ospf.router_id)
                 )
+                self._update_ospf_interface(process, vrf)
+
+                if vrf.name != "default":
+                    process.vrf = vrf.name
+                if vrf.ospf.bfd:
+                    process.bfd_enable = vrf.ospf.bfd
+                self._update_ospf_redistribute(process, vrf)
 
+                self.structured_config.router_ospf.process_ids.append(process)
         # If we have static_routes in default VRF and not EPVN, and underlay is OSPF
         # Then add redistribute static to the underlay OSPF process.
         if self._vrf_default_ipv4_static_routes["redistribute_in_underlay"] and self.shared_utils.underlay_routing_protocol in ["ospf", "ospf-ldp"]:
-            ospf_processes.append({"id": self.inputs.underlay_ospf_process_id, "redistribute": {"static": {"enabled": True}}})
-        if ospf_processes:
-            return {"process_ids": ospf_processes}
+            self.structured_config.router_ospf.process_ids.obtain(self.inputs.underlay_ospf_process_id).redistribute.static.enabled = True
 
-        return None
+    def _update_ospf_redistribute(
+        self: AvdStructuredConfigNetworkServicesProtocol,
+        process: EosCliConfigGen.RouterOspf.ProcessIdsItem,
+        vrf: EosDesigns._DynamicKeys.DynamicNetworkServicesItem.NetworkServicesItem.VrfsItem,
+    ) -> None:
+        """
+        Configures OSPF route redistribution settings for the given VRF.
+
+        This method enables redistribution of BGP and connected routes into OSPF,
+        setting the associated route maps if specified.
+
+        Args:
+            process: The OSPF process configuration object.
+            vrf: The VRF object containing OSPF redistribution settings.
+        """
+        if vrf.ospf.redistribute_bgp.enabled:
+            process.redistribute.bgp.enabled = True
+            if route_map := vrf.ospf.redistribute_bgp.route_map:
+                process.redistribute.bgp.route_map = route_map
+
+        if vrf.ospf.redistribute_connected.enabled:
+            process.redistribute.connected.enabled = True
+            if route_map := vrf.ospf.redistribute_connected.route_map:
+                process.redistribute.connected.route_map = route_map
+
+    def _update_ospf_interface(
+        self: AvdStructuredConfigNetworkServicesProtocol,
+        process: EosCliConfigGen.RouterOspf.ProcessIdsItem,
+        vrf: EosDesigns._DynamicKeys.DynamicNetworkServicesItem.NetworkServicesItem.VrfsItem,
+    ) -> None:
+        """
+        Populates the list of OSPF-enabled interfaces for the given VRF.
+
+        This method iterates through L3 interfaces and SVIs, adding those that have OSPF enabled.
+
+        Args:
+            process: The OSPF process configuration object.
+            vrf: The VRF object containing interface OSPF settings.
+        """
+        for l3_interface in vrf.l3_interfaces:
+            if l3_interface.ospf.enabled:
+                for node_index, node in enumerate(l3_interface.nodes):
+                    if node != self.shared_utils.hostname:
+                        continue
+                    process.no_passive_interfaces.append(l3_interface.interfaces[node_index])
+
+        for svi in vrf.svis:
+            if svi.ospf.enabled:
+                interface_name = f"Vlan{svi.id}"
+                process.no_passive_interfaces.append(interface_name)