From 4a0b4f044a3cb127e51785ee134f38b9f61cf0a2 Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Tue, 19 Mar 2024 15:54:33 +0100 Subject: [PATCH] Start work on Service models --- .../api/configuration_groups/parcel.py | 4 + catalystwan/api/feature_profile_api.py | 13 +- .../feature_profile/sdwan/service.py | 29 ++++- .../sdwan/service/test_models.py | 22 +++- .../feature_profile/sdwan/service/__init__.py | 33 ++++- .../feature_profile/sdwan/service/appqoe.py | 7 +- .../sdwan/service/lan/ethernet.py | 14 +-- .../feature_profile/sdwan/service/lan/gre.py | 14 +-- .../sdwan/service/lan/ipsec.py | 13 +- .../feature_profile/sdwan/service/lan/svi.py | 16 +-- .../feature_profile/sdwan/service/lan/vpn.py | 101 ++++++++------- catalystwan/tests/test_feature_profile_api.py | 117 +++++++++++------- .../converters/feature_template/appqoe.py | 1 + .../feature_template/factory_method.py | 2 + .../converters/feature_template/vpn.py | 56 +++++++++ 15 files changed, 289 insertions(+), 153 deletions(-) create mode 100644 catalystwan/utils/config_migration/converters/feature_template/vpn.py diff --git a/catalystwan/api/configuration_groups/parcel.py b/catalystwan/api/configuration_groups/parcel.py index ec68994f..b0f81653 100644 --- a/catalystwan/api/configuration_groups/parcel.py +++ b/catalystwan/api/configuration_groups/parcel.py @@ -89,6 +89,10 @@ class Global(ParcelAttribute, Generic[T]): ) value: T + def __bool__(self) -> bool: + # if statements use __len__ when __bool__ is not defined + return True + def __len__(self) -> int: if isinstance(self.value, (str, list)): return len(self.value) diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index c553ebfd..810af09e 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/catalystwan/api/feature_profile_api.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Optional, Protocol, Type, Union, overload +from typing import TYPE_CHECKING, Any, Optional, Protocol, Type, Union, get_args, overload from uuid import UUID from pydantic import Json @@ -12,6 +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.typed_list import DataSequence if TYPE_CHECKING: @@ -231,6 +232,14 @@ 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: + """ + 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) + class SystemFeatureProfileAPI: """ @@ -485,7 +494,7 @@ def create_parcel(self, profile_id: UUID, payload: AnySystemParcel) -> ParcelCre return self.endpoint.create(profile_id, payload._get_parcel_type(), payload) - def update(self, profile_id: UUID, payload: AnySystemParcel, parcel_id: UUID) -> ParcelCreationResponse: + def update_parcel(self, profile_id: UUID, payload: AnySystemParcel, parcel_id: UUID) -> ParcelCreationResponse: """ Update System Parcel for selected profile_id based on payload type """ diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py index 9a4faa4c..2d57e225 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py @@ -10,7 +10,13 @@ FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesPayload, + ParcelCreationResponse, ) +from catalystwan.models.configuration.feature_profile.sdwan.service import ( + AnyLanVpnInterfaceParcel, + AnyTopLevelServiceParcel, +) +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel from catalystwan.typed_list import DataSequence @@ -30,6 +36,25 @@ def create_sdwan_service_feature_profile( ... @versions(supported_versions=(">=20.9"), raises=False) - @delete("/v1/feature-profile/sdwan/service/{profile_id}") - def delete_sdwan_service_feature_profile(self, profile_id: UUID) -> None: + @delete("/v1/feature-profile/sdwan/service/{profile_uuid}") + 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( + 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( + self, profile_uuid: UUID, vpn_uuid: UUID, parcel_type: str, payload: AnyLanVpnInterfaceParcel + ) -> ParcelCreationResponse: ... diff --git a/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py b/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py index 7233570f..0ffe4323 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/service/test_models.py @@ -9,6 +9,7 @@ LanVpnDhcpServerParcel, SubnetMask, ) +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel from catalystwan.session import create_manager_session @@ -26,7 +27,6 @@ def setUp(self) -> None: def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): # Arrange - url = f"dataservice/v1/feature-profile/sdwan/service/{self.profile_id}/dhcp-server" dhcp_server_parcel = LanVpnDhcpServerParcel( parcel_name="DhcpServerDefault", parcel_description="Dhcp Server Parcel", @@ -36,11 +36,23 @@ def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): ), ) # Act - response = self.session.post( - url=url, data=dhcp_server_parcel.model_dump_json(by_alias=True, exclude_none=True) - ) # This will be changed to the actual method + 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 response.status_code == 200 + assert parcel_id def tearDown(self) -> None: self.session.api.sdwan_feature_profiles.service.delete_profile(self.profile_id) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py b/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py index e8998294..49911852 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/__init__.py @@ -3,16 +3,47 @@ from pydantic import Field from typing_extensions import Annotated +from .appqoe import AppqoeParcel from .dhcp_server import LanVpnDhcpServerParcel +from .lan.ethernet import InterfaceEthernetData +from .lan.gre import InterfaceGreData +from .lan.ipsec import InterfaceIpsecData +from .lan.svi import InterfaceSviData +from .lan.vpn import LanVpnParcel + +AnyTopLevelServiceParcel = Annotated[ + Union[ + LanVpnDhcpServerParcel, + AppqoeParcel, + # TrackerGroupData, + # WirelessLanData, + # SwitchportData + ], + Field(discriminator="type_"), +] + +AnyLanVpnInterfaceParcel = Annotated[ + Union[ + InterfaceEthernetData, + InterfaceGreData, + InterfaceIpsecData, + InterfaceSviData, + ], + Field(discriminator="type_"), +] AnyServiceParcel = Annotated[ - Union[LanVpnDhcpServerParcel,], # noqa: E231 + Union[AnyTopLevelServiceParcel, LanVpnParcel, AnyLanVpnInterfaceParcel], Field(discriminator="type_"), ] __all__ = [ "LanVpnDhcpServerParcel", + "AppqoeParcel", + "LanVpnParcel", "AnyServiceParcel", + "AnyTopLevelServiceParcel", + "AnyLanVpnInterfaceParcel", ] diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/appqoe.py b/catalystwan/models/configuration/feature_profile/sdwan/service/appqoe.py index a369e9d1..341c6529 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/appqoe.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/appqoe.py @@ -112,7 +112,7 @@ class Appqoe(BaseModel): class ServiceContext(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) - appqoe: List[Appqoe] + appqoe: List[Appqoe] = Field(default_factory=lambda: [Appqoe()]) # Frowarder @@ -161,7 +161,9 @@ class ForwarderRole(BaseModel): service_node_group: List[ForwarderNodeGroup] = Field( serialization_alias="serviceNodeGroup", validation_alias="serviceNodeGroup" ) - service_context: ServiceContext = Field(serialization_alias="serviceContext", validation_alias="serviceContext") + service_context: ServiceContext = Field( + default_factory=ServiceContext, serialization_alias="serviceContext", validation_alias="serviceContext" + ) # Forwarder and Service @@ -248,6 +250,7 @@ class ServiceNodeRole(BaseModel): class AppqoeParcel(_ParcelBase): + type_: Literal["appqoe"] = Field(default="appqoe", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) dreopt: Optional[Union[Global[bool], Default[bool]]] = Field( diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py index 40c39226..60e0a4f1 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, ConfigDict, Field -from catalystwan.api.configuration_groups.parcel import Default, Global, Variable +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( Arp, StaticIPv4Address, @@ -281,7 +281,8 @@ class AdvancedAttributes(BaseModel): ) -class InterfaceEthernetData(BaseModel): +class InterfaceEthernetData(_ParcelBase): + type_: Literal["ethernet"] = Field(default="ethernet", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) shutdown: Union[Global[bool], Variable, Default[bool]] = Default[bool](value=True) @@ -316,12 +317,3 @@ class InterfaceEthernetData(BaseModel): arp: Optional[List[Arp]] = None trustsec: Optional[Trustsec] = None advanced: AdvancedAttributes = AdvancedAttributes() - - -class InterfaceEthernetCreationPayload(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - name: str - description: Optional[str] = None - data: InterfaceEthernetData - metadata: Optional[dict] = None 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 5858f7d2..5f7cec6b 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/gre.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict, Field -from catalystwan.api.configuration_groups.parcel import Default, Global, Variable +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( IkeCiphersuite, IkeGroup, @@ -190,17 +190,9 @@ class AdvancedGre(BaseModel): application: Optional[Union[Global[TunnelApplication], Variable]] = None -class InterfaceGreData(BaseModel): +class InterfaceGreData(_ParcelBase): + type_: Literal["gre"] = Field(default="gre", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) basic: BasicGre advanced: Optional[AdvancedGre] = None - - -class InterfaceGreCreationPayload(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - name: str - description: Optional[str] = None - data: InterfaceGreData - metadata: Optional[dict] = None diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ipsec.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ipsec.py index de9c90d1..d037cc3b 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ipsec.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ipsec.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict, Field -from catalystwan.api.configuration_groups.parcel import Default, Global, Variable +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( IkeCiphersuite, IkeGroup, @@ -28,7 +28,8 @@ class IpsecAddress(BaseModel): mask: Union[Variable, Global[str]] -class InterfaceIpsecData(BaseModel): +class InterfaceIpsecData(_ParcelBase): + type_: Literal["ipsec"] = Field(default="ipsec", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) interface_name: Union[Global[str], Variable] = Field(serialization_alias="ifName", validation_alias="ifName") @@ -128,11 +129,3 @@ class InterfaceIpsecData(BaseModel): tunnel_route_via: Optional[Union[Global[str], Variable, Default[None]]] = Field( serialization_alias="tunnelRouteVia", validation_alias="tunnelRouteVia", default=None ) - - -class InterfaceIpsecCreationPayload(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - name: str - description: Optional[str] = None - data: InterfaceIpsecData diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/svi.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/svi.py index f4d6d653..59a58c84 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/svi.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/svi.py @@ -1,11 +1,11 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates -from typing import List, Optional, Union +from typing import List, Literal, Optional, Union from uuid import UUID from pydantic import BaseModel, ConfigDict, Field -from catalystwan.api.configuration_groups.parcel import Default, Global, Variable +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase from catalystwan.models.configuration.feature_profile.sdwan.service.lan.common import ( Arp, StaticIPv4Address, @@ -140,7 +140,8 @@ class AclQos(BaseModel): ) -class InterfaceSviData(BaseModel): +class InterfaceSviData(_ParcelBase): + type_: Literal["svi"] = Field(default="svi", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) shutdown: Union[Global[bool], Variable, Default[bool]] = Default[bool](value=True) @@ -166,12 +167,3 @@ class InterfaceSviData(BaseModel): serialization_alias="dhcpClientV6", validation_alias="dhcpClientV6", default=Default[bool](value=False) ) advanced: AdvancedSviAttributes = AdvancedSviAttributes() - - -class InterfaceSviCreationPayload(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - name: str - description: Optional[str] = None - data: InterfaceSviData - metadata: Optional[dict] = None 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 e6f0ac95..83ecc565 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py @@ -1,11 +1,12 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates +from ipaddress import IPv4Address from typing import List, Literal, Optional, Union from uuid import UUID -from pydantic import BaseModel, ConfigDict, Field +from pydantic import AliasPath, BaseModel, ConfigDict, Field -from catalystwan.api.configuration_groups.parcel import Default, Global, Variable +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase, as_default from catalystwan.models.configuration.feature_profile.common import Prefix ProtocolIPv4 = Literal[ @@ -97,10 +98,14 @@ class DnsIPv4(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) primary_dns_address_ipv4: Union[Variable, Global[str], Default[None]] = Field( - serialization_alias="primaryDnsAddressIpv4", validation_alias="primaryDnsAddressIpv4" + default=Default[None](value=None), + serialization_alias="primaryDnsAddressIpv4", + validation_alias="primaryDnsAddressIpv4", ) secondary_dns_address_ipv4: Union[Variable, Global[str], Default[None]] = Field( - serialization_alias="secondaryDnsAddressIpv4", validation_alias="secondaryDnsAddressIpv4" + default=Default[None](value=None), + serialization_alias="secondaryDnsAddressIpv4", + validation_alias="secondaryDnsAddressIpv4", ) @@ -390,8 +395,12 @@ class NatPool(BaseModel): prefix_length: Union[Variable, Global[int]] = Field( serialization_alias="prefixLength", validation_alias="prefixLength" ) - range_start: Union[Variable, Global[str]] = Field(serialization_alias="rangeStart", validation_alias="rangeStart") - range_end: Union[Variable, Global[str]] = Field(serialization_alias="rangeEnd", validation_alias="rangeEnd") + range_start: Union[Variable, Global[str], Global[IPv4Address]] = Field( + serialization_alias="rangeStart", validation_alias="rangeStart" + ) + range_end: Union[Variable, Global[str], Global[IPv4Address]] = Field( + serialization_alias="rangeEnd", validation_alias="rangeEnd" + ) overload: Union[Variable, Global[bool], Default[bool]] = Default[bool](value=True) direction: Union[Variable, Global[Direction]] tracking_object: Optional[dict] = Field( @@ -460,10 +469,10 @@ class Nat64v4Pool(BaseModel): nat64_v4_pool_name: Union[Variable, Global[str]] = Field( serialization_alias="nat64V4PoolName", validation_alias="nat64V4PoolName" ) - nat64_v4_pool_range_start: Union[Variable, Global[str]] = Field( + nat64_v4_pool_range_start: Union[Variable, Global[str], Global[IPv4Address]] = Field( serialization_alias="nat64V4PoolRangeStart", validation_alias="nat64V4PoolRangeStart" ) - nat64_v4_pool_range_end: Union[Variable, Global[str]] = Field( + nat64_v4_pool_range_end: Union[Variable, Global[str], Global[IPv4Address]] = Field( serialization_alias="nat64V4PoolRangeEnd", validation_alias="nat64V4PoolRangeEnd" ) nat64_v4_pool_overload: Union[Variable, Global[bool], Default[bool]] = Field( @@ -558,79 +567,67 @@ class MplsVpnIPv6RouteTarget(BaseModel): ) -class LanVpnData(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) +class LanVpnParcel(_ParcelBase): + type_: Literal["vpn"] = Field(default="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(serialization_alias="vpnId", validation_alias="vpnId") - name: Union[Variable, Global[str], Default[None]] + vpn_id: Union[Variable, Global[int], Default[int]] = Field( + default=as_default(1), validation_alias=AliasPath("data", "vpnId") + ) + vpn_name: Union[Variable, Global[str], Default[None]] = Field( + default=Default[None](value=None), validation_alias=AliasPath("data", "name") + ) omp_admin_distance: Optional[Union[Variable, Global[int], Default[None]]] = Field( - serialization_alias="ompAdminDistance", validation_alias="ompAdminDistance", default=None + validation_alias=AliasPath("data", "ompAdminDistance"), default=None ) omp_admin_distance_ipv6: Optional[Union[Variable, Global[int], Default[None]]] = Field( - serialization_alias="ompAdminDistanceIpv6", validation_alias="ompAdminDistanceIpv6", default=None + validation_alias=AliasPath("data", "ompAdminDistanceIpv6"), default=None ) - dns_ipv4: Optional[DnsIPv4] = Field(serialization_alias="dnsIpv4", validation_alias="dnsIpv4", default=None) - dns_ipv6: Optional[DnsIPv6] = Field(serialization_alias="dnsIpv6", validation_alias="dnsIpv6", default=None) + dns_ipv4: Optional[DnsIPv4] = Field(validation_alias=AliasPath("data", "dnsIpv4"), default=None) + dns_ipv6: Optional[DnsIPv6] = Field(validation_alias=AliasPath("data", "dnsIpv6"), default=None) new_host_mapping: Optional[List[HostMapping]] = Field( - serialization_alias="newHostMapping", validation_alias="newHostMapping", default=None + validation_alias=AliasPath("data", "newHostMapping"), default=None ) omp_advertise_ipv4: Optional[List[OmpAdvertiseIPv4]] = Field( - serialization_alias="ompAdvertiseIpv4", validation_alias="ompAdvertiseIpv4", default=None + validation_alias=AliasPath("data", "ompAdvertiseIpv4"), default=None ) omp_advertise_ipv6: Optional[List[OmpAdvertiseIPv6]] = Field( - serialization_alias="ompAdvertiseIpv6", validation_alias="ompAdvertiseIpv6", default=None - ) - ipv4_route: Optional[List[StaticRouteIPv4]] = Field( - serialization_alias="ipv4Route", validation_alias="ipv4Route", default=None - ) - ipv6_route: Optional[List[StaticRouteIPv6]] = Field( - serialization_alias="ipv6Route", validation_alias="ipv6Route", default=None + validation_alias=AliasPath("data", "ompAdvertiseIpv6"), default=None ) + ipv4_route: Optional[List[StaticRouteIPv4]] = Field(validation_alias=AliasPath("data", "ipv4Route"), default=None) + ipv6_route: Optional[List[StaticRouteIPv6]] = Field(validation_alias=AliasPath("data", "ipv6Route"), default=None) service: Optional[List[Service]] = None service_route: Optional[List[ServiceRoute]] = Field( - serialization_alias="serviceRoute", validation_alias="serviceRoute", default=None - ) - gre_route: Optional[List[StaticGreRouteIPv4]] = Field( - serialization_alias="greRoute", validation_alias="greRoute", default=None + validation_alias=AliasPath("data", "serviceRoute"), default=None ) + gre_route: Optional[List[StaticGreRouteIPv4]] = Field(validation_alias=AliasPath("data", "greRoute"), default=None) ipsec_route: Optional[List[StaticIpsecRouteIPv4]] = Field( - serialization_alias="ipsecRoute", validation_alias="ipsecRoute", default=None + validation_alias=AliasPath("data", "ipsecRoute"), default=None ) - nat_pool: Optional[List[NatPool]] = Field(serialization_alias="natPool", validation_alias="natPool", default=None) + nat_pool: Optional[List[NatPool]] = Field(validation_alias=AliasPath("data", "natPool"), default=None) nat_port_forwarding: Optional[List[NatPortForward]] = Field( - serialization_alias="natPortForwarding", validation_alias="natPortForwarding", default=None - ) - static_nat: Optional[List[StaticNat]] = Field( - serialization_alias="staticNat", validation_alias="staticNat", default=None + validation_alias=AliasPath("data", "natPortForwarding"), default=None ) + static_nat: Optional[List[StaticNat]] = Field(validation_alias=AliasPath("data", "staticNat"), default=None) static_nat_subnet: Optional[List[StaticNatSubnet]] = Field( - serialization_alias="staticNatSubnet", validation_alias="staticNatSubnet", default=None - ) - nat64_v4_pool: Optional[List[Nat64v4Pool]] = Field( - serialization_alias="nat64V4Pool", validation_alias="nat64V4Pool", default=None + validation_alias=AliasPath("data", "staticNatSubnet"), default=None ) + nat64_v4_pool: Optional[List[Nat64v4Pool]] = Field(validation_alias=AliasPath("data", "nat64V4Pool"), default=None) route_leak_from_global: Optional[List[RouteLeakFromGlobal]] = Field( - serialization_alias="routeLeakFromGlobal", validation_alias="routeLeakFromGlobal", default=None + validation_alias=AliasPath("data", "routeLeakFromGlobal"), default=None ) route_leak_from_service: Optional[List[RouteLeakFromService]] = Field( - serialization_alias="routeLeakFromService", validation_alias="routeLeakFromService", default=None + validation_alias=AliasPath("data", "routeLeakFromService"), default=None ) route_leak_between_services: Optional[List[RouteLeakBetweenServices]] = Field( - serialization_alias="routeLeakBetweenServices", validation_alias="routeLeakBetweenServices", default=None + validation_alias=AliasPath("data", "routeLeakBetweenServices"), default=None ) mpls_vpn_ipv4_route_target: Optional[MplsVpnIPv4RouteTarget] = Field( - serialization_alias="mplsVpnIpv4RouteTarget", validation_alias="mplsVpnIpv4RouteTarget", default=None + validation_alias=AliasPath("data", "mplsVpnIpv4RouteTarget"), default=None ) mpls_vpn_ipv6_route_target: Optional[MplsVpnIPv6RouteTarget] = Field( - serialization_alias="mplsVpnIpv6RouteTarget", validation_alias="mplsVpnIpv6RouteTarget", default=None + validation_alias=AliasPath("data", "mplsVpnIpv6RouteTarget"), default=None ) enable_sdra: Optional[Union[Global[bool], Default[bool]]] = Field( - serialization_alias="enableSdra", validation_alias="enableSdra", default=None + validation_alias=AliasPath("data", "enableSdra"), default=None ) - - -class LanVpnCreationPayload(BaseModel): - name: str - description: Optional[str] = None - data: LanVpnData - metadata: Optional[dict] = None diff --git a/catalystwan/tests/test_feature_profile_api.py b/catalystwan/tests/test_feature_profile_api.py index 955f4ed0..a1fd935d 100644 --- a/catalystwan/tests/test_feature_profile_api.py +++ b/catalystwan/tests/test_feature_profile_api.py @@ -1,10 +1,16 @@ import unittest -from unittest.mock import patch +from ipaddress import IPv4Address +from unittest.mock import Mock from uuid import UUID from parameterized import parameterized # type: ignore -from catalystwan.api.feature_profile_api import SystemFeatureProfileAPI +from catalystwan.api.configuration_groups.parcel import 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.system import ( AAAParcel, BannerParcel, @@ -38,73 +44,94 @@ class TestSystemFeatureProfileAPI(unittest.TestCase): def setUp(self): self.profile_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=SystemFeatureProfile) + self.api = SystemFeatureProfileAPI(self.mock_session) + self.api.endpoint = self.mock_endpoint @parameterized.expand(endpoint_mapping.items()) - @patch("catalystwan.session.ManagerSession") - @patch("catalystwan.endpoints.configuration.feature_profile.sdwan.system.SystemFeatureProfile") - def test_delete_method_with_valid_arguments(self, parcel, expected_path, mock_endpoint, mock_session): - # Arrange - api = SystemFeatureProfileAPI(mock_session) - api.endpoint = mock_endpoint - + def test_delete_method_with_valid_arguments(self, parcel, expected_path): # Act - api.delete_parcel(self.profile_uuid, parcel, self.parcel_uuid) + self.api.delete_parcel(self.profile_uuid, parcel, self.parcel_uuid) # Assert - mock_endpoint.delete.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) + self.mock_endpoint.delete.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) @parameterized.expand(endpoint_mapping.items()) - @patch("catalystwan.session.ManagerSession") - @patch("catalystwan.endpoints.configuration.feature_profile.sdwan.system.SystemFeatureProfile") - def test_get_method_with_valid_arguments(self, parcel, expected_path, mock_endpoint, mock_session): - # Arrange - api = SystemFeatureProfileAPI(mock_session) - api.endpoint = mock_endpoint - + def test_get_method_with_valid_arguments(self, parcel, expected_path): # Act - api.get_parcels(self.profile_uuid, parcel, self.parcel_uuid) + self.api.get_parcels(self.profile_uuid, parcel, self.parcel_uuid) # Assert - mock_endpoint.get_by_id.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) + self.mock_endpoint.get_by_id.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) @parameterized.expand(endpoint_mapping.items()) - @patch("catalystwan.session.ManagerSession") - @patch("catalystwan.endpoints.configuration.feature_profile.sdwan.system.SystemFeatureProfile") - def test_get_all_method_with_valid_arguments(self, parcel, expected_path, mock_endpoint, mock_session): - # Arrange - api = SystemFeatureProfileAPI(mock_session) - api.endpoint = mock_endpoint + def test_get_all_method_with_valid_arguments(self, parcel, expected_path): + # Act + self.api.get_parcels(self.profile_uuid, parcel) + # Assert + self.mock_endpoint.get_all.assert_called_once_with(self.profile_uuid, expected_path) + + @parameterized.expand(endpoint_mapping.items()) + def test_create_method_with_valid_arguments(self, parcel, expected_path): # Act - api.get_parcels(self.profile_uuid, parcel) + self.api.create_parcel(self.profile_uuid, parcel) # Assert - mock_endpoint.get_all.assert_called_once_with(self.profile_uuid, expected_path) + self.mock_endpoint.create.assert_called_once_with(self.profile_uuid, expected_path, parcel) @parameterized.expand(endpoint_mapping.items()) - @patch("catalystwan.session.ManagerSession") - @patch("catalystwan.endpoints.configuration.feature_profile.sdwan.system.SystemFeatureProfile") - def test_create_method_with_valid_arguments(self, parcel, expected_path, mock_endpoint, mock_session): - # Arrange - api = SystemFeatureProfileAPI(mock_session) - api.endpoint = mock_endpoint + def test_update_method_with_valid_arguments(self, parcel, expected_path): + # Act + self.api.update_parcel(self.profile_uuid, parcel, self.parcel_uuid) + # Assert + self.mock_endpoint.update.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid, parcel) + + +top_level_service_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"), + ), + ), + ) +] + + +class TestServiceFeatureProfileAPI(unittest.TestCase): + def setUp(self): + self.profile_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): # Act - api.create_parcel(self.profile_uuid, parcel) + self.api.create_parcel(self.profile_uuid, parcel) # Assert - mock_endpoint.create.assert_called_once_with(self.profile_uuid, expected_path, parcel) + self.mock_endpoint.create_top_level_service_parcel.assert_called_once_with( + self.profile_uuid, parcel_type, parcel + ) - @parameterized.expand(endpoint_mapping.items()) - @patch("catalystwan.session.ManagerSession") - @patch("catalystwan.endpoints.configuration.feature_profile.sdwan.system.SystemFeatureProfile") - def test_update_method_with_valid_arguments(self, parcel, expected_path, mock_endpoint, mock_session): + def test_post_method_with_vpn_parcel(self): # Arrange - api = SystemFeatureProfileAPI(mock_session) - api.endpoint = mock_endpoint - + vpn_parcel = LanVpnParcel( + parcel_name="TestVpnParcel", + parcel_description="Test Vpn Parcel", + ) # Act - api.update(self.profile_uuid, parcel, self.parcel_uuid) + self.api.create_parcel(self.profile_uuid, vpn_parcel) # Assert - mock_endpoint.update.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid, parcel) + self.mock_endpoint.create_lan_vpn_service_parcel.assert_called_once_with(self.profile_uuid, vpn_parcel) diff --git a/catalystwan/utils/config_migration/converters/feature_template/appqoe.py b/catalystwan/utils/config_migration/converters/feature_template/appqoe.py index 5f1a75b0..586d9995 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/appqoe.py +++ b/catalystwan/utils/config_migration/converters/feature_template/appqoe.py @@ -26,6 +26,7 @@ def create_parcel(name: str, description: str, template_values: dict) -> AppqoeP AppqoeParcel: The created AppqoeParcel object. """ values = deepcopy(template_values) + print(values) for appqoe_item in values.get("service_context", {}).get("appqoe", []): if item_name := appqoe_item.get("name"): appqoe_item["name"] = as_default(value=item_name.value) diff --git a/catalystwan/utils/config_migration/converters/feature_template/factory_method.py b/catalystwan/utils/config_migration/converters/feature_template/factory_method.py index 7cfc2f88..2288c147 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/factory_method.py +++ b/catalystwan/utils/config_migration/converters/feature_template/factory_method.py @@ -24,6 +24,7 @@ from .security import SecurityTemplateConverter from .thousandeyes import ThousandEyesTemplateConverter from .ucse import UcseTemplateConverter +from .vpn import LanVpnParcelTemplateConverter logger = logging.getLogger(__name__) @@ -43,6 +44,7 @@ DhcpTemplateConverter, SNMPTemplateConverter, AppqoeTemplateConverter, + LanVpnParcelTemplateConverter, ] diff --git a/catalystwan/utils/config_migration/converters/feature_template/vpn.py b/catalystwan/utils/config_migration/converters/feature_template/vpn.py new file mode 100644 index 00000000..49b42a82 --- /dev/null +++ b/catalystwan/utils/config_migration/converters/feature_template/vpn.py @@ -0,0 +1,56 @@ +from copy import deepcopy + +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import DnsIPv4, LanVpnParcel, Nat64v4Pool + + +class LanVpnParcelTemplateConverter: + """ + A class for converting template values into a ThousandEyesParcel object. + """ + + supported_template_types = ("cisco_vpn",) + + @staticmethod + def create_parcel(name: str, description: str, template_values: dict) -> LanVpnParcel: + """ + Creates a ThousandEyesParcel object based on the provided template values. + + Args: + name (str): The name of the parcel. + description (str): The description of the parcel. + template_values (dict): A dictionary containing the template values. + + Returns: + ThousandEyesParcel: A ThousandEyesParcel object with the provided values. + """ + values = deepcopy(template_values) + print(values) + if vpn_name := values.pop("name", None): + values["vpn_name"] = vpn_name + if nat := values.pop("nat", {}).pop("natpool", []): + values["nat_pool"] = nat + if nat64 := values.pop("nat64", {}).pop("v4", {}).pop("pool", []): + nat64_items = [] + for entry in nat64: + nat64_item = Nat64v4Pool( + nat64_v4_pool_name=entry["name"], + nat64_v4_pool_range_start=entry["start_address"], + nat64_v4_pool_range_end=entry["end_address"], + nat64_v4_pool_overload=entry["overload"], + ) + nat64_items.append(nat64_item) + values["nat64_v4_pool"] = nat64_items + if dns := values.pop("dns", {}): + dns_ipv4 = DnsIPv4() + for entry in dns: + if entry["role"] == "primary": + dns_ipv4.primary_dns_address_ipv4 = entry["dns_addr"] + elif entry["role"] == "secondary": + dns_ipv4.secondary_dns_address_ipv4 = entry["dns_addr"] + + parcel_values = { + "parcel_name": name, + "parcel_description": description, + **values, + } + return LanVpnParcel(**parcel_values) # type: ignore