From 7f165b9aac82c283a1fc2bb42c82d5e61cec770a Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Wed, 3 Apr 2024 15:03:36 +0200 Subject: [PATCH] Add EIGRP model. Add unit test. Add integration test. --- .../feature_profile/sdwan/test_service.py | 27 +++++ .../feature_profile/sdwan/service/eigrp.py | 24 ++-- catalystwan/tests/test_feature_profile_api.py | 2 + .../converters/feature_template/eigrp.py | 107 +++++++++++++++++- 4 files changed, 148 insertions(+), 12 deletions(-) diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py index 8d34b409..78a74c5a 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py @@ -2,11 +2,17 @@ from catalystwan.api.configuration_groups.parcel import Global, as_global, as_variable from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels +from catalystwan.models.configuration.feature_profile.common import Prefix from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import ( AddressPool, LanVpnDhcpServerParcel, SubnetMask, ) +from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import ( + AddressFamily, + EigrpParcel, + SummaryAddress, +) from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ethernet import InterfaceEthernetParcel from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre, InterfaceGreParcel from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ipsec import ( @@ -104,6 +110,27 @@ def test_when_default_ospfv3_ipv6_expect_successful_post(self): # Assert assert parcel_id + def test_when_default_values_eigrp_parcel_expect_successful_post(self): + eigrp_parcel = EigrpParcel( + parcel_name="TestEigrpParcel", + parcel_description="Test Eigrp Parcel", + as_number=Global[int](value=1), + address_family=AddressFamily( + network=[ + SummaryAddress( + prefix=Prefix( + address=as_global("10.3.2.1"), + mask=as_global("255.255.255.0"), + ) + ) + ] + ), + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, eigrp_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/eigrp.py b/catalystwan/models/configuration/feature_profile/sdwan/service/eigrp.py index 838e44cd..cc59b0ab 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/eigrp.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/eigrp.py @@ -26,20 +26,22 @@ class KeychainDetails(BaseModel): - key_id: Union[Global[int], Variable, Default[None]] = Field(serialization_alias="keyId", validation_alias="keyId") - keystring: Union[Global[str], Variable, Default[None]] + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") + + key_id: Union[Global[int], Variable, Default[None]] = Field( + default=Default[None](value=None), serialization_alias="keyId", validation_alias="keyId" + ) + keystring: Union[Global[str], Variable, Default[None]] = Field(default=Default[None](value=None)) class EigrpAuthentication(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") - auth_type: Union[Global[EigrpAuthType], Variable, Default[None]] = Field( - serialization_alias="type", validation_alias="type" - ) + auth_type: Union[Global[EigrpAuthType], Variable, Default[None]] = Default[None](value=None) auth_key: Optional[Union[Global[str], Variable, Default[None]]] = Field( - serialization_alias="authKey", validation_alias="authKey" + serialization_alias="authKey", validation_alias="authKey", default=Default[None](value=None) ) - key: Optional[List[KeychainDetails]] = Field(serialization_alias="key", validation_alias="key") + key: Optional[List[KeychainDetails]] = None class TableMap(BaseModel): @@ -59,9 +61,9 @@ class IPv4StaticRoute(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") name: Union[Global[str], Variable] - shutdown: Optional[Union[Global[int], Variable, Default[bool]]] = Default[bool](value=False) - summary_address: Optional[List[SummaryAddress]] = Field( - serialization_alias="summaryAddress", validation_alias="summaryAddress" + shutdown: Optional[Union[Global[bool], Variable, Default[bool]]] = Default[bool](value=False) + summary_address: List[SummaryAddress] = Field( + serialization_alias="summaryAddress", validation_alias="summaryAddress", default_factory=list ) @@ -76,7 +78,7 @@ class AddressFamily(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") redistribute: Optional[List[RedistributeIntoEigrp]] = None - network: List[SummaryAddress] + network: List[SummaryAddress] = Field(min_length=1) class EigrpParcel(_ParcelBase): diff --git a/catalystwan/tests/test_feature_profile_api.py b/catalystwan/tests/test_feature_profile_api.py index 292175d5..aafac78e 100644 --- a/catalystwan/tests/test_feature_profile_api.py +++ b/catalystwan/tests/test_feature_profile_api.py @@ -19,6 +19,7 @@ LanVpnParcel, OspfParcel, ) +from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import EigrpParcel 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 @@ -108,6 +109,7 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path): OspfParcel: "routing/ospf", Ospfv3IPv4Parcel: "routing/ospfv3/ipv4", Ospfv3IPv6Parcel: "routing/ospfv3/ipv6", + EigrpParcel: "routing/eigrp", } service_interface_parcels = [ diff --git a/catalystwan/utils/config_migration/converters/feature_template/eigrp.py b/catalystwan/utils/config_migration/converters/feature_template/eigrp.py index c7010bed..2766f698 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/eigrp.py +++ b/catalystwan/utils/config_migration/converters/feature_template/eigrp.py @@ -1,6 +1,18 @@ from copy import deepcopy +from typing import List, Optional +from catalystwan.api.configuration_groups.parcel import Default, as_default, as_global, as_variable +from catalystwan.models.configuration.feature_profile.common import Prefix from catalystwan.models.configuration.feature_profile.sdwan.service import EigrpParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import ( + AddressFamily, + EigrpAuthentication, + IPv4StaticRoute, + RedistributeIntoEigrp, + RedistributeProtocol, + SummaryAddress, + TableMap, +) class EigrpTemplateConverter: @@ -8,9 +20,19 @@ class EigrpTemplateConverter: delete_keys = ("as_num",) + # Default values + lan_eigrp_auto_syst_id = "{{lan_eigrp_auto_syst_id}}" + lan_eigrp_addr_fami_netw_1_ip = "{{lan_eigrp_addr_fami_netw_1_ip}}" + lan_eigrp_addr_fami_netw_1_mask = "{{lan_eigrp_addr_fami_netw_1_mask}}" + def create_parcel(self, name: str, description: str, template_values: dict) -> EigrpParcel: + print(template_values) values = self.prepare_values(template_values) self.configure_as_number(values) + self.configure_address_family_interface(values) + self.configure_address_family(values) + self.configure_authentication(values) + self.configure_table_map(values) self.cleanup_keys(values) return EigrpParcel(parcel_name=name, parcel_description=description, **values) @@ -18,7 +40,90 @@ def prepare_values(self, template_values: dict) -> dict: return deepcopy(template_values)["eigrp"] def configure_as_number(self, values: dict) -> None: - values["as_number"] = values.pop("as_num") + values["as_number"] = values.pop("as_num", as_variable(self.lan_eigrp_auto_syst_id)) + + def configure_address_family(self, values: dict) -> None: + address_family = values.get("address_family", []) # feature template sends list instead of dict + if not address_family: + return + address_family = address_family[0] + values["address_family"] = AddressFamily( + redistribute=self._set_redistribute(address_family), + network=self._set_adress_family_addresses(address_family), + ) + + def _set_adress_family_addresses(self, values: dict) -> List[SummaryAddress]: + summary_address = values.get("network", []) + if not summary_address: + return [ + SummaryAddress( + prefix=Prefix( + address=as_variable(self.lan_eigrp_addr_fami_netw_1_ip), + mask=as_variable(self.lan_eigrp_addr_fami_netw_1_mask), + ) + ) + ] + return [self._set_summary_address(addr) for addr in summary_address] + + def _set_redistribute(self, values: dict) -> Optional[List[RedistributeIntoEigrp]]: + redistributes = values.get("topology", {}).get("base", {}).get("redistribute", []) + if not redistributes: + return None + return [ + RedistributeIntoEigrp( + protocol=as_global(redistribute["protocol"].value, RedistributeProtocol), + # route_policy=redistribute.get("route_policy", None), + # route polict is represented as a string in feature template and as UUID in model + ) + for redistribute in redistributes + ] + + def configure_address_family_interface(self, values: dict) -> None: + interfaces = values.get("af_interface", []) + if not interfaces: + return + interfaces_list = [] + for interface in interfaces: + interfaces_list.append( + IPv4StaticRoute( + name=interface["name"], + shutdown=interface.get("shutdown", as_default(False)), + summary_address=self._set_summary_addresses(interface), + ) + ) + values["af_interface"] = interfaces_list + + def _set_summary_addresses(self, values: dict) -> List[SummaryAddress]: + summary_address = values.get("summary_address", []) + return [self._set_summary_address(addr) for addr in summary_address] + + def _set_summary_address(self, addr: dict) -> SummaryAddress: + return SummaryAddress( + prefix=Prefix( + address=as_global(addr["prefix"].value.network.network_address), + mask=as_global(str(addr["prefix"].value.netmask)), + ) + ) + + def configure_authentication(self, values: dict) -> None: + auth = values.get("authentication", None) + if not auth: + return + values["authentication"] = EigrpAuthentication( + auth_type=auth.get("type", Default[None](value=None)), + auth_key=auth.get("key", Default[None](value=None)), + key=auth.get("keychain", {}).get("key", None), + # There should be more keys + ) + + def configure_table_map(self, values: dict) -> None: + table_map = values.get("table_map", None) + if not table_map: + return + values["table_map"] = TableMap( + # name=table_map.get("name", Default[None](value=None)), this should be Global[UUID] not Global[int] + filter=table_map.get("filter", as_default(False)), + ) def cleanup_keys(self, values: dict) -> None: for key in self.delete_keys: