Skip to content
This repository has been archived by the owner on Nov 21, 2024. It is now read-only.

dev: support device config items for 20.13 #841

Merged
merged 20 commits into from
Oct 27, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions catalystwan/api/builders/feature_profiles/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
from catalystwan.endpoints.configuration.feature_profile.sdwan.cli import CliFeatureProfile
from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
from catalystwan.models.configuration.feature_profile.sdwan.cli import AnyCliParcel
from catalystwan.models.configuration.feature_profile.sdwan.cli.config import ConfigParcel

if TYPE_CHECKING:
from catalystwan.session import ManagerSession
Expand All @@ -30,7 +29,7 @@ def __init__(self, session: ManagerSession) -> None:
self._profile: FeatureProfileCreationPayload
self._api = CliFeatureProfileAPI(session)
self._endpoints = CliFeatureProfile(session)
self._cli_configs: List[ConfigParcel] = []
self._cli_configs: List[AnyCliParcel] = list()

def add_profile_name_and_description(self, feature_profile: FeatureProfileCreationPayload) -> None:
"""
Expand Down
184 changes: 165 additions & 19 deletions catalystwan/api/feature_profile_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +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.endpoints.configuration.feature_profile.sdwan.uc_voice import UcVoiceFeatureProfile
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
Expand Down Expand Up @@ -106,6 +107,10 @@
InterfaceEthPPPoEParcel,
)
from catalystwan.models.configuration.feature_profile.sdwan.transport.wan.interface.t1e1serial import T1E1SerialParcel
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice import AnyUcVoiceParcel
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.dsp_farm import DspFarmParcel
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.media_profile import MediaProfileParcel
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.trunk_group import TrunkGroupParcel
from catalystwan.typed_list import DataSequence

if TYPE_CHECKING:
Expand All @@ -131,7 +136,7 @@
ParcelAssociationPayload,
ParcelCreationResponse,
)
from catalystwan.models.configuration.feature_profile.sdwan.cli import ConfigParcel
from catalystwan.models.configuration.feature_profile.sdwan.cli import AnyCliParcel
from catalystwan.models.configuration.feature_profile.sdwan.dns_security import AnyDnsSecurityParcel, DnsParcel
from catalystwan.models.configuration.feature_profile.sdwan.embedded_security import (
AnyEmbeddedSecurityParcel,
Expand Down Expand Up @@ -193,17 +198,18 @@ def __init__(self, session: ManagerSession):

class SDWANFeatureProfilesAPI:
def __init__(self, session: ManagerSession):
self.policy_object = PolicyObjectFeatureProfileAPI(session=session)
self.system = SystemFeatureProfileAPI(session=session)
self.application_priority = ApplicationPriorityFeatureProfileAPI(session=session)
self.cli = CliFeatureProfileAPI(session=session)
self.dns_security = DnsSecurityFeatureProfileAPI(session=session)
self.embedded_security = EmbeddedSecurityFeatureProfileAPI(session=session)
self.other = OtherFeatureProfileAPI(session=session)
self.policy_object = PolicyObjectFeatureProfileAPI(session=session)
self.service = ServiceFeatureProfileAPI(session=session)
self.sig_security = SIGSecurityAPI(session=session)
self.system = SystemFeatureProfileAPI(session=session)
self.topology = TopologyFeatureProfileAPI(session=session)
self.transport = TransportFeatureProfileAPI(session=session)
self.embedded_security = EmbeddedSecurityFeatureProfileAPI(session=session)
self.cli = CliFeatureProfileAPI(session=session)
self.dns_security = DnsSecurityFeatureProfileAPI(session=session)
self.sig_security = SIGSecurityAPI(session=session)
self.application_priority = ApplicationPriorityFeatureProfileAPI(session=session)
self.uc_voice = UcVoiceFeatureProfileAPI(session=session)


class FeatureProfileAPI(Protocol):
Expand Down Expand Up @@ -1856,23 +1862,34 @@ def get_parcel_by_id(
self,
profile_id: UUID,
parcel_id: UUID,
) -> DataSequence[Parcel[Any]]:
parcel_type: str = "config",
) -> Parcel[AnyCliParcel]:
"""
Get all CLI Parcels for selected profile_id and selected type or get one CLI Parcel given parcel id
"""
return self.endpoint.get_by_id(profile_id, parcel_id)
return self.endpoint.get_by_id(profile_id, parcel_type, parcel_id)

def create_parcel(self, profile_id: UUID, config: ConfigParcel) -> ParcelCreationResponse:
def get_parcel(
self,
profile_id: UUID,
parcel_type: str = "config",
) -> DataSequence[Parcel[AnyCliParcel]]:
"""
Get all CLI Parcels for selected profile_id and selected type or get one CLI Parcel given parcel id
"""
return self.endpoint.get_all(profile_id, parcel_type)

def create_parcel(self, profile_id: UUID, config: AnyCliParcel) -> ParcelCreationResponse:
"""
Create CLI Parcel for selected profile_id
"""
return self.endpoint.create(profile_id, config)
return self.endpoint.create(profile_id, config._get_parcel_type(), config)

def update_parcel(self, profile_id: UUID, config: ConfigParcel, parcel_id: UUID) -> ParcelCreationResponse:
def update_parcel(self, profile_id: UUID, config: AnyCliParcel, parcel_id: UUID) -> ParcelCreationResponse:
"""
Update CLI Parcel for selected profile_id
"""
return self.endpoint.update(profile_id, parcel_id, config)
return self.endpoint.update(profile_id, config._get_parcel_type(), parcel_id, config)

def delete_parcel(self, profile_id: UUID, parcel_id: UUID) -> None:
"""
Expand Down Expand Up @@ -1987,7 +2004,7 @@ def delete_profile(self, profile_id: UUID) -> None:
"""
Delete SIG Security Feature Profile
"""
self.endpoint.delete_sig_security_feature_profile(profile_id)
self.endpoint.delete_sig_security_feature_profile(str(profile_id))

def delete_all_profiles(self) -> None:
"""
Expand All @@ -2001,13 +2018,13 @@ def create_parcel(self, profile_uuid: UUID, payload: SIGParcel) -> ParcelCreatio
"""
Create SIG Security Parcel for selected profile_id
"""
return self.endpoint.create_sig_security_parcel(profile_uuid, payload)
return self.endpoint.create_sig_security_parcel(str(profile_uuid), payload)

def delete_parcel(self, profile_uuid: UUID, parcel_uuid: UUID) -> None:
"""
Delete Service Parcel for selected profile_uuid based on payload type
"""
return self.endpoint.delete_sig_security_parcel(profile_uuid, parcel_uuid)
return self.endpoint.delete_sig_security_parcel(str(profile_uuid), str(parcel_uuid))


class ApplicationPriorityFeatureProfileAPI:
Expand Down Expand Up @@ -2040,7 +2057,7 @@ def delete_profile(self, profile_id: UUID) -> None:
"""
Delete Application Priority Feature Profile
"""
self.endpoint.delete_application_priority_feature_profile(profile_id)
self.endpoint.delete_application_priority_feature_profile(str(profile_id))

def delete_all_profiles(self) -> None:
"""
Expand Down Expand Up @@ -2174,7 +2191,7 @@ def delete_profile(self, profile_id: UUID) -> None:
"""
Delete Topology Feature Profile
"""
self.endpoint.delete_topology_feature_profile(profile_id)
self.endpoint.delete_topology_feature_profile(str(profile_id))

def delete_all_profiles(self) -> None:
"""
Expand Down Expand Up @@ -2235,3 +2252,132 @@ def get_parcel(
Get one Topology Parcel given profile id, parcel type and parcel id
"""
return self.endpoint.get_any_parcel_by_id(profile_id, parcel_type._get_parcel_type(), parcel_id)


class UcVoiceFeatureProfileAPI:
"""
SDWAN Feature Profile UC Voice APIs
"""

def __init__(self, session: ManagerSession):
self.session = session
self.endpoint = UcVoiceFeatureProfile(session)

def get_profiles(
self, limit: Optional[int] = None, offset: Optional[int] = None
) -> DataSequence[FeatureProfileInfo]:
"""
Get all UC Voice Feature Profiles
"""
payload = GetFeatureProfilesParams(limit=limit if limit else None, offset=offset if offset else None)

return self.endpoint.get_uc_voice_feature_profiles(payload)

def create_profile(self, name: str, description: str) -> FeatureProfileCreationResponse:
"""
Create UC Voice Feature Profile
"""
payload = FeatureProfileCreationPayload(name=name, description=description)
return self.endpoint.create_uc_voice_feature_profile(payload)

def delete_profile(self, profile_id: UUID) -> None:
"""
Delete UC Voice Feature Profile
"""
self.endpoint.delete_uc_voice_feature_profile(str(profile_id))

def delete_all_profiles(self) -> None:
"""
Delete all UC Voice Feature Profiles
"""
profiles = self.get_profiles()
for profile in profiles:
self.delete_profile(profile.profile_id)

@overload
def get_parcels(
self,
profile_id: UUID,
parcel_type: Type[DspFarmParcel],
) -> DataSequence[Parcel[DspFarmParcel]]:
...

@overload
def get_parcels(
self,
profile_id: UUID,
parcel_type: Type[MediaProfileParcel],
) -> DataSequence[Parcel[MediaProfileParcel]]:
...

@overload
def get_parcels(
self,
profile_id: UUID,
parcel_type: Type[TrunkGroupParcel],
) -> DataSequence[Parcel[TrunkGroupParcel]]:
...

def get_parcels(self, profile_id: UUID, parcel_type: Type[AnyUcVoiceParcel]) -> DataSequence[Parcel]:
"""
Get all UC Voice 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[DspFarmParcel],
parcel_id: UUID,
) -> Parcel[DspFarmParcel]:
...

@overload
def get_parcel(
self,
profile_id: UUID,
parcel_type: Type[MediaProfileParcel],
parcel_id: UUID,
) -> Parcel[MediaProfileParcel]:
...

@overload
def get_parcel(
self,
profile_id: UUID,
parcel_type: Type[TrunkGroupParcel],
parcel_id: UUID,
) -> Parcel[TrunkGroupParcel]:
...

def get_parcel(
self,
profile_id: UUID,
parcel_type: Type[AnyUcVoiceParcel],
parcel_id: UUID,
) -> Parcel:
"""
Get one UC Voice Parcel given profile id, parcel type and parcel id
"""
return self.endpoint.get_by_id(profile_id, parcel_type._get_parcel_type(), parcel_id)

def create_parcel(self, profile_id: UUID, payload: AnyUcVoiceParcel) -> ParcelCreationResponse:
"""
Create UC Voice Parcel for selected profile_id based on payload type
"""

return self.endpoint.create(profile_id, payload._get_parcel_type(), payload)

def update_parcel(self, profile_id: UUID, payload: AnyUcVoiceParcel, parcel_id: UUID) -> ParcelCreationResponse:
"""
Update UC Voice Parcel for selected profile_id based on payload type
"""

return self.endpoint.update(profile_id, payload._get_parcel_type(), parcel_id, payload)

def delete_parcel(self, profile_id: UUID, parcel_type: Type[AnyUcVoiceParcel], parcel_id: UUID) -> None:
"""
Delete UC Voice Parcel for selected profile_id based on payload type
"""
return self.endpoint.delete(profile_id, parcel_type._get_parcel_type(), parcel_id)
Loading