Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

dev: add uc-voice model that were added in 20.14 #15

Merged
merged 5 commits into from
Dec 13, 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
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
TransportAndManagementProfileBuilder,
CliFeatureProfileBuilder,
ApplicationPriorityFeatureProfileBuilder,
UcVoiceFeatureProfileBuilder
]

BUILDER_MAPPING: Mapping[ProfileType, Callable] = {
Expand Down
201 changes: 167 additions & 34 deletions catalystwan/api/builders/feature_profiles/uc_voice.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,36 @@

import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, List, Optional
from typing import TYPE_CHECKING, Dict, List, Optional, Union
from uuid import UUID

from catalystwan.api.builders.feature_profiles.report import FeatureProfileBuildReport, handle_build_report
from catalystwan.api.configuration_groups.parcel import as_default
from catalystwan.api.feature_profile_api import UcVoiceFeatureProfileAPI
from catalystwan.endpoints.configuration.feature_profile.sdwan.uc_voice import UcVoiceFeatureProfile
from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload, RefIdItem
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice import (
AnalogInterfaceParcel,
AnyUcVoiceParcel,
CallRoutingParcel,
DigitalInterfaceParcel,
MediaProfileParcel,
ServerGroupParcel,
SrstParcel,
TranslationProfileParcel,
TranslationRuleParcel,
TrunkGroupParcel,
VoiceGlobalParcel,
VoiceTenantParcel,
)
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.analog_interface import (
Association as AnalogInterfaceAssociation,
)
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.call_routing import (
Association as CallRoutingAssociation,
)
from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.digital_interface import (
Association as DigitalInterfaceAssociation,
)

logger = logging.getLogger(__name__)
Expand All @@ -22,95 +41,209 @@
from catalystwan.session import ManagerSession


ParcelWithAssociations = Union[CallRoutingParcel, DigitalInterfaceParcel, AnalogInterfaceParcel]
Association = Union[List[DigitalInterfaceAssociation], List[AnalogInterfaceAssociation], List[CallRoutingAssociation]]


@dataclass
class TranslationProfile:
tpp: TranslationProfileParcel
calling: Optional[TranslationRuleParcel] = None
called: Optional[TranslationRuleParcel] = None


def is_uuid(uuid: Optional[Union[str, UUID]]) -> bool:
if isinstance(uuid, UUID) or uuid is None:
return True
try:
UUID(uuid)
return True
except ValueError:
return False


class UcVoiceFeatureProfileBuilder:
"""
A class for building UcVoice feature profiles.
A class for building UC Voice feature profiles with a modular approach.

This class provides methods to construct, associate, and manage various parcels and
configurations required for UC Voice feature profiles.
"""

ASSOCIABLE_PARCELS = (
MediaProfileParcel,
ServerGroupParcel,
SrstParcel,
TranslationProfileParcel,
TranslationRuleParcel,
TrunkGroupParcel,
VoiceGlobalParcel,
VoiceTenantParcel,
)

ASSOCIATON_FIELDS = {
"media_profile",
"server_group",
"translation_profile",
"trunk_group",
"voice_tenant",
"supervisory_disconnect",
}

def __init__(self, session: ManagerSession) -> None:
"""
Initialize a new instance of the Service class.
Initializes a new instance of the UC Voice Feature Profile Builder.

Args:
session (ManagerSession): The ManagerSession object used for API communication.
profile_uuid (UUID): The UUID of the profile.
session (ManagerSession): The session object used for API communication.
"""
self._profile: FeatureProfileCreationPayload
self._api = UcVoiceFeatureProfileAPI(session)
self._endpoints = UcVoiceFeatureProfile(session)
self._independent_items: List[AnyUcVoiceParcel] = []
self._independent_parcels: List[AnyUcVoiceParcel] = []
self._translation_profiles: List[TranslationProfile] = []
self._pushed_associable_parcels: Dict[str, UUID] = {} # Maps parcel names to their created UUIDs.
self._parcels_with_associations: List[ParcelWithAssociations] = []

def add_profile_name_and_description(self, feature_profile: FeatureProfileCreationPayload) -> None:
"""
Adds a name and description to the feature profile.
Adds a name and description to the feature profile being built.

Args:
name (str): The name of the feature profile.
description (str): The description of the feature profile.

Returns:
None
feature_profile (FeatureProfileCreationPayload): The feature profile payload containing
the name and description.
"""
self._profile = feature_profile

def add_parcel(self, parcel: AnyUcVoiceParcel) -> None:
def add_independent_parcel(self, parcel: AnyUcVoiceParcel) -> None:
"""
Adds a parcel to the feature profile.
Adds an independent parcel to the feature profile.

Args:
parcel (AnySystemParcel): The parcel to add.

Returns:
None
parcel (AnyUcVoiceParcel): The parcel to be added. Parcels are independent configurations
that do not require associations with other parcels.
"""
self._independent_items.append(parcel)
self._independent_parcels.append(parcel)

def add_translation_profile(
self,
tpp: TranslationProfileParcel,
calling: Optional[TranslationRuleParcel] = None,
called: Optional[TranslationRuleParcel] = None,
):
) -> None:
"""
Adds a translation profile to the feature profile.

Args:
tpp (TranslationProfileParcel): The main translation profile parcel.
calling (Optional[TranslationRuleParcel]): The calling rule parcel. Optional.
called (Optional[TranslationRuleParcel]): The called rule parcel. Optional.

Raises:
ValueError: If neither a calling nor a called rule is provided.
"""
if not calling and not called:
raise ValueError("There must be at least one translation rule to create a translation profile")
raise ValueError("There must be at least one translation rule to create a translation profile.")
self._translation_profiles.append(TranslationProfile(tpp=tpp, called=called, calling=calling))

def add_parcel_with_associations(self, parcel: ParcelWithAssociations) -> None:
"""
Adds a parcel with associations to the feature profile.

Args:
parcel (ParcelWithAssociations): A parcel that includes associations with other entities.
Associations are relationships between parcels and other resources.
"""
self._parcels_with_associations.append(parcel)

def build(self) -> FeatureProfileBuildReport:
"""
Builds the feature profile.
Builds the complete UC Voice feature profile.

This method creates the feature profile on the system and processes all added parcels
and translation profiles, resolving associations as needed.

Returns:
UUID: The UUID of the created feature profile.
FeatureProfileBuildReport: A report containing the details of the created feature profile.
"""

profile_uuid = self._endpoints.create_uc_voice_feature_profile(self._profile).id
self.build_report = FeatureProfileBuildReport(profile_uuid=profile_uuid, profile_name=self._profile.name)
for parcel in self._independent_items:
self._create_parcel(profile_uuid, parcel)

# Create independent parcels
for ip in self._independent_parcels:
parcel_uuid = self._create_parcel(profile_uuid, ip)
if parcel_uuid and isinstance(ip, self.ASSOCIABLE_PARCELS):
self._pushed_associable_parcels[ip.parcel_name] = parcel_uuid

# Create translation profiles
for tp in self._translation_profiles:
self._create_translation_profile(profile_uuid, tp)
parcel_uuid = self._create_translation_profile(profile_uuid, tp)
if parcel_uuid:
self._pushed_associable_parcels[tp.tpp.parcel_name] = parcel_uuid

# Create parcels with associations
for pwa in self._parcels_with_associations:
if pwa.association:
self._populate_association(pwa.association)
self._create_parcel(profile_uuid, pwa)

return self.build_report

@handle_build_report
def _create_parcel(self, profile_uuid: UUID, parcel: AnyUcVoiceParcel) -> UUID:
"""
Internal method to create a parcel.

Args:
profile_uuid (UUID): The UUID of the feature profile being built.
parcel (AnyUcVoiceParcel): The parcel to create.

Returns:
UUID: The UUID of the created parcel.
"""
return self._api.create_parcel(profile_uuid, parcel).id

def _create_translation_profile(self, profile_uuid: UUID, tp: TranslationProfile):
def _create_translation_profile(self, profile_uuid: UUID, tp: TranslationProfile) -> UUID:
"""
Internal method to create a translation profile.

Args:
profile_uuid (UUID): The UUID of the feature profile being built.
tp (TranslationProfile): The translation profile to create.

Returns:
UUID: The UUID of the created translation profile parcel.
"""
if tp.called:
called_uuid = self._create_parcel(profile_uuid, tp.called)
if called_uuid:
if called_uuid := self._create_parcel(profile_uuid, tp.called):
tp.tpp.set_ref_by_call_type(called_uuid, "called")
if tp.calling:
calling_uuid = self._create_parcel(profile_uuid, tp.calling)
if calling_uuid:
if calling_uuid := self._create_parcel(profile_uuid, tp.calling):
tp.tpp.set_ref_by_call_type(calling_uuid, "calling")
self._create_parcel(profile_uuid, tp.tpp)
return self._create_parcel(profile_uuid, tp.tpp)

def _populate_association(self, association: Association) -> None:
"""
Resolves associations for a parcel.

Updates references in the parcel's associations to use the actual UUIDs of previously
created parcels.

Args:
association (Association): A list of associations to resolve.
"""
for model in association:
for field_name in self.ASSOCIATON_FIELDS.intersection(model.model_fields_set):
attr = getattr(model, field_name)
if isinstance(attr, RefIdItem):
if is_uuid(attr.ref_id.value) or attr.ref_id.value is None:
continue
resolved_uuid = self._pushed_associable_parcels.get(attr.ref_id.value)
if resolved_uuid:
attr.ref_id.value = str(resolved_uuid)
else:
logger.warning(
f"Unresolved reference in field '{field_name}' with value '{attr.ref_id.value}' "
f"for model '{model.__class__.__name__}'. Setting to Default[None]."
)
attr.ref_id = as_default(None)
21 changes: 18 additions & 3 deletions catalystwan/api/feature_profile_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -2259,10 +2259,24 @@ class UcVoiceFeatureProfileAPI:
SDWAN Feature Profile UC Voice APIs
"""

ENDPOINT_PARCEL_TYPE_MAP = {"analog-interface": "tdm-sip/analog-interface"}

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

def _get_endpoint_parcel_type(self, parcel_type: str) -> str:
"""
Returns the mapped endpoint parcel type if it exists, otherwise returns the input key.

Args:
parcel_type (str): The parcel type to look up.

Returns:
str: The mapped parcel type or the input key if not found.
"""
return self.ENDPOINT_PARCEL_TYPE_MAP.get(parcel_type, parcel_type)

def get_profiles(
self, limit: Optional[int] = None, offset: Optional[int] = None
) -> DataSequence[FeatureProfileInfo]:
Expand Down Expand Up @@ -2360,14 +2374,15 @@ def get_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)
return self.endpoint.get_by_id(
profile_id, self._get_endpoint_parcel_type(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)
return self.endpoint.create(profile_id, self._get_endpoint_parcel_type(payload._get_parcel_type()), payload)

def update_parcel(self, profile_id: UUID, payload: AnyUcVoiceParcel, parcel_id: UUID) -> ParcelCreationResponse:
"""
Expand Down
2 changes: 1 addition & 1 deletion catalystwan/integration_tests/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,9 @@ def load_config() -> dict:
raise CatalystwanException("Missing environment variables")
return dict(
url=url,
port=port,
username=username,
password=password,
port=port,
)


Expand Down
Loading
Loading