From 5b0b70f9f1a7f68b276593da5b8b9cb6f61c7648 Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Thu, 21 Mar 2024 16:05:11 +0100 Subject: [PATCH] Add Interface GRE model. Add unittests. Add integration tests. Change feature profile integration test structure. Add more Castable literals to the normalizer. Change name factory method to parcel_factory. Change VPN model type to lan/vpn. --- catalystwan/api/feature_profile_api.py | 14 ++- .../feature_profile/sdwan/service.py | 8 +- .../feature_profile/sdwan/base.py | 16 +++ .../sdwan/service/test_models.py | 59 ---------- .../{other/test_models.py => test_other.py} | 24 ++-- .../feature_profile/sdwan/test_service.py | 78 ++++++++++++ .../{system/test_models.py => test_system.py} | 46 +++----- .../feature_profile/sdwan/service/__init__.py | 7 +- .../feature_profile/sdwan/service/lan/gre.py | 51 ++++---- .../feature_profile/sdwan/service/lan/vpn.py | 2 +- catalystwan/tests/test_feature_profile_api.py | 59 +++++----- .../converters/feature_template/__init__.py | 2 +- .../converters/feature_template/gre.py | 111 ++++++++++++++++++ .../converters/feature_template/normalizer.py | 14 +++ .../{factory_method.py => parcel_factory.py} | 2 + catalystwan/workflows/config_migration.py | 22 +++- 16 files changed, 340 insertions(+), 175 deletions(-) create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/base.py delete mode 100644 catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py rename catalystwan/integration_tests/feature_profile/sdwan/{other/test_models.py => test_other.py} (57%) create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/test_service.py rename catalystwan/integration_tests/feature_profile/sdwan/{system/test_models.py => test_system.py} (71%) create mode 100644 catalystwan/utils/config_migration/converters/feature_template/gre.py rename catalystwan/utils/config_migration/converters/feature_template/{factory_method.py => parcel_factory.py} (97%) diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index 810af09e8..a9c7860f3 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/catalystwan/api/feature_profile_api.py @@ -12,7 +12,7 @@ from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.url import URLParcel -from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel, AnyTopLevelServiceParcel +from catalystwan.models.configuration.feature_profile.sdwan.service import AnyLanVpnInterfaceParcel, AnyServiceParcel from catalystwan.typed_list import DataSequence if TYPE_CHECKING: @@ -232,13 +232,17 @@ def delete_profile(self, profile_id: UUID) -> None: """ self.endpoint.delete_sdwan_service_feature_profile(profile_id) - def create_parcel(self, profile_id: UUID, payload: AnyServiceParcel) -> ParcelCreationResponse: + def create_parcel( + self, profile_uuid: UUID, payload: AnyServiceParcel, vpn_uuid: Optional[UUID] = None + ) -> ParcelCreationResponse: """ Create Service Parcel for selected profile_id based on payload type """ - if type(payload) in get_args(AnyTopLevelServiceParcel)[0].__args__: - return self.endpoint.create_top_level_service_parcel(profile_id, payload._get_parcel_type(), payload) - return self.endpoint.create_lan_vpn_service_parcel(profile_id, payload) + if type(payload) in get_args(AnyLanVpnInterfaceParcel)[0].__args__: + return self.endpoint.create_lan_vpn_interface_parcel( + profile_uuid, vpn_uuid, payload._get_parcel_type(), payload + ) + return self.endpoint.create_service_parcel(profile_uuid, payload._get_parcel_type(), payload) class SystemFeatureProfileAPI: diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py index 2d57e2256..d6b8db64b 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py @@ -16,7 +16,6 @@ AnyLanVpnInterfaceParcel, AnyTopLevelServiceParcel, ) -from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel from catalystwan.typed_list import DataSequence @@ -42,16 +41,11 @@ def delete_sdwan_service_feature_profile(self, profile_uuid: UUID) -> None: @versions(supported_versions=(">=20.9"), raises=False) @post("/v1/feature-profile/sdwan/service/{profile_uuid}/{parcel_type}") - def create_top_level_service_parcel( + def create_service_parcel( self, profile_uuid: UUID, parcel_type: str, payload: AnyTopLevelServiceParcel ) -> ParcelCreationResponse: ... - @versions(supported_versions=(">=20.9"), raises=False) - @post("/v1/feature-profile/sdwan/service/{profile_uuid}/lan/vpn/") - def create_lan_vpn_service_parcel(self, profile_uuid: UUID, payload: LanVpnParcel) -> ParcelCreationResponse: - ... - @versions(supported_versions=(">=20.9"), raises=False) @post("/v1/feature-profile/sdwan/service/{profile_uuid}/lan/vpn/{vpn_uuid}/interface/{parcel_type}") def create_lan_vpn_interface_parcel( diff --git a/catalystwan/integration_tests/feature_profile/sdwan/base.py b/catalystwan/integration_tests/feature_profile/sdwan/base.py new file mode 100644 index 000000000..9e0430152 --- /dev/null +++ b/catalystwan/integration_tests/feature_profile/sdwan/base.py @@ -0,0 +1,16 @@ +import os +import unittest +from typing import cast + +from catalystwan.session import create_manager_session + + +class TestFeatureProfileModels(unittest.TestCase): + def setUp(self) -> None: + # TODO: Add those params to PyTest + self.session = create_manager_session( + url=cast(str, os.environ.get("TEST_VMANAGE_URL", "localhost")), + port=cast(int, int(os.environ.get("TEST_VMANAGE_PORT", 443))), # type: ignore + username=cast(str, os.environ.get("TEST_VMANAGE_USERNAME", "admin")), + password=cast(str, os.environ.get("TEST_VMANAGE_PASSWORD", "admin")), + ) diff --git a/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py b/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py deleted file mode 100644 index 0ffe4323c..000000000 --- a/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py +++ /dev/null @@ -1,59 +0,0 @@ -import os -import unittest -from ipaddress import IPv4Address -from typing import cast - -from catalystwan.api.configuration_groups.parcel import Global -from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import ( - AddressPool, - LanVpnDhcpServerParcel, - SubnetMask, -) -from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel -from catalystwan.session import create_manager_session - - -class TestServiceFeatureProfileModels(unittest.TestCase): - def setUp(self) -> None: - self.session = create_manager_session( - url=cast(str, os.environ.get("TEST_VMANAGE_URL")), - port=cast(int, int(os.environ.get("TEST_VMANAGE_PORT"))), # type: ignore - username=cast(str, os.environ.get("TEST_VMANAGE_USERNAME")), - password=cast(str, os.environ.get("TEST_VMANAGE_PASSWORD")), - ) - self.profile_id = self.session.api.sdwan_feature_profiles.service.create_profile( - "TestProfile", "Description" - ).id - - def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): - # Arrange - dhcp_server_parcel = LanVpnDhcpServerParcel( - parcel_name="DhcpServerDefault", - parcel_description="Dhcp Server Parcel", - address_pool=AddressPool( - network_address=Global[IPv4Address](value=IPv4Address("10.0.0.2")), - subnet_mask=Global[SubnetMask](value="255.255.255.255"), - ), - ) - # Act - parcel_id = self.session.api.sdwan_feature_profiles.service.create_parcel( - self.profile_id, dhcp_server_parcel - ).id - # Assert - assert parcel_id - - def test_when_default_values_service_vpn_parcel_expect_successful_post(self): - # Arrange - vpn_parcel = LanVpnParcel( - parcel_name="TestVpnParcel", - parcel_description="Test Vpn Parcel", - vpn_id=Global[int](value=2), - ) - # Act - parcel_id = self.session.api.sdwan_feature_profiles.service.create_parcel(self.profile_id, vpn_parcel).id - # Assert - assert parcel_id - - def tearDown(self) -> None: - self.session.api.sdwan_feature_profiles.service.delete_profile(self.profile_id) - self.session.close() diff --git a/catalystwan/integration_tests/feature_profile/sdwan/other/test_models.py b/catalystwan/integration_tests/feature_profile/sdwan/test_other.py similarity index 57% rename from catalystwan/integration_tests/feature_profile/sdwan/other/test_models.py rename to catalystwan/integration_tests/feature_profile/sdwan/test_other.py index cde5f51c5..339ff58a4 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/other/test_models.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_other.py @@ -1,22 +1,14 @@ -import os -import unittest -from typing import cast - from catalystwan.api.configuration_groups.parcel import Global, as_global +from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels from catalystwan.models.configuration.feature_profile.sdwan.other import ThousandEyesParcel, UcseParcel from catalystwan.models.configuration.feature_profile.sdwan.other.ucse import AccessPort, Imc, LomType, SharedLom -from catalystwan.session import create_manager_session -class TestSystemOtherProfileModels(unittest.TestCase): +class TestSystemOtherProfileModels(TestFeatureProfileModels): def setUp(self) -> None: - self.session = create_manager_session( - url=cast(str, os.environ.get("TEST_VMANAGE_URL")), - port=cast(int, int(os.environ.get("TEST_VMANAGE_PORT"))), # type: ignore - username=cast(str, os.environ.get("TEST_VMANAGE_USERNAME")), - password=cast(str, os.environ.get("TEST_VMANAGE_PASSWORD")), - ) - self.profile_id = self.session.api.sdwan_feature_profiles.other.create_profile("TestProfile", "Description").id + super().setUp() + self.api = self.session.api.sdwan_feature_profiles.other + self.profile_id = self.api.create_profile("TestProfile", "Description").id def test_when_default_values_thousandeyes_parcel_expect_successful_post(self): # Arrange @@ -25,7 +17,7 @@ def test_when_default_values_thousandeyes_parcel_expect_successful_post(self): parcel_description="ThousandEyes Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.other.create_parcel(self.profile_id, te_parcel).id + parcel_id = self.api.create_parcel(te_parcel, self.profile_id).id # Assert assert parcel_id @@ -45,10 +37,10 @@ def test_when_default_values_ucse_parcel_expect_successful_post(self): ), ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.other.create_parcel(self.profile_id, ucse_parcel).id + parcel_id = self.api.create_parcel(ucse_parcel, self.profile_id).id # Assert assert parcel_id def tearDown(self) -> None: - self.session.api.sdwan_feature_profiles.other.delete_profile(self.profile_id) + self.api.delete_profile(self.profile_id) self.session.close() diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py new file mode 100644 index 000000000..682acc54b --- /dev/null +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py @@ -0,0 +1,78 @@ +from ipaddress import IPv4Address + +from catalystwan.api.configuration_groups.parcel import Global, as_global +from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels +from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import ( + AddressPool, + LanVpnDhcpServerParcel, + SubnetMask, +) +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre, InterfaceGreParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel + + +class TestServiceFeatureProfileModels(TestFeatureProfileModels): + def setUp(self) -> None: + super().setUp() + self.api = self.session.api.sdwan_feature_profiles.service + self.profile_uuid = self.api.create_profile("TestProfileService", "Description").id + + def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): + # Arrange + dhcp_server_parcel = LanVpnDhcpServerParcel( + parcel_name="DhcpServerDefault", + parcel_description="Dhcp Server Parcel", + address_pool=AddressPool( + network_address=Global[IPv4Address](value=IPv4Address("10.0.0.2")), + subnet_mask=Global[SubnetMask](value="255.255.255.255"), + ), + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, dhcp_server_parcel).id + # Assert + assert parcel_id + + def test_when_default_values_service_vpn_parcel_expect_successful_post(self): + # Arrange + vpn_parcel = LanVpnParcel( + parcel_name="TestVpnParcel", + parcel_description="Test Vpn Parcel", + vpn_id=Global[int](value=2), + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, vpn_parcel).id + # Assert + assert parcel_id + + def tearDown(self) -> None: + self.api.delete_profile(self.profile_uuid) + self.session.close() + + +class TestServiceFeatureProfileVPNInterfaceModels(TestFeatureProfileModels): + def setUp(self) -> None: + super().setUp() + self.api = self.session.api.sdwan_feature_profiles.service + self.profile_uuid = self.api.create_profile("TestProfileService", "Description").id + self.vpn_parcel_uuid = self.api.create_parcel( + self.profile_uuid, + LanVpnParcel( + parcel_name="TestVpnParcel", parcel_description="Test Vpn Parcel", vpn_id=Global[int](value=2) + ), + ).id + + def test_when_default_values_gre_parcel_expect_successful_post(self): + # Arrange + gre_parcel = InterfaceGreParcel( + parcel_name="TestGreParcel", + parcel_description="Test Gre Parcel", + basic=BasicGre(if_name=as_global("gre1"), tunnel_destination=as_global(IPv4Address("4.4.4.4"))), + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, gre_parcel, self.vpn_parcel_uuid).id + # Assert + assert parcel_id + + def tearDown(self) -> None: + self.api.delete_profile(self.profile_uuid) + self.session.close() diff --git a/catalystwan/integration_tests/feature_profile/sdwan/system/test_models.py b/catalystwan/integration_tests/feature_profile/sdwan/test_system.py similarity index 71% rename from catalystwan/integration_tests/feature_profile/sdwan/system/test_models.py rename to catalystwan/integration_tests/feature_profile/sdwan/test_system.py index 814c88bf2..a3ebdadff 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/system/test_models.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_system.py @@ -1,7 +1,4 @@ -import os -import unittest -from typing import cast - +from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels from catalystwan.models.configuration.feature_profile.sdwan.system import ( BannerParcel, BasicParcel, @@ -14,18 +11,13 @@ SecurityParcel, SNMPParcel, ) -from catalystwan.session import create_manager_session -class TestSystemFeatureProfileModels(unittest.TestCase): +class TestSystemFeatureProfileModels(TestFeatureProfileModels): def setUp(self) -> None: - self.session = create_manager_session( - url=cast(str, os.environ.get("TEST_VMANAGE_URL")), - port=cast(int, int(os.environ.get("TEST_VMANAGE_PORT"))), # type: ignore - username=cast(str, os.environ.get("TEST_VMANAGE_USERNAME")), - password=cast(str, os.environ.get("TEST_VMANAGE_PASSWORD")), - ) - self.profile_id = self.session.api.sdwan_feature_profiles.system.create_profile("TestProfile", "Description").id + super().setUp() + self.api = self.session.api.sdwan_feature_profiles.system + self.profile_id = self.api.create_profile("TestProfile", "Description").id def test_when_default_values_banner_parcel_expect_successful_post(self): # Arrange @@ -34,7 +26,7 @@ def test_when_default_values_banner_parcel_expect_successful_post(self): parcel_description="Banner Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, banner_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, banner_parcel).id # Assert assert parcel_id @@ -47,7 +39,7 @@ def test_when_fully_specified_banner_parcel_expect_successful_post(self): banner_parcel.add_login("Login") banner_parcel.add_motd("Hello! Welcome to the network!") # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, banner_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, banner_parcel).id # Assert assert parcel_id @@ -58,7 +50,7 @@ def test_when_default_values_logging_parcel_expect_successful_post(self): parcel_description="Logging Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, logging_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, logging_parcel).id # Assert assert parcel_id @@ -102,7 +94,7 @@ def test_when_fully_specified_logging_parcel_expect_successful_post(self): profile_properties="TLSProfile", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, logging_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, logging_parcel).id # Assert assert parcel_id @@ -113,7 +105,7 @@ def test_when_default_values_bfd_parcel_expect_successful_post(self): parcel_description="BFD Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, bfd_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, bfd_parcel).id # Assert assert parcel_id @@ -131,7 +123,7 @@ def test_when_fully_specified_bfd_parcel_expect_successful_post(self): bfd_parcel.add_color(color="biz-internet") bfd_parcel.add_color(color="public-internet") # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, bfd_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, bfd_parcel).id # Assert assert parcel_id @@ -142,7 +134,7 @@ def test_when_default_values_basic_parcel_expect_successful_post(self): parcel_description="Basic Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, basic_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, basic_parcel).id # Assert assert parcel_id @@ -153,7 +145,7 @@ def test_when_default_values_security_parcel_expect_successful_post(self): parcel_description="Security Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, security_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, security_parcel).id # Assert assert parcel_id @@ -164,7 +156,7 @@ def test_when_default_values_ntp_parcel_expect_successful_post(self): parcel_description="NTP Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, ntp_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, ntp_parcel).id # Assert assert parcel_id @@ -175,7 +167,7 @@ def test_when_default_values_global_parcel_expect_successful_post(self): parcel_description="Global Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, global_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, global_parcel).id # Assert assert parcel_id @@ -186,7 +178,7 @@ def test_when_default_values_mrf_parcel_expect_successful_post(self): parcel_description="MRF Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, mrf_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, mrf_parcel).id # Assert assert parcel_id @@ -197,7 +189,7 @@ def test_when_default_values_snmp_parcel_expect_successful_post(self): parcel_description="SNMP Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, snmp_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, snmp_parcel).id # Assert assert parcel_id @@ -208,10 +200,10 @@ def test_when_default_values_omp_parcel_expect_successful_post(self): parcel_description="OMP Parcel", ) # Act - parcel_id = self.session.api.sdwan_feature_profiles.system.create_parcel(self.profile_id, omp_parcel).id + parcel_id = self.api.create_parcel(self.profile_id, omp_parcel).id # Assert assert parcel_id def tearDown(self) -> None: - self.session.api.sdwan_feature_profiles.system.delete_profile(self.profile_id) + self.api.delete_profile(self.profile_id) self.session.close() diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py b/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py index 49911852c..44eaa1200 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py @@ -6,7 +6,7 @@ from .appqoe import AppqoeParcel from .dhcp_server import LanVpnDhcpServerParcel from .lan.ethernet import InterfaceEthernetData -from .lan.gre import InterfaceGreData +from .lan.gre import InterfaceGreParcel from .lan.ipsec import InterfaceIpsecData from .lan.svi import InterfaceSviData from .lan.vpn import LanVpnParcel @@ -15,6 +15,7 @@ Union[ LanVpnDhcpServerParcel, AppqoeParcel, + LanVpnParcel, # TrackerGroupData, # WirelessLanData, # SwitchportData @@ -25,7 +26,7 @@ AnyLanVpnInterfaceParcel = Annotated[ Union[ InterfaceEthernetData, - InterfaceGreData, + InterfaceGreParcel, InterfaceIpsecData, InterfaceSviData, ], @@ -33,7 +34,7 @@ ] AnyServiceParcel = Annotated[ - Union[AnyTopLevelServiceParcel, LanVpnParcel, AnyLanVpnInterfaceParcel], + Union[AnyTopLevelServiceParcel, AnyLanVpnInterfaceParcel], Field(discriminator="type_"), ] diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py index 5f7cec6b6..98036bebe 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py @@ -1,8 +1,9 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates +from ipaddress import IPv4Address, IPv6Address from typing import Literal, Optional, Union -from pydantic import BaseModel, ConfigDict, Field +from pydantic import AliasPath, BaseModel, ConfigDict, Field from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( @@ -21,14 +22,14 @@ class GreAddress(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") address: Union[Variable, Global[str]] mask: Union[Variable, Global[str]] class TunnelSourceIP(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") tunnel_source: Union[Global[str], Variable] = Field( serialization_alias="tunnelSource", validation_alias="tunnelSource" @@ -39,9 +40,9 @@ class TunnelSourceIP(BaseModel): class TunnelSourceIPv6(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") - tunnel_source_v6: Union[Global[str], Variable] = Field( + tunnel_source_v6: Union[Global[str], Global[IPv6Address], Variable] = Field( serialization_alias="tunnelSourceV6", validation_alias="tunnelSourceV6" ) tunnel_route_via: Optional[Union[Global[str], Variable, Default[None]]] = Field( @@ -50,7 +51,7 @@ class TunnelSourceIPv6(BaseModel): class TunnelSourceInterface(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") tunnel_source_interface: Union[Global[str], Variable] = Field( serialization_alias="tunnelSourceInterface", validation_alias="tunnelSourceInterface" @@ -61,13 +62,13 @@ class TunnelSourceInterface(BaseModel): class GreSourceIp(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") source_ip: TunnelSourceIP = Field(serialization_alias="sourceIp", validation_alias="sourceIp") class GreSourceNotLoopback(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") source_not_loopback: TunnelSourceInterface = Field( serialization_alias="sourceNotLoopback", validation_alias="sourceNotLoopback" @@ -75,7 +76,7 @@ class GreSourceNotLoopback(BaseModel): class GreSourceLoopback(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") source_loopback: TunnelSourceInterface = Field( serialization_alias="sourceLoopback", validation_alias="sourceLoopback" @@ -83,39 +84,39 @@ class GreSourceLoopback(BaseModel): class GreSourceIPv6(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") source_ipv6: TunnelSourceIPv6 = Field(serialization_alias="sourceIpv6", validation_alias="sourceIpv6") class BasicGre(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") - interface_name: Optional[Union[Global[str], Variable]] = Field( - serialization_alias="ifName", validation_alias="ifName", default=None + if_name: Union[Global[str], Variable] = Field( + serialization_alias="ifName", validation_alias="ifName", description="Minimum length of the value should be 4." ) - description: Optional[Union[Global[str], Variable, Default[None]]] = None + description: Union[Global[str], Variable, Default[None]] = Field(default=Default[None](value=None)) address: Optional[GreAddress] = None - ipv6_address: Optional[Union[Global[str], Variable, Default[None]]] = Field( + ipv6_address: Optional[Union[Global[str], Global[IPv6Address], Variable, Default[None]]] = Field( serialization_alias="ipv6Address", validation_alias="ipv6Address", default=None ) shutdown: Optional[Union[Global[bool], Variable, Default[bool]]] = Default[bool](value=False) tunnel_protection: Optional[Union[Global[bool], Variable, Default[bool]]] = Field( serialization_alias="tunnelProtection", validation_alias="tunnelProtection", default=Default[bool](value=False) ) - tunnel_mode: Optional[Union[Global[GreTunnelMode], Default[GreTunnelMode]]] = Field( + tunnel_mode: Union[Global[GreTunnelMode], Default[GreTunnelMode]] = Field( + default=Default[GreTunnelMode](value="ipv4"), serialization_alias="tunnelMode", validation_alias="tunnelMode", - default=Default[GreTunnelMode](value="ipv4"), ) tunnel_source_type: Optional[Union[GreSourceIp, GreSourceNotLoopback, GreSourceLoopback, GreSourceIPv6]] = Field( serialization_alias="tunnelSourceType", validation_alias="tunnelSourceType", default=None ) - tunnel_destination: Optional[Union[Global[str], Variable]] = Field( + tunnel_destination: Union[Global[str], Global[IPv4Address], Variable] = Field( serialization_alias="tunnelDestination", validation_alias="tunnelDestination", default=None ) - tunnel_destination_v6: Optional[Union[Global[str], Variable]] = Field( - serialization_alias="tunnelDestinationV6", validation_alias="tunnelDestinationV6", default=None + tunnel_destination_v6: Optional[Union[Global[str], Global[IPv6Address], Variable]] = Field( + default=None, serialization_alias="tunnelDestinationV6", validation_alias="tunnelDestinationV6" ) mtu: Optional[Union[Global[int], Variable, Default[int]]] = Default[int](value=1500) mtu_v6: Optional[Union[Global[int], Variable, Default[None]]] = Field( @@ -185,14 +186,14 @@ class BasicGre(BaseModel): class AdvancedGre(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") application: Optional[Union[Global[TunnelApplication], Variable]] = None -class InterfaceGreData(_ParcelBase): +class InterfaceGreParcel(_ParcelBase): type_: Literal["gre"] = Field(default="gre", exclude=True) - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") - basic: BasicGre - advanced: Optional[AdvancedGre] = None + basic: BasicGre = Field(validation_alias=AliasPath("data", "basic")) + advanced: Optional[AdvancedGre] = Field(default=None, validation_alias=AliasPath("data", "advanced")) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py index 762f5deb6..41521d0ae 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py @@ -578,7 +578,7 @@ class MplsVpnIPv6RouteTarget(BaseModel): class LanVpnParcel(_ParcelBase): - type_: Literal["vpn"] = Field(default="vpn", exclude=True) + type_: Literal["lan/vpn"] = Field(default="lan/vpn", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") vpn_id: Union[Variable, Global[int], Default[int]] = Field( diff --git a/catalystwan/tests/test_feature_profile_api.py b/catalystwan/tests/test_feature_profile_api.py index a1fd935d1..314da4dd4 100644 --- a/catalystwan/tests/test_feature_profile_api.py +++ b/catalystwan/tests/test_feature_profile_api.py @@ -5,12 +5,12 @@ from parameterized import parameterized # type: ignore -from catalystwan.api.configuration_groups.parcel import Global +from catalystwan.api.configuration_groups.parcel import as_global from catalystwan.api.feature_profile_api import ServiceFeatureProfileAPI, SystemFeatureProfileAPI from catalystwan.endpoints.configuration.feature_profile.sdwan.service import ServiceFeatureProfile from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile from catalystwan.models.configuration.feature_profile.sdwan.service import LanVpnDhcpServerParcel, LanVpnParcel -from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import AddressPool, SubnetMask +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre, InterfaceGreParcel from catalystwan.models.configuration.feature_profile.sdwan.system import ( AAAParcel, BannerParcel, @@ -25,7 +25,7 @@ SNMPParcel, ) -endpoint_mapping = { +system_endpoint_mapping = { AAAParcel: "aaa", BannerParcel: "banner", BasicParcel: "basic", @@ -49,7 +49,7 @@ def setUp(self): self.api = SystemFeatureProfileAPI(self.mock_session) self.api.endpoint = self.mock_endpoint - @parameterized.expand(endpoint_mapping.items()) + @parameterized.expand(system_endpoint_mapping.items()) def test_delete_method_with_valid_arguments(self, parcel, expected_path): # Act self.api.delete_parcel(self.profile_uuid, parcel, self.parcel_uuid) @@ -57,7 +57,7 @@ def test_delete_method_with_valid_arguments(self, parcel, expected_path): # Assert self.mock_endpoint.delete.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) - @parameterized.expand(endpoint_mapping.items()) + @parameterized.expand(system_endpoint_mapping.items()) def test_get_method_with_valid_arguments(self, parcel, expected_path): # Act self.api.get_parcels(self.profile_uuid, parcel, self.parcel_uuid) @@ -65,7 +65,7 @@ def test_get_method_with_valid_arguments(self, parcel, expected_path): # Assert self.mock_endpoint.get_by_id.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) - @parameterized.expand(endpoint_mapping.items()) + @parameterized.expand(system_endpoint_mapping.items()) def test_get_all_method_with_valid_arguments(self, parcel, expected_path): # Act self.api.get_parcels(self.profile_uuid, parcel) @@ -73,7 +73,7 @@ def test_get_all_method_with_valid_arguments(self, parcel, expected_path): # Assert self.mock_endpoint.get_all.assert_called_once_with(self.profile_uuid, expected_path) - @parameterized.expand(endpoint_mapping.items()) + @parameterized.expand(system_endpoint_mapping.items()) def test_create_method_with_valid_arguments(self, parcel, expected_path): # Act self.api.create_parcel(self.profile_uuid, parcel) @@ -81,7 +81,7 @@ def test_create_method_with_valid_arguments(self, parcel, expected_path): # Assert self.mock_endpoint.create.assert_called_once_with(self.profile_uuid, expected_path, parcel) - @parameterized.expand(endpoint_mapping.items()) + @parameterized.expand(system_endpoint_mapping.items()) def test_update_method_with_valid_arguments(self, parcel, expected_path): # Act self.api.update_parcel(self.profile_uuid, parcel, self.parcel_uuid) @@ -90,16 +90,18 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path): self.mock_endpoint.update.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid, parcel) -top_level_service_parcels = [ +service_endpoint_mapping = { + LanVpnDhcpServerParcel: "dhcp-server", + LanVpnParcel: "lan/vpn", +} + +service_interface_parcels = [ ( - "dhcp-server", - LanVpnDhcpServerParcel( - parcel_name="DhcpServerDefault", - parcel_description="Dhcp Server Parcel", - address_pool=AddressPool( - network_address=Global[IPv4Address](value=IPv4Address("10.0.0.2")), - subnet_mask=Global[SubnetMask](value="255.255.255.255"), - ), + "gre", + InterfaceGreParcel( + parcel_name="TestGreParcel", + parcel_description="Test Gre Parcel", + basic=BasicGre(if_name=as_global("gre1"), tunnel_destination=as_global(IPv4Address("4.4.4.4"))), ), ) ] @@ -108,30 +110,27 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path): class TestServiceFeatureProfileAPI(unittest.TestCase): def setUp(self): self.profile_uuid = UUID("054d1b82-9fa7-43c6-98fb-4355da0d77ff") + self.vpn_uuid = UUID("054d1b82-9fa7-43c6-98fb-4355da0d77ff") self.parcel_uuid = UUID("7113505f-8cec-4420-8799-1a209357ba7e") self.mock_session = Mock() self.mock_endpoint = Mock(spec=ServiceFeatureProfile) self.api = ServiceFeatureProfileAPI(self.mock_session) self.api.endpoint = self.mock_endpoint - @parameterized.expand(top_level_service_parcels) - def test_post_method_with_top_level_parcel(self, parcel_type, parcel): + @parameterized.expand(service_endpoint_mapping.items()) + def test_post_method_parcel(self, parcel, parcel_type): # Act self.api.create_parcel(self.profile_uuid, parcel) # Assert - self.mock_endpoint.create_top_level_service_parcel.assert_called_once_with( - self.profile_uuid, parcel_type, parcel - ) + self.mock_endpoint.create_service_parcel.assert_called_once_with(self.profile_uuid, parcel_type, parcel) - def test_post_method_with_vpn_parcel(self): - # Arrange - vpn_parcel = LanVpnParcel( - parcel_name="TestVpnParcel", - parcel_description="Test Vpn Parcel", - ) + @parameterized.expand(service_interface_parcels) + def test_post_method_interface_parcel(self, parcel_type, parcel): # Act - self.api.create_parcel(self.profile_uuid, vpn_parcel) + self.api.create_parcel(self.profile_uuid, parcel, self.vpn_uuid) # Assert - self.mock_endpoint.create_lan_vpn_service_parcel.assert_called_once_with(self.profile_uuid, vpn_parcel) + self.mock_endpoint.create_lan_vpn_interface_parcel.assert_called_once_with( + self.profile_uuid, self.vpn_uuid, parcel_type, parcel + ) diff --git a/catalystwan/utils/config_migration/converters/feature_template/__init__.py b/catalystwan/utils/config_migration/converters/feature_template/__init__.py index 477bbd793..9a8ba538f 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/__init__.py +++ b/catalystwan/utils/config_migration/converters/feature_template/__init__.py @@ -1,7 +1,7 @@ from typing import List -from .factory_method import choose_parcel_converter, create_parcel_from_template from .normalizer import template_definition_normalization +from .parcel_factory import choose_parcel_converter, create_parcel_from_template __all__ = ["create_parcel_from_template", "choose_parcel_converter", "template_definition_normalization"] diff --git a/catalystwan/utils/config_migration/converters/feature_template/gre.py b/catalystwan/utils/config_migration/converters/feature_template/gre.py new file mode 100644 index 000000000..022efc688 --- /dev/null +++ b/catalystwan/utils/config_migration/converters/feature_template/gre.py @@ -0,0 +1,111 @@ +from copy import deepcopy +from ipaddress import IPv4Interface +from typing import Tuple + +from catalystwan.api.configuration_groups.parcel import as_global +from catalystwan.models.configuration.feature_profile.sdwan.service import InterfaceGreParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import IkeGroup +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import ( + GreAddress, + GreSourceIPv6, + TunnelSourceIPv6, +) + + +class InterfaceGRETemplateConverter: + supported_template_types = ("cisco_vpn_interface_gre",) + + delete_keys = ( + "dead_peer_detection", + "ike", + "ipsec", + "ip", + "tunnel_source_v6", + "tunnel_route_via", + "authentication_type", + "access_list", + "ipv6", + "rewrite_rule", + "multiplexing", + "tracker", + ) + + def create_parcel(self, name: str, description: str, template_values: dict) -> InterfaceGreParcel: + """ + Create a new InterfaceGreParcel object. + + Args: + name (str): The name of the parcel. + description (str): The description of the parcel. + template_values (dict): A dictionary containing template values. + + Returns: + InterfaceGreParcel: The created InterfaceGreParcel object. + """ + basic_values, advanced_values = self.prepare_values(template_values) + self.configure_dead_peer_detection(basic_values) + self.configure_ike(basic_values) + self.configure_ipsec(basic_values) + self.configure_tunnel(basic_values) + self.configure_ipv6_address(basic_values) + self.configure_gre_address(basic_values) + self.cleanup_keys(basic_values) + parcel_values = self.prepare_parcel_values(name, description, basic_values, advanced_values) + return InterfaceGreParcel(**parcel_values) # type: ignore + + def prepare_values(self, template_values: dict) -> Tuple[dict, dict]: + values = deepcopy(template_values) + advanced_application = values.pop("application", None) + basic_values = {**values} + if advanced_application: + advanced_values = {"application": advanced_application} + return basic_values, advanced_values + + def prepare_parcel_values(self, name, description, basic_values, advanced_values): + return { + "parcel_name": name, + "parcel_description": description, + "basic": basic_values, + "advanced": advanced_values, + } + + def configure_dead_peer_detection(self, values: dict) -> None: + values["dpd_interval"] = values.get("dead_peer_detection", {}).get("dpd_interval") + values["dpd_retries"] = values.get("dead_peer_detection", {}).get("dpd_retries") + + def configure_ipv6_address(self, values: dict) -> None: + values["ipv6_address"] = values.get("ipv6", {}).get("address") + + def configure_gre_address(self, values: dict) -> None: + address = values.get("ip", {}).get("address", {}) + if address: + network = IPv4Interface(address.value).network + gre_address = GreAddress( + address=as_global(str(network.network_address)), + mask=as_global(str(network.netmask)), + ) + values["address"] = gre_address + + def configure_ike(self, values: dict) -> None: + ike = values.get("ike", {}) + if ike: + if ike_group := ike.get("ike_group"): + ike["ike_group"] = as_global(ike_group.value, IkeGroup) + ike.update(ike.get("authentication_type", {}).get("pre_shared_key", {})) + values.update(ike) + + def configure_ipsec(self, values: dict) -> None: + values.update(values.get("ipsec", {})) + + def configure_tunnel(self, values: dict) -> None: + if tunnel_source_v6 := values.get("tunnel_source_v6"): + values["tunnel_source_type"] = GreSourceIPv6( + source_ipv6=TunnelSourceIPv6( + tunnel_source_v6=tunnel_source_v6, + tunnel_route_via=values.get("tunnel_route_via"), + ) + ) + + def cleanup_keys(self, values: dict) -> None: + for key in self.delete_keys: + values.pop(key, None) diff --git a/catalystwan/utils/config_migration/converters/feature_template/normalizer.py b/catalystwan/utils/config_migration/converters/feature_template/normalizer.py index 1a779010f..0fb2c596e 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/normalizer.py +++ b/catalystwan/utils/config_migration/converters/feature_template/normalizer.py @@ -4,6 +4,14 @@ from catalystwan.api.configuration_groups.parcel import Global, as_global from catalystwan.models.common import TLOCColor from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import SubnetMask +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( + IkeCiphersuite, + IkeMode, + IpsecCiphersuite, + PfsGroup, + TunnelApplication, +) +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import GreTunnelMode from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import Direction from catalystwan.models.configuration.feature_profile.sdwan.system.logging_parcel import ( AuthType, @@ -23,6 +31,12 @@ TLOCColor, SubnetMask, Direction, + IkeCiphersuite, + IkeMode, + IpsecCiphersuite, + PfsGroup, + TunnelApplication, + GreTunnelMode, ] CastedTypes = Union[ diff --git a/catalystwan/utils/config_migration/converters/feature_template/factory_method.py b/catalystwan/utils/config_migration/converters/feature_template/parcel_factory.py similarity index 97% rename from catalystwan/utils/config_migration/converters/feature_template/factory_method.py rename to catalystwan/utils/config_migration/converters/feature_template/parcel_factory.py index f86b459f3..4d76bd85f 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/factory_method.py +++ b/catalystwan/utils/config_migration/converters/feature_template/parcel_factory.py @@ -17,6 +17,7 @@ from .bfd import BFDTemplateConverter from .bgp import BGPTemplateConverter from .global_ import GlobalTemplateConverter +from .gre import InterfaceGRETemplateConverter from .logging_ import LoggingTemplateConverter from .normalizer import template_definition_normalization from .ntp import NTPTemplateConverter @@ -45,6 +46,7 @@ SNMPTemplateConverter, AppqoeTemplateConverter, LanVpnParcelTemplateConverter, + InterfaceGRETemplateConverter, ] diff --git a/catalystwan/workflows/config_migration.py b/catalystwan/workflows/config_migration.py index a04c67932..93e74a652 100644 --- a/catalystwan/workflows/config_migration.py +++ b/catalystwan/workflows/config_migration.py @@ -51,6 +51,7 @@ "dhcp", "cisco_dhcp_server", "cisco_vpn", + "cisco_vpn_interface_gre", ] FEATURE_PROFILE_SYSTEM = [ @@ -87,6 +88,11 @@ "ucse", ] +FEATURE_PROFILE_SERVICE = [ + "cisco_vpn", + "cisco_vpn_interface_gre", +] + def log_progress(task: str, completed: int, total: int) -> None: logger.info(f"{task} {completed}/{total}") @@ -121,6 +127,17 @@ def transform(ux1: UX1Config) -> UX2Config: description="other", ), ) + fp_service_uuid = uuid4() + transformed_fp_service = TransformedFeatureProfile( + header=TransformHeader( + type="service", + origin=fp_service_uuid, + ), + feature_profile=FeatureProfileCreationPayload( + name=f"{dt.template_name}_service", + description="service", + ), + ) for template in templates: # Those feature templates IDs are real UUIDs and are used to map to the feature profiles @@ -128,12 +145,14 @@ def transform(ux1: UX1Config) -> UX2Config: transformed_fp_system.header.subelements.add(UUID(template.templateId)) elif template.templateType in FEATURE_PROFILE_OTHER: transformed_fp_other.header.subelements.add(UUID(template.templateId)) + elif template.templateType in FEATURE_PROFILE_SERVICE: + transformed_fp_service.header.subelements.add(UUID(template.templateId)) transformed_cg = TransformedConfigGroup( header=TransformHeader( type="config_group", origin=uuid4(), - subelements=set([fp_system_uuid, fp_other_uuid]), + subelements=set([fp_system_uuid, fp_other_uuid, fp_service_uuid]), ), config_group=ConfigGroupCreationPayload( name=dt.template_name, @@ -145,6 +164,7 @@ def transform(ux1: UX1Config) -> UX2Config: # Add to UX2 ux2.feature_profiles.append(transformed_fp_system) ux2.feature_profiles.append(transformed_fp_other) + ux2.feature_profiles.append(transformed_fp_service) ux2.config_groups.append(transformed_cg) for ft in ux1.templates.feature_templates: