From 47c96e99347556942e37fecfa0c80bbb299daeff Mon Sep 17 00:00:00 2001 From: Jakub Krajewski <95274389+jpkrajewski@users.noreply.github.com> Date: Thu, 25 Apr 2024 15:42:59 +0200 Subject: [PATCH] dev-uxmt: System Feature Profile API type hinting (#44) * Changes to feature profile system api. Add generic typing to endpoints. Move parcel classes to separate file * fix: pre commit issues * fullConfigCli field is optional * fix imports after rebase * fix tests after rebase * remove print statements --------- Co-authored-by: sbasan --- catalystwan/api/feature_profile_api.py | 109 ++++++++------- .../feature_profile/sdwan/other.py | 3 +- .../feature_profile/sdwan/policy_object.py | 2 +- .../feature_profile/sdwan/service.py | 3 +- .../feature_profile/sdwan/system.py | 130 +++++++++++++++++- .../feature_profile/sdwan/transport.py | 3 +- catalystwan/endpoints/configuration_group.py | 4 +- .../models/configuration/config_migration.py | 62 +++++---- .../configuration/feature_profile/common.py | 100 +------------- .../configuration/feature_profile/parcel.py | 125 +++++++++++++++++ catalystwan/tests/test_feature_profile_api.py | 14 +- .../converters/feature_template/ospfv3.py | 3 - endpoints-md.py | 21 ++- 13 files changed, 379 insertions(+), 200 deletions(-) create mode 100644 catalystwan/models/configuration/feature_profile/parcel.py diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index 17499b22..aa08e294 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/catalystwan/api/feature_profile_api.py @@ -29,6 +29,8 @@ FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesPayload, +) +from catalystwan.models.configuration.feature_profile.parcel import ( Parcel, ParcelAssociationPayload, ParcelCreationResponse, @@ -357,56 +359,56 @@ def get_parcels( def get_parcels( self, profile_id: UUID, - parcel_type: Type[BFDParcel], - ) -> DataSequence[Parcel[BFDParcel]]: + parcel_type: Type[BannerParcel], + ) -> DataSequence[Parcel[BannerParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[LoggingParcel], - ) -> DataSequence[Parcel[LoggingParcel]]: + parcel_type: Type[BasicParcel], + ) -> DataSequence[Parcel[BasicParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[BannerParcel], - ) -> DataSequence[Parcel[BannerParcel]]: + parcel_type: Type[BFDParcel], + ) -> DataSequence[Parcel[BFDParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[BasicParcel], - ) -> DataSequence[Parcel[BasicParcel]]: + parcel_type: Type[GlobalParcel], + ) -> DataSequence[Parcel[GlobalParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[GlobalParcel], - ) -> DataSequence[Parcel[GlobalParcel]]: + parcel_type: Type[LoggingParcel], + ) -> DataSequence[Parcel[LoggingParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[NTPParcel], - ) -> DataSequence[Parcel[NTPParcel]]: + parcel_type: Type[MRFParcel], + ) -> DataSequence[Parcel[MRFParcel]]: ... @overload def get_parcels( self, profile_id: UUID, - parcel_type: Type[MRFParcel], - ) -> DataSequence[Parcel[MRFParcel]]: + parcel_type: Type[NTPParcel], + ) -> DataSequence[Parcel[NTPParcel]]: ... @overload @@ -433,119 +435,124 @@ def get_parcels( ) -> DataSequence[Parcel[SNMPParcel]]: ... - # get by id + def get_parcels( + self, + profile_id: UUID, + parcel_type: Type[AnySystemParcel], + ) -> DataSequence: + """ + Get all System Parcels given profile id and parcel type + """ + return self.endpoint.get_all(profile_id, parcel_type._get_parcel_type()) @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, parcel_type: Type[AAAParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[AAAParcel]]: + ) -> Parcel[AAAParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[BFDParcel], + parcel_type: Type[BannerParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[BFDParcel]]: + ) -> Parcel[BannerParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[LoggingParcel], + parcel_type: Type[BasicParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[LoggingParcel]]: + ) -> Parcel[BasicParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[BannerParcel], + parcel_type: Type[BFDParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[BannerParcel]]: + ) -> Parcel[BFDParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[BasicParcel], + parcel_type: Type[GlobalParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[BasicParcel]]: + ) -> Parcel[GlobalParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[GlobalParcel], + parcel_type: Type[LoggingParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[GlobalParcel]]: + ) -> Parcel[LoggingParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[NTPParcel], + parcel_type: Type[MRFParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[NTPParcel]]: + ) -> Parcel[MRFParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, - parcel_type: Type[MRFParcel], + parcel_type: Type[NTPParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[MRFParcel]]: + ) -> Parcel[NTPParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, parcel_type: Type[OMPParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[OMPParcel]]: + ) -> Parcel[OMPParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, parcel_type: Type[SecurityParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[SecurityParcel]]: + ) -> Parcel[SecurityParcel]: ... @overload - def get_parcels( + def get_parcel( self, profile_id: UUID, parcel_type: Type[SNMPParcel], parcel_id: UUID, - ) -> DataSequence[Parcel[SNMPParcel]]: + ) -> Parcel[SNMPParcel]: ... - def get_parcels( + def get_parcel( self, profile_id: UUID, parcel_type: Type[AnySystemParcel], - parcel_id: Union[UUID, None] = None, - ) -> DataSequence[Parcel[Any]]: + parcel_id: UUID, + ) -> Parcel: """ - Get all System Parcels for selected profile_id and selected type or get one System Parcel given parcel id + Get one System Parcel given profile id, parcel type and parcel id """ - - if not parcel_id: - return self.endpoint.get_all(profile_id, parcel_type._get_parcel_type()) return self.endpoint.get_by_id(profile_id, parcel_type._get_parcel_type(), parcel_id) def create_parcel(self, profile_id: UUID, payload: AnySystemParcel) -> ParcelCreationResponse: diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/other.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/other.py index d297b82d..56cff5bc 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/other.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/other.py @@ -11,9 +11,8 @@ FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesPayload, - Parcel, - ParcelId, ) +from catalystwan.models.configuration.feature_profile.parcel import Parcel, ParcelId from catalystwan.typed_list import DataSequence diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/policy_object.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/policy_object.py index f1b3b219..c7e6fb04 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/policy_object.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/policy_object.py @@ -4,7 +4,7 @@ from uuid import UUID from catalystwan.endpoints import APIEndpoints, delete, get, post, put, versions -from catalystwan.models.configuration.feature_profile.common import Parcel, ParcelCreationResponse +from catalystwan.models.configuration.feature_profile.parcel import Parcel, ParcelCreationResponse from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel from catalystwan.typed_list import DataSequence diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py index 1b3bf387..bfa418fb 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/service.py @@ -10,9 +10,8 @@ FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesPayload, - ParcelAssociationPayload, - ParcelCreationResponse, ) +from catalystwan.models.configuration.feature_profile.parcel import ParcelAssociationPayload, ParcelCreationResponse from catalystwan.models.configuration.feature_profile.sdwan.service import ( AnyLanVpnInterfaceParcel, AnyTopLevelServiceParcel, diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py index d9243200..ecd9f17e 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py @@ -11,11 +11,21 @@ FeatureProfileCreationResponse, FeatureProfileInfo, GetFeatureProfilesPayload, - Parcel, - ParcelId, SchemaTypeQuery, ) +from catalystwan.models.configuration.feature_profile.parcel import Parcel, ParcelId from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.aaa import AAAParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.banner import BannerParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.basic import BasicParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.bfd import BFDParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.global_parcel import GlobalParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.logging_parcel import LoggingParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.mrf import MRFParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.ntp import NTPParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.omp import OMPParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.security import SecurityParcel +from catalystwan.models.configuration.feature_profile.sdwan.system.snmp import SNMPParcel from catalystwan.typed_list import DataSequence @@ -55,13 +65,123 @@ def edit_aaa_profile_parcel_for_system(self, profile_id: UUID, parcel_id: UUID, ... @versions(supported_versions=(">=20.9"), raises=False) - @get("/v1/feature-profile/sdwan/system/{profile_id}/{parcel_type}") - def get_all(self, profile_id: UUID, parcel_type: UUID) -> DataSequence[Parcel]: + @get("/v1/feature-profile/sdwan/system/{profile_id}/{parcel_type}", resp_json_key="data") + def get_all(self, profile_id: UUID, parcel_type: UUID) -> DataSequence[Parcel[AnySystemParcel]]: ... @versions(supported_versions=(">=20.9"), raises=False) @get("/v1/feature-profile/sdwan/system/{profile_id}/{parcel_type}/{parcel_id}") - def get_by_id(self, profile_id: UUID, parcel_type: str, parcel_id: UUID) -> Parcel: + def get_by_id(self, profile_id: UUID, parcel_type: str, parcel_id: UUID) -> Parcel[AnySystemParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/aaa", resp_json_key="data") + def get_all_aaa(self, profile_id: UUID) -> DataSequence[Parcel[AAAParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/aaa/{parcel_id}") + def get_by_id_aaa(self, profile_id: UUID, parcel_id: UUID) -> Parcel[AAAParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/banner", resp_json_key="data") + def get_all_banner(self, profile_id: UUID) -> DataSequence[Parcel[BannerParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/banner/{parcel_id}") + def get_by_id_banner(self, profile_id: UUID, parcel_id: UUID) -> Parcel[BannerParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/basic", resp_json_key="data") + def get_all_basic(self, profile_id: UUID) -> DataSequence[Parcel[BasicParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/basic/{parcel_id}") + def get_by_id_basic(self, profile_id: UUID, parcel_id: UUID) -> Parcel[BasicParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/bfd", resp_json_key="data") + def get_all_bfd(self, profile_id: UUID) -> DataSequence[Parcel[BFDParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/bfd/{parcel_id}") + def get_by_id_bfd(self, profile_id: UUID, parcel_id: UUID) -> Parcel[BFDParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/global", resp_json_key="data") + def get_all_global(self, profile_id: UUID) -> DataSequence[Parcel[GlobalParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/global/{parcel_id}") + def get_by_id_global(self, profile_id: UUID, parcel_id: UUID) -> Parcel[GlobalParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/logging", resp_json_key="data") + def get_all_logging(self, profile_id: UUID) -> DataSequence[Parcel[LoggingParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/logging/{parcel_id}") + def get_by_id_logging(self, profile_id: UUID, parcel_id: UUID) -> Parcel[LoggingParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/mrf", resp_json_key="data") + def get_all_mrf(self, profile_id: UUID) -> DataSequence[Parcel[MRFParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/mrf/{parcel_id}") + def get_by_id_mrf(self, profile_id: UUID, parcel_id: UUID) -> Parcel[MRFParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/ntp", resp_json_key="data") + def get_all_ntp(self, profile_id: UUID) -> DataSequence[Parcel[NTPParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/ntp/{parcel_id}") + def get_by_id_ntp(self, profile_id: UUID, parcel_id: UUID) -> Parcel[NTPParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/omp", resp_json_key="data") + def get_all_omp(self, profile_id: UUID) -> DataSequence[Parcel[OMPParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/omp/{parcel_id}") + def get_by_id_omp(self, profile_id: UUID, parcel_id: UUID) -> Parcel[OMPParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/security", resp_json_key="data") + def get_all_security(self, profile_id: UUID) -> DataSequence[Parcel[SecurityParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/security/{parcel_id}") + def get_by_id_security(self, profile_id: UUID, parcel_id: UUID) -> Parcel[SecurityParcel]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/snmp", resp_json_key="data") + def get_all_snmp(self, profile_id: UUID) -> DataSequence[Parcel[SNMPParcel]]: + ... + + @versions(supported_versions=(">=20.9"), raises=False) + @get("/v1/feature-profile/sdwan/system/{profile_id}/snmp/{parcel_id}") + def get_by_id_snmp(self, profile_id: UUID, parcel_id: UUID) -> Parcel[SNMPParcel]: ... @versions(supported_versions=(">=20.9"), raises=False) diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py index 3a3b1fe8..a54bccd5 100644 --- a/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py +++ b/catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py @@ -12,10 +12,9 @@ FeatureProfileEditPayload, FeatureProfileInfo, GetFeatureProfilesPayload, - ParcelCreationResponse, - ParcelId, SchemaTypeQuery, ) +from catalystwan.models.configuration.feature_profile.parcel import ParcelCreationResponse, ParcelId from catalystwan.models.configuration.feature_profile.sdwan.transport import ( AnyTransportParcel, CellularControllerParcel, diff --git a/catalystwan/endpoints/configuration_group.py b/catalystwan/endpoints/configuration_group.py index 8419675f..0a236a9c 100644 --- a/catalystwan/endpoints/configuration_group.py +++ b/catalystwan/endpoints/configuration_group.py @@ -59,7 +59,9 @@ class ConfigGroup(BaseModel): ) origin: Optional[str] = None topology: Optional[str] = None - full_config_cli: bool = Field(serialization_alias="fullConfigCli", validation_alias="fullConfigCli") + full_config_cli: Optional[bool] = Field( + default=None, serialization_alias="fullConfigCli", validation_alias="fullConfigCli" + ) class ConfigGroupResponsePayload(BaseModel): diff --git a/catalystwan/models/configuration/config_migration.py b/catalystwan/models/configuration/config_migration.py index 25fa010b..d5fc2775 100644 --- a/catalystwan/models/configuration/config_migration.py +++ b/catalystwan/models/configuration/config_migration.py @@ -1,19 +1,14 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates -from typing import Any, Dict, List, Set, Tuple, Union +from typing import Any, Dict, List, Set, Tuple from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, model_validator -from typing_extensions import Annotated from catalystwan.api.templates.device_template.device_template import DeviceTemplate, GeneralTemplate from catalystwan.endpoints.configuration_group import ConfigGroupCreationPayload from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload, ProfileType -from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel -from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel -from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel -from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel -from catalystwan.models.configuration.feature_profile.sdwan.transport import AnyTransportParcel +from catalystwan.models.configuration.feature_profile.parcel import AnyParcel from catalystwan.models.configuration.topology_group import TopologyGroup from catalystwan.models.policy import AnyPolicyDefinitionInfo, AnyPolicyListInfo from catalystwan.models.policy.centralized import CentralizedPolicyInfo @@ -21,11 +16,6 @@ from catalystwan.models.policy.security import AnySecurityPolicyInfo from catalystwan.models.templates import FeatureTemplateInformation, TemplateInformation -AnyParcel = Annotated[ - Union[AnySystemParcel, AnyPolicyObjectParcel, AnyServiceParcel, AnyOtherParcel, AnyTransportParcel], - Field(discriminator="type_"), -] - class DeviceTemplateWithInfo(DeviceTemplate): model_config = ConfigDict(populate_by_name=True) @@ -65,16 +55,24 @@ def get_flattened_general_templates(self) -> List[GeneralTemplate]: class UX1Policies(BaseModel): model_config = ConfigDict(populate_by_name=True) centralized_policies: List[CentralizedPolicyInfo] = Field( - default=[], serialization_alias="centralizedPolicies", validation_alias="centralizedPolicies" + default=[], + serialization_alias="centralizedPolicies", + validation_alias="centralizedPolicies", ) localized_policies: List[LocalizedPolicyInfo] = Field( - default=[], serialization_alias="localizedPolicies", validation_alias="localizedPolicies" + default=[], + serialization_alias="localizedPolicies", + validation_alias="localizedPolicies", ) security_policies: List[AnySecurityPolicyInfo] = Field( - default=[], serialization_alias="securityPolicies", validation_alias="securityPolicies" + default=[], + serialization_alias="securityPolicies", + validation_alias="securityPolicies", ) policy_definitions: List[AnyPolicyDefinitionInfo] = Field( - default=[], serialization_alias="policyDefinitions", validation_alias="policyDefinitions" + default=[], + serialization_alias="policyDefinitions", + validation_alias="policyDefinitions", ) policy_lists: List[AnyPolicyListInfo] = Field( default=[], serialization_alias="policyLists", validation_alias="policyLists" @@ -83,10 +81,14 @@ class UX1Policies(BaseModel): class UX1Templates(BaseModel): feature_templates: List[FeatureTemplateInformation] = Field( - default=[], serialization_alias="featureTemplates", validation_alias="featureTemplates" + default=[], + serialization_alias="featureTemplates", + validation_alias="featureTemplates", ) device_templates: List[DeviceTemplateWithInfo] = Field( - default=[], serialization_alias="deviceTemplates", validation_alias="deviceTemplates" + default=[], + serialization_alias="deviceTemplates", + validation_alias="deviceTemplates", ) @@ -129,19 +131,27 @@ class UX2Config(BaseModel): # All UX2 Configuration items - Mega Model model_config = ConfigDict(populate_by_name=True) topology_groups: List[TransformedTopologyGroup] = Field( - default=[], serialization_alias="topologyGroups", validation_alias="topologyGroups" + default=[], + serialization_alias="topologyGroups", + validation_alias="topologyGroups", ) config_groups: List[TransformedConfigGroup] = Field( - default=[], serialization_alias="configurationGroups", validation_alias="configurationGroups" + default=[], + serialization_alias="configurationGroups", + validation_alias="configurationGroups", ) policy_groups: List[TransformedConfigGroup] = Field( default=[], serialization_alias="policyGroups", validation_alias="policyGroups" ) feature_profiles: List[TransformedFeatureProfile] = Field( - default=[], serialization_alias="featureProfiles", validation_alias="featureProfiles" + default=[], + serialization_alias="featureProfiles", + validation_alias="featureProfiles", ) profile_parcels: List[TransformedParcel] = Field( - default=[], serialization_alias="profileParcels", validation_alias="profileParcels" + default=[], + serialization_alias="profileParcels", + validation_alias="profileParcels", ) @model_validator(mode="before") @@ -159,10 +169,14 @@ def insert_parcel_type_from_headers(cls, values: Dict[str, Any]): class UX2ConfigRollback(BaseModel): config_group_ids: List[UUID] = Field( - default_factory=list, serialization_alias="ConfigGroupIds", validation_alias="ConfigGroupIds" + default_factory=list, + serialization_alias="ConfigGroupIds", + validation_alias="ConfigGroupIds", ) feature_profile_ids: List[Tuple[UUID, ProfileType]] = Field( - default_factory=list, serialization_alias="FeatureProfileIds", validation_alias="FeatureProfileIds" + default_factory=list, + serialization_alias="FeatureProfileIds", + validation_alias="FeatureProfileIds", ) def add_config_group(self, config_group_id: UUID) -> None: diff --git a/catalystwan/models/configuration/feature_profile/common.py b/catalystwan/models/configuration/feature_profile/common.py index 10afa1f5..94ef4a59 100644 --- a/catalystwan/models/configuration/feature_profile/common.py +++ b/catalystwan/models/configuration/feature_profile/common.py @@ -2,7 +2,7 @@ from datetime import datetime from ipaddress import IPv4Address -from typing import Generic, List, Literal, Optional, TypeVar, Union +from typing import List, Literal, Optional, Union from uuid import UUID from pydantic import BaseModel, ConfigDict, Field @@ -10,61 +10,9 @@ from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, as_global from catalystwan.models.configuration.common import Solution -T = TypeVar("T") - - IPV4Address = str IPv6Address = str -ParcelType = Literal[ - "appqoe", - "as-path", - "lan/vpn", - "lan/vpn/interface/ethernet", - "lan/vpn/interface/gre", - "lan/vpn/interface/ipsec", - "lan/vpn/interface/svi", - "dhcp-server", - "tracker", - "trackergroup", - "routing/bgp", - "routing/eigrp", - "routing/multicast", - "routing/ospf", - "routing/ospfv3/ipv4", - "routing/ospfv3/ipv6", - "wirelesslan", - "switchport", - "app-probe", - "app-list", - "color", - "data-prefix", - "expanded-community", - "class", - "data-ipv6-prefix", - "ipv6-prefix", - "prefix", - "policer", - "preferred-color-group", - "sla-class", - "tloc", - "standard-community", - "security-localdomain", - "security-fqdn", - "security-ipssignature", - "security-urllist", - "security-urllist", - "security-port", - "security-protocolname", - "security-geolocation", - "security-zone", - "security-localapp", - "security-data-ip-prefix", - "unified/advanced-malware-protection", - "unified/intrusion-prevention", - "unified/url-filtering", -] - ProfileType = Literal[ "transport", "system", @@ -129,41 +77,6 @@ class FeatureProfileCreationResponse(BaseModel): id: UUID -class ParcelCreationResponse(BaseModel): - model_config = ConfigDict(populate_by_name=True) - - id: UUID = Field(serialization_alias="parcelId", validation_alias="parcelId") - - -class Parcel(BaseModel, Generic[T]): - parcel_id: str = Field(alias="parcelId") - parcel_type: ParcelType = Field(alias="parcelType") - created_by: str = Field(alias="createdBy") - last_updated_by: str = Field(alias="lastUpdatedBy") - created_on: int = Field(alias="createdOn") - last_updated_on: int = Field(alias="lastUpdatedOn") - payload: T - - -class Header(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - generated_on: int = Field(alias="generatedOn") - - -class ParcelInfo(BaseModel, Generic[T]): - model_config = ConfigDict(arbitrary_types_allowed=True) - - header: Header - data: List[Parcel[T]] - - -class ParcelAssociationPayload(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) - - parcel_id: UUID = Field(serialization_alias="parcelId", validation_alias="parcelId") - - class Prefix(BaseModel): address: Union[Variable, Global[str], Global[IPv4Address], Global[IPv6Address]] mask: Union[Variable, Global[str]] @@ -175,22 +88,11 @@ class SchemaTypeQuery(BaseModel): schema_type: SchemaType = Field(alias="schemaType") -class ParcelId(BaseModel): - id: str = Field(alias="parcelId") - - class GetFeatureProfilesPayload(BaseModel): limit: Optional[int] offset: Optional[int] -class ParcelSequence(BaseModel, Generic[T]): - model_config = ConfigDict(arbitrary_types_allowed=True) - - header: Header - data: List[Parcel[T]] - - class DNSIPv4(BaseModel): primary_dns_address_ipv4: Union[Default[None], Global[str], Variable] = Field( default=Default[None](value=None), alias="primaryDnsAddressIpv4" diff --git a/catalystwan/models/configuration/feature_profile/parcel.py b/catalystwan/models/configuration/feature_profile/parcel.py new file mode 100644 index 00000000..0067b23e --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/parcel.py @@ -0,0 +1,125 @@ +from typing import Generic, List, Literal, TypeVar, Union +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, model_validator +from typing_extensions import Annotated + +from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel +from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel +from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel +from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel + +ParcelType = Literal[ + "aaa", + "banner", + "basic", + "bfd", + "global", + "logging", + "mrf", + "ntp", + "omp", + "security", + "snmp", + "appqoe", + "lan/vpn", + "lan/vpn/interface/ethernet", + "lan/vpn/interface/gre", + "lan/vpn/interface/ipsec", + "lan/vpn/interface/svi", + "dhcp-server", + "tracker", + "trackergroup", + "routing/bgp", + "routing/eigrp", + "routing/multicast", + "routing/ospf", + "routing/ospfv3/ipv4", + "routing/ospfv3/ipv6", + "wirelesslan", + "switchport", + "app-probe", + "app-list", + "color", + "data-prefix", + "expanded-community", + "class", + "data-ipv6-prefix", + "ipv6-prefix", + "prefix", + "policer", + "preferred-color-group", + "sla-class", + "tloc", + "standard-community", + "security-localdomain", + "security-fqdn", + "security-ipssignature", + "security-urllist", + "security-urllist", + "security-port", + "security-protocolname", + "security-geolocation", + "security-zone", + "security-localapp", + "security-data-ip-prefix", +] + + +AnyParcel = Annotated[ + Union[AnySystemParcel, AnyPolicyObjectParcel, AnyServiceParcel, AnyOtherParcel], + Field(discriminator="type_"), +] + +T = TypeVar("T", bound=AnyParcel) + + +class Parcel(BaseModel, Generic[T]): + parcel_id: str = Field(alias="parcelId") + parcel_type: ParcelType = Field(alias="parcelType") + created_by: str = Field(alias="createdBy") + last_updated_by: str = Field(alias="lastUpdatedBy") + created_on: int = Field(alias="createdOn") + last_updated_on: int = Field(alias="lastUpdatedOn") + payload: T + + @model_validator(mode="before") + def validate_payload(cls, data): + data["payload"]["type_"] = data["parcelType"] + return data + + +class Header(BaseModel): + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + + generated_on: int = Field(alias="generatedOn") + + +class ParcelInfo(BaseModel, Generic[T]): + model_config = ConfigDict(arbitrary_types_allowed=True) + + header: Header + data: List[Parcel[T]] + + +class ParcelSequence(BaseModel, Generic[T]): + model_config = ConfigDict(arbitrary_types_allowed=True) + + header: Header + data: List[Parcel[T]] + + +class ParcelCreationResponse(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + id: UUID = Field(serialization_alias="parcelId", validation_alias="parcelId") + + +class ParcelAssociationPayload(BaseModel): + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) + + parcel_id: UUID = Field(serialization_alias="parcelId", validation_alias="parcelId") + + +class ParcelId(BaseModel): + id: str = Field(alias="parcelId") diff --git a/catalystwan/tests/test_feature_profile_api.py b/catalystwan/tests/test_feature_profile_api.py index 29a727ed..e543327b 100644 --- a/catalystwan/tests/test_feature_profile_api.py +++ b/catalystwan/tests/test_feature_profile_api.py @@ -14,7 +14,7 @@ from catalystwan.endpoints.configuration.feature_profile.sdwan.service import ServiceFeatureProfile from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile from catalystwan.endpoints.configuration.feature_profile.sdwan.transport import TransportFeatureProfile -from catalystwan.models.configuration.feature_profile.common import ParcelAssociationPayload, ParcelCreationResponse +from catalystwan.models.configuration.feature_profile.parcel import ParcelAssociationPayload, ParcelCreationResponse from catalystwan.models.configuration.feature_profile.sdwan.service import ( AppqoeParcel, InterfaceEthernetParcel, @@ -84,7 +84,7 @@ def test_delete_method_with_valid_arguments(self, parcel, expected_path): @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) + self.api.get_parcel(self.profile_uuid, parcel, self.parcel_uuid) # Assert self.mock_endpoint.get_by_id.assert_called_once_with(self.profile_uuid, expected_path, self.parcel_uuid) @@ -136,7 +136,10 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path): 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"))), + basic=BasicGre( + if_name=as_global("gre1"), + tunnel_destination=as_global(IPv4Address("4.4.4.4")), + ), ), ), ( @@ -230,7 +233,10 @@ def test_post_method_create_then_assigin_subparcel(self, parcel_type, parcel): # Assert self.mock_endpoint.create_service_parcel.assert_called_once_with(self.profile_uuid, parcel_type, parcel) self.mock_endpoint.associate_parcel_with_vpn.assert_called_once_with( - self.profile_uuid, self.vpn_uuid, parcel_type, ParcelAssociationPayload(parcel_id=self.parcel_uuid) + self.profile_uuid, + self.vpn_uuid, + parcel_type, + ParcelAssociationPayload(parcel_id=self.parcel_uuid), ) diff --git a/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py b/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py index 35fa39a6..3905eb0e 100644 --- a/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py +++ b/catalystwan/utils/config_migration/converters/feature_template/ospfv3.py @@ -228,7 +228,6 @@ def configure_redistribute(self, values: dict) -> None: return None redistribute_list = [] for redistribute in redistributes: - print(redistribute) redistribute_list.append( RedistributedRoute( protocol=as_global(redistribute.get("protocol").value, RedistributeProtocol), @@ -251,7 +250,6 @@ def _set_range(self, area_value: dict) -> Optional[List[SummaryRouteIPv6]]: return None range_list = [] for range_ in ranges: - print(range_) range_list.append( SummaryRouteIPv6( network=range_.get("address"), @@ -267,7 +265,6 @@ def configure_redistribute(self, values: dict) -> None: return None redistribute_list = [] for redistribute in redistributes: - print(redistribute) if redistribute.get("protocol").value not in get_args(RedistributeProtocolIPv6): continue redistribute_list.append( diff --git a/endpoints-md.py b/endpoints-md.py index 0c55ef87..b8174a9b 100644 --- a/endpoints-md.py +++ b/endpoints-md.py @@ -12,6 +12,7 @@ from urllib.request import pathname2url from packaging.specifiers import SpecifierSet # type: ignore +from typing_extensions import get_original_bases from catalystwan import __package__ from catalystwan.endpoints import BASE_PATH, APIEndpointRequestMeta, TypeSpecifier, request, versions, view @@ -110,12 +111,20 @@ def from_type_specifier(typespec: TypeSpecifier) -> CompositeTypeLink: if payloadtype.__module__ == "builtins": return CompositeTypeLink.text_only(payloadtype.__name__) elif sourcefile := getsourcefile(payloadtype): - return CompositeTypeLink( - link_text=payloadtype.__name__, - sourcefile=create_sourcefile_link(sourcefile), - lineno=getsourcelines(payloadtype)[1], - origin=generate_origin_string(typespec), - ) + try: + lineno = getsourcelines(payloadtype)[1] + link_text = payloadtype.__name__ + except OSError: + base, *_ = get_original_bases(payloadtype) + lineno = getsourcelines(base)[1] + link_text = base.__name__ + finally: + return CompositeTypeLink( + link_text=link_text, + sourcefile=create_sourcefile_link(sourcefile), + lineno=lineno, + origin=generate_origin_string(typespec), + ) return CompositeTypeLink.text_only("") def md(self) -> str: