From 19b9007e122e8ad8582500bbfdb9d5459b058f04 Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Fri, 29 Mar 2024 13:14:21 +0100 Subject: [PATCH] Add ospfv3ipv6 converter. add unittest add integration tests --- .../feature_profile/sdwan/test_service.py | 41 +++++++ .../feature_profile/sdwan/service/ospfv3.py | 10 +- catalystwan/tests/test_feature_profile_api.py | 3 + .../converters/feature_template/ospfv3.py | 106 +++++++++++++++--- 4 files changed, 141 insertions(+), 19 deletions(-) diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py index 2c8970eb..8d34b409 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py +++ b/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() diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/ospfv3.py b/catalystwan/models/configuration/feature_profile/sdwan/service/ospfv3.py index d60c2f14..78274ab1 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/ospfv3.py +++ b/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 diff --git a/catalystwan/tests/test_feature_profile_api.py b/catalystwan/tests/test_feature_profile_api.py index df7dab34..292175d5 100644 --- a/catalystwan/tests/test_feature_profile_api.py +++ b/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 = [ diff --git a/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py b/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py index 47afd9c2..35fa39a6 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py +++ b/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,11 +42,21 @@ 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", @@ -50,19 +64,26 @@ class Ospfv3Ipv4TemplateSubconverter: "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) @@ -70,7 +91,7 @@ def configure_basic_ospf_v3_attributes(self, values: dict) -> None: 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