Add ospfv3ipv6 converter. add unittest add integration tests

catalystwan/integration_tests/feature_profile/sdwan/test_service.py

@@ -17,6 +17,13 @@
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.svi import InterfaceSviParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.ospf import OspfParcel
+from catalystwan.models.configuration.feature_profile.sdwan.service.ospfv3 import (
+    Ospfv3InterfaceParametres,
+    Ospfv3IPv4Area,
+    Ospfv3IPv4Parcel,
+    Ospfv3IPv6Area,
+    Ospfv3IPv6Parcel,
+)
 
 
 class TestServiceFeatureProfileModels(TestFeatureProfileModels):
@@ -63,6 +70,40 @@ def test_when_default_values_ospf_parcel_expect_successful_post(self):
         # Assert
         assert parcel_id
 
+    def test_when_default_ospfv3_ipv4_expect_successful_post(self):
+        # Arrange
+        ospfv3ipv4_parcel = Ospfv3IPv4Parcel(
+            parcel_name="TestOspfv3ipv4",
+            parcel_description="Test Ospfv3ipv4 Parcel",
+            area=[
+                Ospfv3IPv4Area(
+                    area_number=as_global(5),
+                    interfaces=[Ospfv3InterfaceParametres(name=as_global("GigabitEthernet0/0/0"))],
+                )
+            ],
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, ospfv3ipv4_parcel).id
+        # Assert
+        assert parcel_id
+
+    def test_when_default_ospfv3_ipv6_expect_successful_post(self):
+        # Arrange
+        ospfv3ipv4_parcel = Ospfv3IPv6Parcel(
+            parcel_name="TestOspfv3ipv6",
+            parcel_description="Test Ospfv3ipv6 Parcel",
+            area=[
+                Ospfv3IPv6Area(
+                    area_number=as_global(7),
+                    interfaces=[Ospfv3InterfaceParametres(name=as_global("GigabitEthernet0/0/0"))],
+                )
+            ],
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, ospfv3ipv4_parcel).id
+        # Assert
+        assert parcel_id
+
     def tearDown(self) -> None:
         self.api.delete_profile(self.profile_uuid)
         self.session.close()

catalystwan/models/configuration/feature_profile/sdwan/service/ospfv3.py

@@ -1,6 +1,6 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
-from ipaddress import IPv4Address
+from ipaddress import IPv4Address, IPv6Interface
 from typing import List, Literal, Optional, Union
 from uuid import UUID
 
@@ -95,9 +95,9 @@ class Ospfv3InterfaceParametres(BaseModel):
 class SummaryRouteIPv6(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
-    network: Union[Global[str], Variable]
+    network: Union[Global[str], Global[IPv6Interface], Variable]
     no_advertise: Union[Global[bool], Variable, Default[bool]] = Field(
-        serialization_alias="noAdvertise", validation_alias="noAdvertise"
+        serialization_alias="noAdvertise", validation_alias="noAdvertise", default=Default[bool](value=False)
     )
     cost: Optional[Union[Global[int], Variable, Default[None]]] = None
 
@@ -160,7 +160,7 @@ class Ospfv3IPv4Area(BaseModel):
     area_type_config: Optional[Union[StubArea, NssaArea, NormalArea, DefaultArea]] = Field(
         serialization_alias="areaTypeConfig", validation_alias="areaTypeConfig", default=None
     )
-    interfaces: List[Ospfv3InterfaceParametres]
+    interfaces: List[Ospfv3InterfaceParametres] = Field(min_length=1)
     ranges: Optional[List[SummaryRoute]] = None
 
 
@@ -171,7 +171,7 @@ class Ospfv3IPv6Area(BaseModel):
     area_type_config: Optional[Union[StubArea, NssaArea, NormalArea, DefaultArea]] = Field(
         serialization_alias="areaTypeConfig", validation_alias="areaTypeConfig", default=None
     )
-    interfaces: List[Ospfv3InterfaceParametres]
+    interfaces: List[Ospfv3InterfaceParametres] = Field(min_length=1)
     ranges: Optional[List[SummaryRouteIPv6]] = None
 
 

catalystwan/tests/test_feature_profile_api.py

@@ -21,6 +21,7 @@
 )
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ipsec import IpsecAddress, IpsecTunnelMode
+from catalystwan.models.configuration.feature_profile.sdwan.service.ospfv3 import Ospfv3IPv4Parcel, Ospfv3IPv6Parcel
 from catalystwan.models.configuration.feature_profile.sdwan.system import (
     AAAParcel,
     BannerParcel,
@@ -105,6 +106,8 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path):
     AppqoeParcel: "appqoe",
     LanVpnParcel: "lan/vpn",
     OspfParcel: "routing/ospf",
+    Ospfv3IPv4Parcel: "routing/ospfv3/ipv4",
+    Ospfv3IPv6Parcel: "routing/ospfv3/ipv6",
 }
 
 service_interface_parcels = [

catalystwan/utils/config_migration/converters/feature_template/ospfv3.py

@@ -1,7 +1,7 @@
 from copy import deepcopy
-from typing import List, Optional, Tuple, Union
+from typing import List, Optional, Tuple, Type, Union, cast, get_args
 
-from catalystwan.api.configuration_groups.parcel import as_global
+from catalystwan.api.configuration_groups.parcel import Default, Global, as_global
 from catalystwan.models.configuration.feature_profile.common import Prefix
 from catalystwan.models.configuration.feature_profile.sdwan.service import Ospfv3IPv4Parcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.ospfv3 import (
@@ -15,12 +15,16 @@
     NssaArea,
     Ospfv3InterfaceParametres,
     Ospfv3IPv4Area,
+    Ospfv3IPv6Area,
     Ospfv3IPv6Parcel,
     RedistributedRoute,
+    RedistributedRouteIPv6,
     RedistributeProtocol,
+    RedistributeProtocolIPv6,
     SpfTimers,
     StubArea,
     SummaryRoute,
+    SummaryRouteIPv6,
 )
 from catalystwan.utils.config_migration.converters.exceptions import CatalystwanConverterCantConvertException
 
@@ -38,39 +42,56 @@ def create_parcel(
     ) -> Tuple[Ospfv3IPv4Parcel, Ospfv3IPv6Parcel]:
         if template_values.get("ospfv3") is None:
             raise CatalystwanConverterCantConvertException("Feature Template does not contain OSPFv3 configuration")
-        ospfv3ipv4 = Ospfv3Ipv4TemplateSubconverter().create_parcel(name, description, template_values)
-        return ospfv3ipv4  # type: ignore
+        ospfv3ipv4 = cast(
+            Ospfv3IPv4Parcel, Ospfv3Ipv4TemplateSubconverter().create_parcel(name, description, template_values)
+        )
+        ospfv3ipv6 = cast(
+            Ospfv3IPv6Parcel, Ospfv3Ipv6TemplateSubconverter().create_parcel(name, description, template_values)
+        )
+        return ospfv3ipv4, ospfv3ipv6
+
 
+class BaseOspfv3TemplateSubconverter:
+    key_address_family: str
+    key_distance: str
+    parcel_model: Union[Type[Ospfv3IPv4Parcel], Type[Ospfv3IPv6Parcel]]
+    area_model: Union[Type[Ospfv3IPv4Area], Type[Ospfv3IPv6Area]]
 
-class Ospfv3Ipv4TemplateSubconverter:
     delete_keys = (
         "default_information",
         "router_id",
         "table_map",
         "max_metric",
         "timers",
         "distance_ipv4",
+        "distance_ipv6",
         "auto_cost",
         "compatible",
     )
 
-    def create_parcel(self, name: str, description: str, template_values: dict) -> Ospfv3IPv4Parcel:
-        values = deepcopy(template_values).get("ospfv3", {}).get("address_family", {}).get("ipv4", {})
+    def create_parcel(
+        self, name: str, description: str, template_values: dict
+    ) -> Union[Ospfv3IPv4Parcel, Ospfv3IPv6Parcel]:
+        values = self.get_values(template_values)
         self.configure_basic_ospf_v3_attributes(values)
         self.configure_advanced_ospf_v3_attributes(values)
         self.configure_max_metric_router_lsa(values)
         self.configure_area(values)
         self.configure_redistribute(values)
         self.cleanup_keys(values)
-        return Ospfv3IPv4Parcel(parcel_name=name, parcel_description=description, **values)
+        return self.parcel_model(parcel_name=name, parcel_description=description, **values)
+
+    def get_values(self, template_values: dict) -> dict:
+        values = deepcopy(template_values).get("ospfv3", {}).get("address_family", {}).get(self.key_address_family, {})
+        return values
 
     def configure_basic_ospf_v3_attributes(self, values: dict) -> None:
         distance_configuration = self._get_distance_configuration(values)
         basic_values = self._get_basic_values(distance_configuration)
         values["basic"] = BasicOspfv3Attributes(router_id=values.get("router_id"), **basic_values)
 
     def _get_distance_configuration(self, values: dict) -> dict:
-        return values.get("distance_ipv4", {})
+        return values.get(self.key_distance, {})
 
     def _get_basic_values(self, values: dict) -> dict:
         return {
@@ -94,6 +115,8 @@ def _configure_originate(self, values: dict) -> Optional[DefaultOriginate]:
         originate = values.get("default_information", {}).get("originate")
         if originate is None:
             return None
+        if isinstance(originate, Global):
+            return DefaultOriginate(originate=originate)
         metric = originate.get("metric")
         if metric is not None:
             metric = as_global(str(metric.value))
@@ -135,11 +158,11 @@ def configure_area(self, values: dict) -> None:
         area_list = []
         for area_value in area:
             area_list.append(
-                Ospfv3IPv4Area(
+                self.area_model(
                     area_number=area_value.get("a_num"),
                     area_type_config=self._set_area_type_config(area_value),
                     interfaces=self._set_interfaces(area_value),
-                    ranges=self._set_range(area_value),
+                    ranges=self._set_range(area_value),  # type: ignore
                 )
             )
         values["area"] = area_list
@@ -161,9 +184,28 @@ def _set_interfaces(self, area_value: dict) -> List[Ospfv3InterfaceParametres]:
         for interface in interfaces:
             if authentication := interface.pop("authentication", None):
                 area_value["authentication_type"] = authentication.get("type")
+            if network := interface.pop("network", None):
+                interface["network_type"] = network
             interface_list.append(Ospfv3InterfaceParametres(**interface))
         return interface_list
 
+    def _set_range(self, area_value: dict) -> Optional[Union[List[SummaryRoute], List[SummaryRouteIPv6]]]:
+        raise NotImplementedError
+
+    def configure_redistribute(self, values: dict) -> None:
+        raise NotImplementedError
+
+    def cleanup_keys(self, values: dict) -> None:
+        for key in self.delete_keys:
+            values.pop(key, None)
+
+
+class Ospfv3Ipv4TemplateSubconverter(BaseOspfv3TemplateSubconverter):
+    key_address_family = "ipv4"
+    key_distance = "distance_ipv4"
+    parcel_model = Ospfv3IPv4Parcel
+    area_model = Ospfv3IPv4Area
+
     def _set_range(self, area_value: dict) -> Optional[List[SummaryRoute]]:
         ranges = area_value.get("range")
         if ranges is None:
@@ -196,6 +238,42 @@ def configure_redistribute(self, values: dict) -> None:
             )
         values["redistribute"] = redistribute_list
 
-    def cleanup_keys(self, values: dict) -> None:
-        for key in self.delete_keys:
-            values.pop(key, None)
+
+class Ospfv3Ipv6TemplateSubconverter(BaseOspfv3TemplateSubconverter):
+    key_address_family = "ipv6"
+    key_distance = "distance_ipv6"
+    parcel_model = Ospfv3IPv6Parcel
+    area_model = Ospfv3IPv6Area
+
+    def _set_range(self, area_value: dict) -> Optional[List[SummaryRouteIPv6]]:
+        ranges = area_value.get("range")
+        if ranges is None:
+            return None
+        range_list = []
+        for range_ in ranges:
+            print(range_)
+            range_list.append(
+                SummaryRouteIPv6(
+                    network=range_.get("address"),
+                    cost=range_.get("cost"),
+                    no_advertise=range_.get("no_advertise", Default[bool](value=False)),
+                )
+            )
+        return range_list
+
+    def configure_redistribute(self, values: dict) -> None:
+        redistributes = values.get("redistribute", [])
+        if redistributes == []:
+            return None
+        redistribute_list = []
+        for redistribute in redistributes:
+            print(redistribute)
+            if redistribute.get("protocol").value not in get_args(RedistributeProtocolIPv6):
+                continue
+            redistribute_list.append(
+                RedistributedRouteIPv6(
+                    protocol=as_global(redistribute.get("protocol").value, RedistributeProtocolIPv6),
+                    route_policy=redistribute.get("route_map"),
+                )
+            )
+        values["redistribute"] = redistribute_list