From 8b148bdc3d0447a58a5c1deead46cdf1f52dd323 Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Mon, 16 Sep 2024 21:11:32 +0200 Subject: [PATCH] Add update method to service api. Move to literals for variable, global and defautl --- .../api/configuration_groups/parcel.py | 35 +-- catalystwan/api/feature_profile_api.py | 263 +++++++++++++++++- .../feature_profile/sdwan/service.py | 48 +++- .../feature_profile/sdwan/transport.py | 32 ++- .../feature_profile/sdwan/test_service.py | 178 ++++++++++-- .../feature_profile/sdwan/test_transport.py | 18 +- .../sdwan/service/dhcp_server.py | 2 +- .../feature_profile/sdwan/service/lan/vpn.py | 2 +- 8 files changed, 513 insertions(+), 65 deletions(-) diff --git a/catalystwan/api/configuration_groups/parcel.py b/catalystwan/api/configuration_groups/parcel.py index 78bc182a..4be1822d 100644 --- a/catalystwan/api/configuration_groups/parcel.py +++ b/catalystwan/api/configuration_groups/parcel.py @@ -1,6 +1,5 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates -from enum import Enum from typing import Any, Dict, Generic, List, Literal, Optional, Tuple, TypeVar from pydantic import ( @@ -13,6 +12,7 @@ SerializerFunctionWrapHandler, model_serializer, ) +from pydantic.alias_generators import to_camel from typing_extensions import get_origin from catalystwan.exceptions import CatalystwanException @@ -81,24 +81,11 @@ def _get_parcel_type(cls) -> str: raise CatalystwanException(f"{cls.__name__} field parcel type is not set.") -class OptionType(str, Enum): - GLOBAL = "global" - DEFAULT = "default" - VARIABLE = "variable" - - -class ParcelAttribute(BaseModel): - model_config = ConfigDict(extra="forbid") - option_type: OptionType = Field(serialization_alias="optionType", validation_alias="optionType") - - # https://github.com/pydantic/pydantic/discussions/6090 # Usage: Global[str](value="test") -class Global(ParcelAttribute, Generic[T]): - model_config = ConfigDict(populate_by_name=True) - option_type: OptionType = Field( - default=OptionType.GLOBAL, serialization_alias="optionType", validation_alias="optionType" - ) +class Global(BaseModel, Generic[T]): + model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) + option_type: Literal["global"] = "global" value: T def __bool__(self) -> bool: @@ -121,17 +108,15 @@ def __le__(self, other: Any) -> bool: return False -class Variable(ParcelAttribute): - option_type: OptionType = Field( - default=OptionType.VARIABLE, serialization_alias="optionType", validation_alias="optionType" - ) +class Variable(BaseModel): + model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) + option_type: Literal["variable"] = "variable" value: str = Field(pattern=r"^\{\{[.\/\[\]a-zA-Z0-9_-]+\}\}$", min_length=1, max_length=64) -class Default(ParcelAttribute, Generic[T]): - option_type: OptionType = Field( - default=OptionType.DEFAULT, serialization_alias="optionType", validation_alias="optionType" - ) +class Default(BaseModel, Generic[T]): + model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) + option_type: Literal["default"] = "default" value: Optional[T] = None diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index b98167fc..68ef29ff 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/catalystwan/api/feature_profile_api.py @@ -14,7 +14,7 @@ from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile from catalystwan.endpoints.configuration.feature_profile.sdwan.topology import TopologyFeatureProfile from catalystwan.endpoints.configuration.feature_profile.sdwan.transport import TransportFeatureProfile -from catalystwan.exceptions import ManagerHTTPError +from catalystwan.exceptions import CatalystwanException, ManagerHTTPError from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv4acl import Ipv4AclParcel from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv6acl import Ipv6AclParcel from catalystwan.models.configuration.feature_profile.sdwan.application_priority import ( @@ -45,8 +45,17 @@ from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.url import URLParcel from catalystwan.models.configuration.feature_profile.sdwan.routing import AnyRoutingParcel, RoutingBgpParcel from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import EigrpParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ethernet import InterfaceEthernetParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import InterfaceGreParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ipsec import InterfaceIpsecParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.multilink import InterfaceMultilinkParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.svi import InterfaceSviParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import LanVpnParcel from catalystwan.models.configuration.feature_profile.sdwan.service.multicast import MulticastParcel from catalystwan.models.configuration.feature_profile.sdwan.service.route_policy import RoutePolicyParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.switchport import SwitchportParcel +from catalystwan.models.configuration.feature_profile.sdwan.service.wireless_lan import WirelessLanParcel from catalystwan.models.configuration.feature_profile.sdwan.sig_security.sig_security import SIGParcel from catalystwan.models.configuration.feature_profile.sdwan.topology import AnyTopologyParcel from catalystwan.models.configuration.feature_profile.sdwan.topology.custom_control import CustomControlParcel @@ -267,7 +276,11 @@ def _get_vpn_parcel( try: return self.endpoint.get_transport_parcel(profile_id, TransportVpnParcel._get_parcel_type(), vpn_uuid) except ManagerHTTPError: + pass + try: return self.endpoint.get_transport_parcel(profile_id, ManagementVpnParcel._get_parcel_type(), vpn_uuid) + except ManagerHTTPError: + raise CatalystwanException(f"VPN parcel wih uuid: {vpn_uuid} is not found") @overload def get_parcel( @@ -321,6 +334,25 @@ def get_parcel( """ return self.endpoint.get_transport_parcel(profile_id, parcel_type._get_parcel_type(), parcel_id) + def update_parcel( + self, profile_id: UUID, payload: AnyTransportParcel, parcel_id: UUID, vpn_uuid: Optional[UUID] = None + ) -> ParcelCreationResponse: + """ + Update Transport Parcel for selected profile_id based on payload type + """ + if vpn_uuid is not None: + vpn_parcel = self._get_vpn_parcel(profile_id, vpn_uuid).payload + parcel_type = payload._get_parcel_type().removeprefix("wan/vpn/").removeprefix("management/vpn/") + if vpn_parcel._get_parcel_type() == TransportVpnParcel._get_parcel_type(): + return self.endpoint.update_transport_vpn_sub_parcel( + profile_id, vpn_uuid, parcel_type, parcel_id, payload + ) + else: + return self.endpoint.update_management_vpn_sub_parcel( + profile_id, vpn_uuid, parcel_type, parcel_id, payload + ) + return self.endpoint.update_transport_parcel(profile_id, payload._get_parcel_type(), parcel_id, payload) + class OtherFeatureProfileAPI: """ @@ -460,6 +492,235 @@ def delete_parcel(self, profile_uuid: UUID, parcel_type: Type[AnyServiceParcel], """ return self.endpoint.delete_service_parcel(profile_uuid, parcel_type._get_parcel_type(), parcel_uuid) + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[RoutePolicyParcel], + ) -> DataSequence[Parcel[RoutePolicyParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[Ipv4AclParcel], + ) -> DataSequence[Parcel[Ipv4AclParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[Ipv6AclParcel], + ) -> DataSequence[Parcel[Ipv6AclParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[LanVpnParcel], + ) -> DataSequence[Parcel[LanVpnParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[MulticastParcel], + ) -> DataSequence[Parcel[MulticastParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[EigrpParcel], + ) -> DataSequence[Parcel[EigrpParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[SwitchportParcel], + ) -> DataSequence[Parcel[SwitchportParcel]]: + ... + + @overload + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[WirelessLanParcel], + ) -> DataSequence[Parcel[WirelessLanParcel]]: + ... + + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[AnyServiceParcel], + ) -> DataSequence: + """ + Get all Service Parcels given profile id and parcel type + """ + return self.endpoint.get_all(profile_id, parcel_type._get_parcel_type()) + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[RoutePolicyParcel], + parcel_id: UUID, + ) -> Parcel[RoutePolicyParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[Ipv4AclParcel], + parcel_id: UUID, + ) -> Parcel[Ipv4AclParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[Ipv6AclParcel], + parcel_id: UUID, + ) -> Parcel[Ipv6AclParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[LanVpnParcel], + parcel_id: UUID, + ) -> Parcel[LanVpnParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[MulticastParcel], + parcel_id: UUID, + ) -> Parcel[MulticastParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[EigrpParcel], + parcel_id: UUID, + ) -> Parcel[EigrpParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[SwitchportParcel], + parcel_id: UUID, + ) -> Parcel[SwitchportParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[WirelessLanParcel], + parcel_id: UUID, + ) -> Parcel[WirelessLanParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[InterfaceEthernetParcel], + parcel_id: UUID, + vpn_uuid: UUID, + ) -> Parcel[InterfaceEthernetParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[InterfaceGreParcel], + parcel_id: UUID, + vpn_uuid: UUID, + ) -> Parcel[InterfaceGreParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[InterfaceIpsecParcel], + parcel_id: UUID, + vpn_uuid: UUID, + ) -> Parcel[InterfaceIpsecParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[InterfaceSviParcel], + parcel_id: UUID, + vpn_uuid: UUID, + ) -> Parcel[InterfaceSviParcel]: + ... + + @overload + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[InterfaceMultilinkParcel], + parcel_id: UUID, + vpn_uuid: UUID, + ) -> Parcel[InterfaceMultilinkParcel]: + ... + + def get_parcel( + self, + profile_id: UUID, + parcel_type: Type[AnyServiceParcel], + parcel_id: UUID, + vpn_uuid: Optional[UUID] = None, + ) -> Parcel: + """ + Get one Service Parcel given profile id, parcel type and parcel id + """ + if vpn_uuid is not None: + return self.endpoint.get_lan_vpn_sub_parcel( + profile_id, vpn_uuid, parcel_type._get_parcel_type().removeprefix("lan/vpn/"), parcel_id + ) + return self.endpoint.get_by_id(profile_id, parcel_type._get_parcel_type(), parcel_id) + + def update_parcel( + self, + profile_id: UUID, + parcel_type: Type[AnyServiceParcel], + parcel_id: UUID, + payload: AnyServiceParcel, + vpn_uuid: Optional[UUID] = None, + ) -> ParcelCreationResponse: + """ + Update Service Parcel for selected profile_id based on payload type + """ + if vpn_uuid is not None: + return self.endpoint.update_lan_vpn_sub_parcel( + profile_id, vpn_uuid, parcel_type._get_parcel_type().removeprefix("lan/vpn/"), parcel_id, payload + ) + return self.endpoint.update(profile_id, parcel_type._get_parcel_type(), parcel_id, payload) + class SystemFeatureProfileAPI: """ diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py index a8dfe840..5b538167 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py @@ -1,17 +1,22 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates # mypy: disable-error-code="empty-body" -from typing import Optional +from typing import Optional, Union from uuid import UUID -from catalystwan.endpoints import APIEndpoints, delete, get, post, versions +from catalystwan.endpoints import APIEndpoints, delete, get, post, put, versions from catalystwan.models.configuration.feature_profile.common import ( FeatureProfileCreationPayload, FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesParams, ) -from catalystwan.models.configuration.feature_profile.parcel import ParcelAssociationPayload, ParcelCreationResponse +from catalystwan.models.configuration.feature_profile.parcel import ( + Parcel, + ParcelAssociationPayload, + ParcelCreationResponse, +) +from catalystwan.models.configuration.feature_profile.sdwan.routing import AnyRoutingParcel from catalystwan.models.configuration.feature_profile.sdwan.service import ( AnyLanVpnInterfaceParcel, AnyTopLevelServiceParcel, @@ -42,7 +47,7 @@ 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_service_parcel( - self, profile_uuid: UUID, parcel_type: str, payload: AnyTopLevelServiceParcel + self, profile_uuid: UUID, parcel_type: str, payload: Union[AnyTopLevelServiceParcel, AnyRoutingParcel] ) -> ParcelCreationResponse: ... @@ -87,3 +92,38 @@ def associate_with_vpn( self, profile_id: UUID, vpn_id: UUID, parcel_type: str, payload: ParcelAssociationPayload ) -> ParcelCreationResponse: ... + + @versions(supported_versions=(">=20.9"), raises=False) + @put("/v1/feature-profile/sdwan/service/{profile_id}/{parcel_type}/{parcel_id}") + def update( + self, profile_id: UUID, parcel_type: str, parcel_id: UUID, payload: AnyTopLevelServiceParcel + ) -> ParcelCreationResponse: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @put("/v1/feature-profile/sdwan/service/{profile_id}/lan/vpn/{vpn_id}/{parcel_type}/{parcel_id}") + def update_lan_vpn_sub_parcel( + self, profile_id: UUID, vpn_id: UUID, parcel_type: str, parcel_id: UUID, payload: AnyLanVpnInterfaceParcel + ) -> ParcelCreationResponse: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/service/{profile_id}/{parcel_type}/{parcel_id}") + def get_by_id( + self, profile_id: UUID, parcel_type: str, parcel_id: UUID + ) -> Parcel[Union[AnyTopLevelServiceParcel, AnyRoutingParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/service/{profile_id}/{parcel_type}") + def get_all( + self, profile_id: UUID, parcel_type: str + ) -> DataSequence[Parcel[Union[AnyTopLevelServiceParcel, AnyRoutingParcel]]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/service/{profile_id}/lan/vpn/{vpn_id}/{parcel_type}/{parcel_id}") + def get_lan_vpn_sub_parcel( + self, profile_id: UUID, vpn_id: UUID, parcel_type: str, parcel_id: UUID + ) -> Parcel[AnyLanVpnInterfaceParcel]: + ... diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py index a31ba997..cb93f1ab 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py @@ -1,7 +1,7 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates # mypy: disable-error-code="empty-body" -from typing import Optional +from typing import Optional, Union from uuid import UUID from catalystwan.api.configuration_groups.parcel import _ParcelBase @@ -21,6 +21,7 @@ ParcelCreationResponse, ParcelId, ) +from catalystwan.models.configuration.feature_profile.sdwan.routing import AnyRoutingParcel from catalystwan.models.configuration.feature_profile.sdwan.transport import ( AnyTransportParcel, CellularControllerParcel, @@ -82,12 +83,37 @@ def create_management_vpn_sub_parcel( @versions(supported_versions=(">=20.12"), raises=False) @get("/v1/feature-profile/sdwan/transport/{profile_id}/{parcel_type}") - def get_transport_parcels(self, profile_id: UUID, parcel_type: str) -> DataSequence[Parcel[AnyTransportParcel]]: + def get_transport_parcels( + self, profile_id: UUID, parcel_type: str + ) -> DataSequence[Parcel[Union[AnyTransportParcel, AnyRoutingParcel]]]: ... @versions(supported_versions=(">=20.12"), raises=False) @get("/v1/feature-profile/sdwan/transport/{profile_id}/{parcel_type}/{parcel_id}") - def get_transport_parcel(self, profile_id: UUID, parcel_type: str, parcel_id: UUID) -> Parcel[AnyTransportParcel]: + def get_transport_parcel( + self, profile_id: UUID, parcel_type: str, parcel_id: UUID + ) -> Parcel[Union[AnyTransportParcel, AnyRoutingParcel]]: + ... + + @versions(supported_versions=(">=20.12"), raises=False) + @put("/v1/feature-profile/sdwan/transport/{profile_id}/{parcel_type}/{parcel_id}") + def update_transport_parcel( + self, profile_id: UUID, parcel_type: str, parcel_id: UUID, payload: _ParcelBase + ) -> ParcelCreationResponse: + ... + + @versions(supported_versions=(">=20.12"), raises=False) + @put("/v1/feature-profile/sdwan/transport/{profile_id}/wan/vpn/{vpn_id}/{parcel_type}/{parcel_id}") + def update_transport_vpn_sub_parcel( + self, profile_id: UUID, vpn_id: UUID, parcel_type: str, parcel_id: UUID, payload: _ParcelBase + ) -> ParcelCreationResponse: + ... + + @versions(supported_versions=(">=20.12"), raises=False) + @put("/v1/feature-profile/sdwan/transport/{profile_id}/management/vpn/{vpn_id}/{parcel_type}/{parcel_id}") + def update_management_vpn_sub_parcel( + self, profile_id: UUID, vpn_id: UUID, parcel_type: str, parcel_id: UUID, payload: _ParcelBase + ) -> ParcelCreationResponse: ... # diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py index 7b847bbc..d8458ddb 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/test_service.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_service.py @@ -128,6 +128,27 @@ def setUpClass(cls) -> None: cls.api = cls.session.api.sdwan_feature_profiles.service cls.profile_uuid = cls.api.create_profile(create_name_with_run_id("TestProfileService"), "Description").id + def test_update(self): + 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"), + ), + ) + parcel_id = self.api.create_parcel(self.profile_uuid, dhcp_server_parcel).id + parcel = self.api.get_parcel(self.profile_uuid, LanVpnDhcpServerParcel, parcel_id) + assert isinstance(parcel.payload, LanVpnDhcpServerParcel) + assert parcel.payload == dhcp_server_parcel + + dhcp_server_parcel.address_pool.network_address = Global[IPv4Address](value=IPv4Address("10.3.2.1")) + dhcp_server_parcel.address_pool.subnet_mask = Global[SubnetMask](value="248.0.0.0") + parcel_id = self.api.update_parcel(self.profile_uuid, LanVpnDhcpServerParcel, parcel_id, dhcp_server_parcel).id + parcel = self.api.get_parcel(self.profile_uuid, LanVpnDhcpServerParcel, parcel_id) + assert isinstance(parcel.payload, LanVpnDhcpServerParcel) + assert parcel.payload == dhcp_server_parcel + def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): # Arrange dhcp_server_parcel = LanVpnDhcpServerParcel( @@ -141,7 +162,9 @@ def test_when_default_values_dhcp_server_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, dhcp_server_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, LanVpnDhcpServerParcel, parcel_id) + assert isinstance(parcel.payload, LanVpnDhcpServerParcel) + assert parcel.payload == dhcp_server_parcel def test_when_default_values_service_vpn_parcel_expect_successful_post(self): # Arrange @@ -187,7 +210,9 @@ def test_when_default_values_service_vpn_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, vpn_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, LanVpnParcel, parcel_id) + assert isinstance(parcel.payload, LanVpnParcel) + assert parcel.payload == vpn_parcel def test_when_default_values_ospf_parcel_expect_successful_post(self): # Arrange @@ -198,7 +223,9 @@ def test_when_default_values_ospf_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ospf_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, RoutingOspfParcel, parcel_id) + assert isinstance(parcel.payload, RoutingOspfParcel) + assert parcel.payload == ospf_parcel def test_when_default_ospfv3_ipv4_expect_successful_post(self): # Arrange @@ -215,7 +242,9 @@ def test_when_default_ospfv3_ipv4_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ospfv3ipv4_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, RoutingOspfv3IPv4Parcel, parcel_id) + assert isinstance(parcel.payload, RoutingOspfv3IPv4Parcel) + assert parcel.payload == ospfv3ipv4_parcel def test_when_default_ospfv3_ipv6_expect_successful_post(self): # Arrange @@ -232,7 +261,9 @@ def test_when_default_ospfv3_ipv6_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ospfv3ipv6_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, RoutingOspfv3IPv6Parcel, parcel_id) + assert isinstance(parcel.payload, RoutingOspfv3IPv6Parcel) + assert parcel.payload == ospfv3ipv6_parcel def test_when_default_values_eigrp_parcel_expect_successful_post(self): eigrp_parcel = EigrpParcel( @@ -253,7 +284,9 @@ def test_when_default_values_eigrp_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, eigrp_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, EigrpParcel, parcel_id) + assert isinstance(parcel.payload, EigrpParcel) + assert parcel.payload == eigrp_parcel def test_when_default_values_route_policy_parcel_expect_successful_post(self): # Arrange @@ -264,7 +297,9 @@ def test_when_default_values_route_policy_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, route_policy_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, RoutePolicyParcel, parcel_id) + assert isinstance(parcel.payload, RoutePolicyParcel) + assert parcel.payload == route_policy_parcel def test_when_default_values_acl_ipv4_expect_successful_post(self): # Arrange @@ -275,7 +310,9 @@ def test_when_default_values_acl_ipv4_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, Ipv4AclParcel, parcel_id) + assert isinstance(parcel.payload, Ipv4AclParcel) + assert parcel.payload == acl_ipv4_parcel def test_when_fully_specified_acl_ipv4_expect_successful_post(self): # Arrange @@ -300,7 +337,9 @@ def test_when_fully_specified_acl_ipv4_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, Ipv4AclParcel, parcel_id) + assert isinstance(parcel.payload, Ipv4AclParcel) + assert parcel.payload == acl_ipv4_parcel def test_when_default_values_acl_ipv6_expect_successful_post(self): # Arrange @@ -311,7 +350,9 @@ def test_when_default_values_acl_ipv6_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv6_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, Ipv6AclParcel, parcel_id) + assert isinstance(parcel.payload, Ipv6AclParcel) + assert parcel.payload == acl_ipv6_parcel def test_when_fully_specified_acl_ipv6_expect_successful_post(self): # Arrange @@ -333,7 +374,9 @@ def test_when_fully_specified_acl_ipv6_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv6_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, Ipv6AclParcel, parcel_id) + assert isinstance(parcel.payload, Ipv6AclParcel) + assert parcel.payload == acl_ipv6_parcel def test_when_fully_specified_routing_bgp_expect_successful_post(self): # Arrange @@ -457,7 +500,9 @@ def test_when_fully_specified_routing_bgp_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, bgp_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, RoutingBgpParcel, parcel_id) + assert isinstance(parcel.payload, RoutingBgpParcel) + assert parcel.payload == bgp_parcel def test_when_correct_values_switchport_parcel_expect_successful_post(self): # Arrange @@ -503,7 +548,10 @@ def test_when_correct_values_switchport_parcel_expect_successful_post(self): with self.subTest(switchport_parcel=switchport_parcel.parcel_name): parcel_id = self.api.create_parcel(self.profile_uuid, switchport_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, SwitchportParcel, parcel_id) + assert isinstance(parcel.payload, SwitchportParcel) + assert parcel.payload == switchport_parcel + # Cleanup self.api.delete_parcel(self.profile_uuid, SwitchportParcel, parcel_id) @@ -517,7 +565,9 @@ def test_when_default_values_multicast_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, multicast_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, MulticastParcel, parcel_id) + assert isinstance(parcel.payload, MulticastParcel) + assert parcel.payload == multicast_parcel def test_when_fully_specified_values_multicast_expect_successful_post(self): # Arrange @@ -610,7 +660,9 @@ def test_when_fully_specified_values_multicast_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, multicast_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, MulticastParcel, parcel_id) + assert isinstance(parcel.payload, MulticastParcel) + assert parcel.payload == multicast_parcel def test_when_fully_specified_values_wireless_lan_expect_successful_post(self): # Arrange @@ -650,7 +702,9 @@ def test_when_fully_specified_values_wireless_lan_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, wireless_lan_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, WirelessLanParcel, parcel_id) + assert isinstance(parcel.payload, WirelessLanParcel) + assert parcel.payload == wireless_lan_parcel @classmethod def tearDownClass(cls) -> None: @@ -673,6 +727,28 @@ def setUpClass(cls) -> None: ), ).id + def test_update_interface(self): + # Arrange + svi_parcel = InterfaceSviParcel( + parcel_name="TestSviParcel", + parcel_description="Test Svi Parcel", + interface_name=as_global("Vlan1"), + svi_description=as_global("Test Svi Description"), + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, svi_parcel, self.vpn_parcel_uuid).id + parcel = self.api.get_parcel(self.profile_uuid, InterfaceSviParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, InterfaceSviParcel) + assert parcel.payload == svi_parcel + # Assert + svi_parcel.svi_description = as_global("Updated Svi Description") + parcel_id = self.api.update_parcel( + self.profile_uuid, InterfaceSviParcel, parcel_id, svi_parcel, vpn_uuid=self.vpn_parcel_uuid + ).id + parcel = self.api.get_parcel(self.profile_uuid, InterfaceSviParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, InterfaceSviParcel) + assert parcel.payload == svi_parcel + def test_when_default_values_gre_parcel_expect_successful_post(self): # Arrange gre_parcel = InterfaceGreParcel( @@ -690,7 +766,9 @@ def test_when_default_values_gre_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, gre_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, InterfaceGreParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, InterfaceGreParcel) + assert parcel.payload == gre_parcel def test_when_default_values_svi_parcel_expect_successful_post(self): # Arrange @@ -703,7 +781,9 @@ def test_when_default_values_svi_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, svi_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, InterfaceSviParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, InterfaceSviParcel) + assert parcel.payload == svi_parcel def test_when_default_values_ethernet_parcel_expect_successful_post(self): # Arrange @@ -716,7 +796,11 @@ def test_when_default_values_ethernet_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_dynamic_interface_ip_address_for_ethernet_parcel(self): # Arrange @@ -729,7 +813,11 @@ def test_set_dynamic_interface_ip_address_for_ethernet_parcel(self): parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_dynamic_interface_ip_address_as_variable_for_ethernet_parcel(self): # Arrange @@ -742,7 +830,11 @@ def test_set_dynamic_interface_ip_address_as_variable_for_ethernet_parcel(self): parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_primary_static_interface_ip_address_for_ethernet_parcel(self): # Arrange @@ -755,7 +847,11 @@ def test_set_primary_static_interface_ip_address_for_ethernet_parcel(self): parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_primary_static_interface_ip_address_as_variable_for_ethernet_parcel(self): # Arrange @@ -768,7 +864,11 @@ def test_set_primary_static_interface_ip_address_as_variable_for_ethernet_parcel parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_primary_static_with_mask_interface_ip_address_for_ethernet_parcel(self): # Arrange @@ -782,7 +882,11 @@ def test_set_primary_static_with_mask_interface_ip_address_for_ethernet_parcel(s parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_set_primary_static_with_mask_interface_ip_address_as_varialbles_for_ethernet_parcel(self): # Arrange @@ -798,7 +902,11 @@ def test_set_primary_static_with_mask_interface_ip_address_as_varialbles_for_eth parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel def test_when_default_values_ipsec_parcel_expect_successful_post(self): # Arrange @@ -818,7 +926,9 @@ def test_when_default_values_ipsec_parcel_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ipsec_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, InterfaceIpsecParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, InterfaceIpsecParcel) + assert parcel.payload == ipsec_parcel def test_when_routing_parcel_and_vpn_uuid_present_expect_create_then_assign_to_vpn(self): # Arrange @@ -830,7 +940,9 @@ def test_when_routing_parcel_and_vpn_uuid_present_expect_create_then_assign_to_v # Act parcel_id = self.api.create_parcel(self.profile_uuid, multicast_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, MulticastParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid) + assert isinstance(parcel.payload, MulticastParcel) + assert parcel.payload == multicast_parcel def test_when_fully_specified_multilink_interface_parcel_expect_successful_post(self): nim_list = [ @@ -925,7 +1037,11 @@ def test_when_fully_specified_multilink_interface_parcel_expect_successful_post( parcel_id = self.api.create_parcel(self.profile_uuid, multilink_parcel, self.vpn_parcel_uuid).id - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceMultilinkParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceMultilinkParcel) + assert parcel.payload == multilink_parcel def test_when_default_values_ethernet_interface_expect_successful_post(self): # For 20.13 when we send "advanced.intfruMtu" as Default 1500 @@ -944,7 +1060,11 @@ def test_when_default_values_ethernet_interface_expect_successful_post(self): # Act parcel_id = self.api.create_parcel(self.profile_uuid, ethernet_parcel, self.vpn_parcel_uuid).id # Assert - assert parcel_id + parcel = self.api.get_parcel( + self.profile_uuid, InterfaceEthernetParcel, parcel_id, vpn_uuid=self.vpn_parcel_uuid + ) + assert isinstance(parcel.payload, InterfaceEthernetParcel) + assert parcel.payload == ethernet_parcel @classmethod def tearDownClass(cls) -> None: diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_transport.py b/catalystwan/integration_tests/feature_profile/sdwan/test_transport.py index a2e17846..6728689d 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/test_transport.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/test_transport.py @@ -71,6 +71,7 @@ RedistributeItem, RoutingBgpParcel, ) +from catalystwan.models.configuration.feature_profile.sdwan.routing.ospf import RoutingOspfParcel from catalystwan.models.configuration.feature_profile.sdwan.transport.management.ethernet import ( Advanced as ManagementEthernetAdvanced, ) @@ -183,6 +184,19 @@ def setUpClass(cls) -> None: cls.api = cls.session.api.sdwan_feature_profiles.transport cls.profile_uuid = cls.api.create_profile(create_name_with_run_id("TestTransportModels"), "Description").id + def test_when_default_values_ospf_expect_successful_post(self): + # Arrange + ospf_parcel = RoutingOspfParcel( + parcel_name="TestOspfParcel-Defaults", + parcel_description="Test Ospf Parcel", + ) + # Act + parcel_id = self.api.create_parcel(self.profile_uuid, ospf_parcel).id + # Assert + parcel = self.api.get_parcel(self.profile_uuid, RoutingOspfParcel, parcel_id) + assert isinstance(parcel.payload, RoutingOspfParcel) + assert parcel.payload == ospf_parcel + def test_when_fully_specified_management_vpn_parcel_expect_successful_post(self): # Arrange management_vpn_parcel = ManagementVpnParcel( @@ -230,7 +244,9 @@ def test_when_fully_specified_management_vpn_parcel_expect_successful_post(self) # Act parcel_id = self.api.create_parcel(self.profile_uuid, management_vpn_parcel).id # Assert - assert parcel_id + parcel = self.api.get_parcel(self.profile_uuid, ManagementVpnParcel, parcel_id) + assert isinstance(parcel.payload, ManagementVpnParcel) + assert parcel.payload == management_vpn_parcel def test_when_fully_specified_t1e1controller_type_e1_parcel_expect_successful_post(self): # Arrange diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/dhcp_server.py b/catalystwan/models/configuration/feature_profile/sdwan/service/dhcp_server.py index c045ee4e..c5f014fa 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/dhcp_server.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/dhcp_server.py @@ -89,7 +89,7 @@ class LanVpnDhcpServerParcel(_ParcelBase): validation_alias=AliasPath("data", "exclude"), description="Configure IPv4 address to exclude from DHCP address pool", ) - lease_time: Union[Global[int], Variable, Default[int]] = Field( + lease_time: Union[Default[int], Global[int], Variable] = Field( default=Default[int](value=86400), validation_alias=AliasPath("data", "leaseTime"), description="Configure how long a DHCP-assigned IP address is valid", 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 3e45606f..95f970ca 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/vpn.py @@ -351,7 +351,7 @@ class ServiceRoute(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) prefix: RoutePrefix - service: Union[Variable, Global[ServiceRouteType], Default[ServiceRouteType]] = Default[ServiceRouteType]( + service: Union[Default[ServiceRouteType], Variable, Global[ServiceRouteType]] = Default[ServiceRouteType]( value="SIG" ) vpn: Global[int] = Global[int](value=0)