diff --git a/catalystwan/api/builders/feature_profiles/builder_factory.py b/catalystwan/api/builders/feature_profiles/builder_factory.py index fd39a9f2..d1b328da 100644 --- a/catalystwan/api/builders/feature_profiles/builder_factory.py +++ b/catalystwan/api/builders/feature_profiles/builder_factory.py @@ -9,6 +9,7 @@ from catalystwan.api.builders.feature_profiles.service import ServiceFeatureProfileBuilder from catalystwan.api.builders.feature_profiles.system import SystemFeatureProfileBuilder from catalystwan.api.builders.feature_profiles.transport import TransportAndManagementProfileBuilder +from catalystwan.api.builders.feature_profiles.uc_voice import UcVoiceFeatureProfileBuilder from catalystwan.exceptions import CatalystwanException from catalystwan.models.configuration.feature_profile.common import ProfileType @@ -22,6 +23,7 @@ TransportAndManagementProfileBuilder, CliFeatureProfileBuilder, ApplicationPriorityFeatureProfileBuilder, + UcVoiceFeatureProfileBuilder ] BUILDER_MAPPING: Mapping[ProfileType, Callable] = { @@ -31,6 +33,7 @@ "transport": TransportAndManagementProfileBuilder, "cli": CliFeatureProfileBuilder, "application-priority": ApplicationPriorityFeatureProfileBuilder, + "uc-voice": UcVoiceFeatureProfileBuilder, } diff --git a/catalystwan/api/builders/feature_profiles/uc_voice.py b/catalystwan/api/builders/feature_profiles/uc_voice.py new file mode 100644 index 00000000..449a83aa --- /dev/null +++ b/catalystwan/api/builders/feature_profiles/uc_voice.py @@ -0,0 +1,249 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from __future__ import annotations + +import logging +from dataclasses import dataclass +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, 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__) + +if TYPE_CHECKING: + 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 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: + """ + Initializes a new instance of the UC Voice Feature Profile Builder. + + Args: + session (ManagerSession): The session object used for API communication. + """ + self._profile: FeatureProfileCreationPayload + self._api = UcVoiceFeatureProfileAPI(session) + self._endpoints = UcVoiceFeatureProfile(session) + 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 being built. + + Args: + feature_profile (FeatureProfileCreationPayload): The feature profile payload containing + the name and description. + """ + self._profile = feature_profile + + def add_independent_parcel(self, parcel: AnyUcVoiceParcel) -> None: + """ + Adds an independent parcel to the feature profile. + + Args: + parcel (AnyUcVoiceParcel): The parcel to be added. Parcels are independent configurations + that do not require associations with other parcels. + """ + 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.") + 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 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: + 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) + + # 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: + 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) -> 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: + if called_uuid := self._create_parcel(profile_uuid, tp.called): + tp.tpp.set_ref_by_call_type(called_uuid, "called") + if tp.calling: + if calling_uuid := self._create_parcel(profile_uuid, tp.calling): + tp.tpp.set_ref_by_call_type(calling_uuid, "calling") + 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) diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index c79a4365..ab18890a 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/catalystwan/api/feature_profile_api.py @@ -2259,10 +2259,26 @@ 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]: @@ -2360,14 +2376,13 @@ 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: """ diff --git a/catalystwan/integration_tests/base.py b/catalystwan/integration_tests/base.py index cb37f2cf..fe940fcb 100644 --- a/catalystwan/integration_tests/base.py +++ b/catalystwan/integration_tests/base.py @@ -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, ) diff --git a/catalystwan/integration_tests/profile_builder/test_pb_uc_voice.py b/catalystwan/integration_tests/profile_builder/test_pb_uc_voice.py new file mode 100644 index 00000000..8bced1e5 --- /dev/null +++ b/catalystwan/integration_tests/profile_builder/test_pb_uc_voice.py @@ -0,0 +1,89 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from catalystwan.api.configuration_groups.parcel import Global, as_global, as_variable +from catalystwan.integration_tests.base import TestCaseBase, create_name_with_run_id +from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload, RefIdItem +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice import ( + AnalogInterfaceParcel, + TranslationProfileParcel, + TranslationRuleParcel, +) +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.translation_rule import Action, RuleSettings +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.analog_interface import Association, SlotId, ModuleType +from catalystwan.tests.builders.uc_voice import as_default + + +class TestUcVoiceFeatureProfileBuilder(TestCaseBase): + def setUp(self) -> None: + self.fp_name = create_name_with_run_id("FeatureProfileBuilderUcVoice") + self.fp_description = "Transport feature profile" + self.builder = self.session.api.builders.feature_profiles.create_builder("uc-voice") + self.builder.add_profile_name_and_description( + feature_profile=FeatureProfileCreationPayload(name=self.fp_name, description=self.fp_description) + ) + self.api = self.session.api.sdwan_feature_profiles.transport + self.tp = TranslationProfileParcel(parcel_name="TPP", parcel_description="TTP_Desc", translation_profile_settings=[]) + self.tr_calling = TranslationRuleParcel( + parcel_name="2", + parcel_description="desc", + rule_name=Global[int](value=2), + rule_settings=[ + RuleSettings( + action=Global[Action](value="replace"), + match=Global[str](value="/123/"), + replacement_pattern=Global[str](value="/444/"), + rule_num=Global[int](value=2), + ) + ], + ) + self.tr_called = TranslationRuleParcel( + parcel_name="4", + parcel_description="desc", + rule_name=Global[int](value=4), + rule_settings=[ + RuleSettings( + action=Global[Action](value="replace"), + match=Global[str](value="/321/"), + replacement_pattern=Global[str](value="/4445/"), + rule_num=Global[int](value=4), + ) + ], + ) + + def test_when_build_profile_with_translation_profile_and_rules_expect_success(self): + # Arrange + self.builder.add_translation_profile(self.tp, self.tr_calling, self.tr_called) + # Act + report = self.builder.build() + # Assert + assert len(report.failed_parcels) == 0 + + def test_when_build_profile_with_analog_interface_and_translation_profile_and_rules_assosiations_expect_success(self): + # Arrange + ai = AnalogInterfaceParcel( + parcel_name="Ai", + parcel_description="", + enable=as_default(True), + slot_id=as_global("0/1", SlotId), + module_type=as_global("72 Port FXS", ModuleType), + association=[ + Association( + port_range=as_variable("{{test}}"), + translation_profile=RefIdItem(ref_id=as_global("TPP")), + trunk_group=RefIdItem(ref_id=as_default(None)), + trunk_group_priority=as_default(None), + translation_rule_direction=as_default(None) + ) + ] + ) + self.builder.add_translation_profile(self.tp, self.tr_calling, self.tr_called) + self.builder.add_parcel_with_associations(ai) + # Act + report = self.builder.build() + # Assert + assert len(report.failed_parcels) == 0 + + def tearDown(self) -> None: + target_profile = self.api.get_profiles().filter(profile_name=self.fp_name).single_or_default() + if target_profile: + # In case of a failed test, the profile might not have been created + self.api.delete_profile(target_profile.profile_id) diff --git a/catalystwan/models/configuration/feature_profile/common.py b/catalystwan/models/configuration/feature_profile/common.py index b1d30422..1b76e3e7 100644 --- a/catalystwan/models/configuration/feature_profile/common.py +++ b/catalystwan/models/configuration/feature_profile/common.py @@ -295,7 +295,7 @@ class RefIdItem(BaseModel): extra="forbid", populate_by_name=True, ) - ref_id: Global[str] = Field(..., serialization_alias="refId", validation_alias="refId") + ref_id: Union[Global[str], Default[None]] = Field(..., serialization_alias="refId", validation_alias="refId") @classmethod def from_uuid(cls, ref_id: UUID): diff --git a/catalystwan/models/configuration/feature_profile/parcel.py b/catalystwan/models/configuration/feature_profile/parcel.py index b1c0663b..1fc6a819 100644 --- a/catalystwan/models/configuration/feature_profile/parcel.py +++ b/catalystwan/models/configuration/feature_profile/parcel.py @@ -24,6 +24,7 @@ ParcelType = Literal[ "aaa", + "analog-interface", "app-list", "app-probe", "appqoe", @@ -32,6 +33,7 @@ "basic", "bfd", "bgp", + "call-routing", "cellular-controller", "cellular-profile", "cflowd", @@ -42,6 +44,7 @@ "data-ipv6-prefix", "data-prefix", "dhcp-server", + "digital-interface", "dns", "dsp-farm", "expanded-community", @@ -93,10 +96,13 @@ "security-urllist", "security-zone", "security", + "server-group", "sig", "sla-class", "snmp", "standard-community", + "strt", + "supervisory-disconnect", "switchport", "t1-e1-controller", "tloc", @@ -111,7 +117,8 @@ "unified/ssl-decryption-profile", "unified/ssl-decryption", "unified/url-filtering", - "wan/vpn", + "voice-global", + "voice-tenant" "wan/vpn", "wan/vpn/interface/cellular", "wan/vpn/interface/dsl-ipoe", "wan/vpn/interface/dsl-pppoa", diff --git a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py index e2753c7b..de91257a 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/service/lan/ethernet.py @@ -138,18 +138,18 @@ def serialize(self, handler: SerializerFunctionWrapHandler, info: SerializationI class Trustsec(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid") - enable_sgt_propagation: Union[Global[bool], Default[bool]] = Field( + enable_sgt_propogation: Union[Global[bool], Default[bool]] = Field( serialization_alias="enableSGTPropogation", validation_alias="enableSGTPropogation", default=Default[bool](value=False), ) - propagate: Annotated[ + propogate: Annotated[ Optional[Union[Global[bool], Default[bool]]], VersionedField(versions="<=20.12", forbidden=True) ] = Default[bool](value=True) security_group_tag: Optional[Union[Global[int], Variable, Default[None]]] = Field( serialization_alias="securityGroupTag", validation_alias="securityGroupTag", default=None ) - enable_enforced_propagation: Union[Global[bool], Default[None]] = Field( + enable_enforced_propogation: Union[Global[bool], Default[None]] = Field( default=Default[None](value=None), serialization_alias="enableEnforcedPropogation", validation_alias="enableEnforcedPropogation", diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/__init__.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/__init__.py index 82d04f45..7b8777f8 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/__init__.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/__init__.py @@ -5,25 +5,51 @@ from pydantic import Field from typing_extensions import Annotated +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.analog_interface import AnalogInterfaceParcel +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.call_routing import CallRoutingParcel +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.digital_interface import DigitalInterfaceParcel +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.server_group import ServerGroupParcel +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.srst import SrstParcel from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.translation_profile import TranslationProfileParcel from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.translation_rule import TranslationRuleParcel from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.trunk_group import TrunkGroupParcel - -from .dsp_farm import DspFarmParcel +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.voice_global import VoiceGlobalParcel +from catalystwan.models.configuration.feature_profile.sdwan.uc_voice.voice_tenant import VoiceTenantParcel AnyUcVoiceParcel = Annotated[ - Union[DspFarmParcel, MediaProfileParcel, TrunkGroupParcel, TranslationRuleParcel, TranslationProfileParcel], + Union[ + AnalogInterfaceParcel, + CallRoutingParcel, + DigitalInterfaceParcel, + DspFarmParcel, + MediaProfileParcel, + ServerGroupParcel, + SrstParcel, + TranslationProfileParcel, + TranslationRuleParcel, + TrunkGroupParcel, + VoiceGlobalParcel, + VoiceTenantParcel, + ], Field(discriminator="type_"), ] __all__ = ( + "AnalogInterfaceParcel", "AnyUcVoiceParcel", + "CallRoutingParcel", + "DigitalInterfaceParcel", "DspFarmParcel", "MediaProfileParcel", - "TrunkGroupParcel", + "ServerGroupParcel", + "SrstParcel", "TranslationProfileParcel", "TranslationRuleParcel", + "TrunkGroupParcel", + "VoiceGlobalParcel", + "VoiceTenantParcel", ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/analog_interface.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/analog_interface.py new file mode 100644 index 00000000..ef0b09ee --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/analog_interface.py @@ -0,0 +1,438 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase +from catalystwan.models.configuration.feature_profile.common import RefIdItem + +SlotId = Literal[ + "0/1", + "0/2", + "0/3", + "1/0", + "1/1", + "2/0", + "2/1", + "3/0", +] + +ModuleType = Literal[ + "12 Port FXO", + "16 Port FXS", + "2 Port FXO", + "2 Port FXS", + "24 Port FXS", + "4 Port FXO", + "4 Port FXS", + "72 Port FXS", + "8 Port FXS", +] + +SignalChoice = Literal[ + "DID", + "GroundStart", + "LoopStart", +] + + +DidType = Literal[ + "Delay dial", + "Immediate", + "Wink start", +] + + +class VoiceCardBasic(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + shutdown: Union[Variable, Default[bool], Global[bool]] = Field() + signal_choice: Union[Variable, Global[SignalChoice], Default[Literal["loopStart"]]] = Field( + validation_alias="signalChoice", serialization_alias="signalChoice" + ) + description: Optional[Union[Variable, Default[None], Global[str]]] = Field(default=None) + did_type: Optional[Union[Variable, Default[None], Global[DidType]]] = Field( + default=None, validation_alias="didType", serialization_alias="didType" + ) + + +class VoiceCardStationID(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + station_name: Union[Variable, Global[str]] = Field( + validation_alias="stationName", serialization_alias="stationName" + ) + station_number: Union[Variable, Global[str]] = Field( + validation_alias="stationNumber", serialization_alias="stationNumber" + ) + + +CompandType = Literal[ + "A-law", + "U-law", +] + + +Impedance = Literal[ + "600c", + "600r", + "900c", + "900r", + "complex1", + "complex2", + "complex3", + "complex4", + "complex5", + "complex6", +] + + +CallProgressTone = Literal[ + "Argentina", + "Australia", + "Austria", + "Belgium", + "Brazil", + "Canada", + "Chile", + "China", + "Columbia", + "Custom 1", + "Custom 2", + "Cyprus", + "Czech Republic", + "Denmark", + "Egypt", + "Finland", + "France", + "Germany", + "Ghana", + "Greece", + "Hong Kong", + "Hungary", + "Iceland", + "India", + "Indonesia", + "Ireland", + "Israel", + "Italy", + "Japan", + "Jordan", + "Kenya", + "Korea Republic", + "Kuwait", + "Lebanon", + "Luxembourg", + "Malaysia", + "Malta", + "Mexico", + "Nepal", + "Netherlands", + "New Zealand", + "Nigeria", + "Norway", + "Oman", + "Pakistan", + "Panama", + "Peru", + "Philippines", + "Poland", + "Portugal", + "Russian Federation", + "Saudi Arabia", + "Singapore", + "Slovakia", + "Slovenia", + "South Africa", + "Spain", + "Sweden", + "Switzerland", + "Taiwan", + "Thailand", + "Turkey", + "United Arab Emirates", + "United Kingdom", + "United States", + "Venezuela", + "Zimbabwe", +] + + +class LineParams(BaseModel): + model_config = ConfigDict(populate_by_name=True) + attenuation: Union[Variable, Default[int], Global[int]] = Field() + call_progress_tone: Union[Variable, Global[CallProgressTone]] = Field( + validation_alias="callProgressTone", serialization_alias="callProgressTone" + ) + compand_type: Union[Variable, Global[CompandType], Default[Literal["U-law"]]] = Field( + validation_alias="compandType", serialization_alias="compandType" + ) + echo_canceller: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias="echoCanceller", serialization_alias="echoCanceller" + ) + gain: Union[Variable, Default[int], Global[int]] = Field() + impedance: Union[Global[Impedance], Default[Literal["600r"]], Variable] = Field() + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + voice_activity_detection: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias="voiceActivityDetection", serialization_alias="voiceActivityDetection" + ) + + +SupervisoryDisconnect = Literal[ + "Anytone", + "Dualtone", + "signal", +] + + +SupervisoryDisconnectDualtone = Literal[ + "Mid call", + "Pre Connect", +] + +DialType = Literal[ + "dtmf", + "mf", + "pulse", +] + + +DetectionDelayBatteryReversal = Literal[ + "Answer", + "Both", + "Detection Delay", +] + + +class TuningParams(BaseModel): + model_config = ConfigDict(populate_by_name=True) + detection_delay: Union[Variable, Global[int]] = Field( + validation_alias="detectionDelay", serialization_alias="detectionDelay" + ) + detection_delay_battery_reversal: Union[Variable, Global[DetectionDelayBatteryReversal]] = Field( + validation_alias="detectionDelayBatteryReversal", serialization_alias="detectionDelayBatteryReversal" + ) + dial_delay: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="dialDelay", serialization_alias="dialDelay" + ) + dial_type: Union[Global[DialType], Variable, Default[Literal["dtmf"]]] = Field( + validation_alias="dialType", serialization_alias="dialType" + ) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + supervisory_disconnect: Union[Global[SupervisoryDisconnect], Default[Literal["signal"]], Variable] = Field( + validation_alias="supervisoryDisconnect", serialization_alias="supervisoryDisconnect" + ) + timing_guard_out: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="timingGuardOut", serialization_alias="timingGuardOut" + ) + timing_hookflash_out: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="timingHookflashOut", serialization_alias="timingHookflashOut" + ) + timing_sup_disconnect: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="timingSupDisconnect", serialization_alias="timingSupDisconnect" + ) + supervisory_disconnect_dualtone: Optional[Union[Variable, Global[SupervisoryDisconnectDualtone]]] = Field( + default=None, + validation_alias="supervisoryDisconnectDualtone", + serialization_alias="supervisoryDisconnectDualtone", + ) + + +LoopLength = Literal[ + "Long", + "Short", +] + + +DcOffSet = Literal[ + "10-volts", + "20-volts", + "24-volts", + "30-volts", + "35-volts", +] + + +class TuningParamsFxs(BaseModel): + model_config = ConfigDict(populate_by_name=True) + loop_length: Union[Variable, Global[LoopLength], Default[Literal["Short"]]] = Field( + validation_alias="loopLength", serialization_alias="loopLength" + ) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + pulse_digit_detection: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias="pulseDigitDetection", serialization_alias="pulseDigitDetection" + ) + ren: Union[Variable, Global[int]] = Field() + ring: Union[Variable, Default[int], Global[int]] = Field() + timing_hookflash_in: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="timingHookflashIn", serialization_alias="timingHookflashIn" + ) + dc_off_set: Optional[Union[Variable, Global[DcOffSet]]] = Field( + default=None, validation_alias="dcOffSet", serialization_alias="dcOffSet" + ) + timing_hookflash_out_sup: Optional[Union[Variable, Default[int], Global[int]]] = Field( + default=None, validation_alias="timingHookflashOutSup", serialization_alias="timingHookflashOutSup" + ) + + +CallerMode = Literal[ + "BT", + "DTMF", + "FSK", +] + +DtmfModeSelectionEnd = Literal[ + "#", + "*", + "A", + "B", + "C", + "D", +] + +AlertOptions = Literal[ + "Line-Reversal", + "Pre-ring", + "Ring 1", + "Ring 2", + "Ring 3", + "Ring 4", +] + + +class CallerId(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + alert_options: Optional[Union[Global[AlertOptions], Variable, Default[None]]] = Field( + default=None, validation_alias="alertOptions", serialization_alias="alertOptions" + ) + caller_dsp_pre_allocate: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="callerDspPreAllocate", serialization_alias="callerDspPreAllocate" + ) + caller_id_block: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="callerIdBlock", serialization_alias="callerIdBlock" + ) + caller_id_format: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="callerIdFormat", serialization_alias="callerIdFormat" + ) + caller_mode: Optional[Union[Variable, Global[CallerMode], Default[None]]] = Field( + default=None, validation_alias="callerMode", serialization_alias="callerMode" + ) + dtmf_codes: Optional[Union[Variable, Global[str]]] = Field( + default=None, validation_alias="dtmfCodes", serialization_alias="dtmfCodes" + ) + dtmf_mode_selection_end: Optional[Union[Variable, Global[DtmfModeSelectionEnd], Default[None]]] = Field( + default=None, validation_alias="dtmfModeSelectionEnd", serialization_alias="dtmfModeSelectionEnd" + ) + dtmf_mode_selection_start: Optional[Union[Variable, Global[DtmfModeSelectionEnd], Default[None]]] = Field( + default=None, validation_alias="dtmfModeSelectionStart", serialization_alias="dtmfModeSelectionStart" + ) + enable: Optional[Union[Variable, Default[bool], Global[bool]]] = Field(default=None) + + +class DidTimer(BaseModel): + model_config = ConfigDict(populate_by_name=True) + answer_winkwidth: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="answerWinkwidth", serialization_alias="answerWinkwidth" + ) + clear_wait: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="clearWait", serialization_alias="clearWait" + ) + dial_pulse_min_delay: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="dialPulseMinDelay", serialization_alias="dialPulseMinDelay" + ) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + wait_before_wink: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="waitBeforeWink", serialization_alias="waitBeforeWink" + ) + wink_duration: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="winkDuration", serialization_alias="winkDuration" + ) + + +class ConnectionPlar(BaseModel): + model_config = ConfigDict(populate_by_name=True) + connection_plar: Union[Variable, Global[str]] = Field( + validation_alias="connectionPlar", serialization_alias="connectionPlar" + ) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + connection_plar_opx: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="connectionPlarOpx", serialization_alias="connectionPlarOpx" + ) + + +TranslationRuleDirection = Literal[ + "incoming", + "outgoing", +] + + +class Association(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + supervisory_disconnect: Optional[RefIdItem] = Field( + default=None, validation_alias="supervisoryDisconnect", serialization_alias="supervisoryDisconnect" + ) + translation_profile: Optional[RefIdItem] = Field( + default=None, validation_alias="translationProfile", serialization_alias="translationProfile" + ) + translation_rule_direction: Optional[Union[Variable, Global[TranslationRuleDirection], Default[None]]] = Field( + default=None, validation_alias="translationRuleDirection", serialization_alias="translationRuleDirection" + ) + trunk_group: Optional[RefIdItem] = Field( + default=None, validation_alias="trunkGroup", serialization_alias="trunkGroup" + ) + trunk_group_priority: Optional[Union[Variable, Global[int], Default[None]]] = Field( + default=None, validation_alias="trunkGroupPriority", serialization_alias="trunkGroupPriority" + ) + + +class AnalogInterfaceParcel(_ParcelBase): + type_: Literal["analog-interface"] = Field(default="analog-interface", exclude=True) + model_config = ConfigDict(populate_by_name=True) + enable: Union[Variable, Default[bool], Global[bool]] = Field(validation_alias=AliasPath("data", "enable")) + slot_id: Union[Variable, Global[SlotId]] = Field(validation_alias=AliasPath("data", "slotId")) + association: Optional[List[Association]] = Field( + default=None, + validation_alias=AliasPath("data", "association"), + description="Association", + ) + caller_id: Optional[List[CallerId]] = Field(default=None, validation_alias=AliasPath("data", "callerId")) + connection_plar: Optional[List[ConnectionPlar]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "connectionPlar"), + description="Connection plar", + ) + did_timer: Optional[List[DidTimer]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "didTimer"), + description="DID timer", + ) + line_params: Optional[List[LineParams]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "lineParams"), + description="Configure of voice card station Id", + ) + module_type: Optional[Global[ModuleType]] = Field( + default_factory=list, validation_alias=AliasPath("data", "moduleType") + ) + tuning_params: Optional[List[TuningParams]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "tuningParams"), + description="Configure of voice card station Id", + ) + tuning_params_fxs: Optional[List[TuningParamsFxs]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "tuningParamsFxs"), + description="Configure of voice card station Id", + ) + voice_card_basic: Optional[List[VoiceCardBasic]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "voiceCardBasic"), + description="Configure of voice card", + ) + voice_card_station_i_d: Optional[List[VoiceCardStationID]] = Field( + default_factory=list, + validation_alias=AliasPath("data", "voiceCardStationID"), + description="Configure of voice card station Id", + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/call_routing.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/call_routing.py new file mode 100644 index 00000000..a483ab5f --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/call_routing.py @@ -0,0 +1,179 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase +from catalystwan.models.configuration.feature_profile.common import RefIdItem + +VoiceType = Literal[ + "pots", + "sip", +] + +TranslationRuleDirection = Literal[ + "incoming", + "outgoing", +] + +FwdDigitChoice = Literal[ + "all", + "none", + "some", +] + +TransportChoice = Literal[ + "tcp", + "udp", +] + + +class Voice(BaseModel): + model_config = ConfigDict(populate_by_name=True) + dialpeertag: Union[Variable, Global[int]] = Field() + call_type: Optional[Global[TranslationRuleDirection]] = Field( + default=None, validation_alias="callType", serialization_alias="callType" + ) + description: Optional[Union[Variable, Default[None], Global[str]]] = Field( + default=None, + validation_alias=AliasPath("data", "description"), + ) + destination_address: Optional[Union[Variable, Global[str]]] = Field( + default=None, validation_alias="destinationAddress", serialization_alias="destinationAddress" + ) + fwd_digit_choice: Optional[Union[Variable, Global[FwdDigitChoice]]] = Field( + default=None, validation_alias="fwdDigitChoice", serialization_alias="fwdDigitChoice" + ) + num_digits: Optional[Union[Variable, Global[int]]] = Field( + default=None, validation_alias="numDigits", serialization_alias="numDigits" + ) + number_pattern: Optional[Union[Variable, Global[str]]] = Field( + default=None, validation_alias="numberPattern", serialization_alias="numberPattern" + ) + port: Optional[Union[Variable, Global[str]]] = Field(default=None) + preference: Optional[Union[Variable, Global[int], Default[None]]] = Field(default=None) + prefix: Optional[Union[Variable, Global[int], Default[None]]] = Field(default=None) + transport_choice: Optional[Union[Variable, Global[TransportChoice], Default[None]]] = Field( + default=None, validation_alias="transportChoice", serialization_alias="transportChoice" + ) + type: Optional[Global[VoiceType]] = Field(default=None) + + +Fallback = Literal[ + "G.711alaw", + "G.711ulaw", + "None", +] + + +class ModemPassThrough(BaseModel): + model_config = ConfigDict(populate_by_name=True) + dial_peer_range: Union[Variable, Global[str]] = Field( + validation_alias="dialPeerRange", serialization_alias="dialPeerRange" + ) + protocol: Union[Variable, Global[Fallback]] = Field() + + +Primary = Literal[ + "Fax Pass-through G711alaw", + "Fax Pass-through G711alaw No ECM", + "Fax Pass-through G711ulaw", + "Fax Pass-through G711ulaw No ECM", + "None", + "T.38 Fax Relay Version 0", + "T.38 Fax Relay Version 0 NSE", + "T.38 Fax Relay Version 0 NSE No ECM", + "T.38 Fax Relay Version 0 NSE Rate 14.4", + "T.38 Fax Relay Version 0 NSE Rate 14.4 No ECM", + "T.38 Fax Relay Version 0 NSE Rate 9.6", + "T.38 Fax Relay Version 0 NSE Rate 9.6 No ECM", + "T.38 Fax Relay Version 0 NSE force", + "T.38 Fax Relay Version 0 NSE force No ECM", + "T.38 Fax Relay Version 0 NSE force Rate 14.4", + "T.38 Fax Relay Version 0 NSE force Rate 14.4 No ECM", + "T.38 Fax Relay Version 0 NSE force Rate 9.6", + "T.38 Fax Relay Version 0 NSE force Rate 9.6 No ECM", + "T.38 Fax Relay Version 0 No ECM", + "T.38 Fax Relay Version 0 Rate 14.4", + "T.38 Fax Relay Version 0 Rate 14.4 No ECM", + "T.38 Fax Relay Version 0 Rate 9.6 No ECM", + "T.38 Fax Relay Version 3", + "T.38 Fax Relay Version 3 NSE", + "T.38 Fax Relay Version 3 NSE force", +] + + +class FaxProtocol(BaseModel): + model_config = ConfigDict(populate_by_name=True) + dial_peer_range: Union[Variable, Global[str]] = Field( + validation_alias="dialPeerRange", serialization_alias="dialPeerRange" + ) + primary: Union[Variable, Global[Primary]] = Field() + fallback: Optional[Union[Variable, Global[Fallback]]] = Field(default=None) + high_speed: Optional[Union[Variable, Global[int]]] = Field( + default=None, validation_alias="highSpeed", serialization_alias="highSpeed" + ) + low_speed: Optional[Union[Variable, Global[int]]] = Field( + default=None, validation_alias="lowSpeed", serialization_alias="lowSpeed" + ) + + +class Association(BaseModel): + model_config = ConfigDict(populate_by_name=True) + dial_peer_range: Union[Variable, Global[str]] = Field( + validation_alias="dialPeerRange", serialization_alias="dialPeerRange" + ) + media_profile: Optional[RefIdItem] = Field( + default=None, validation_alias="mediaProfile", serialization_alias="mediaProfile" + ) + server_group: Optional[RefIdItem] = Field( + default=None, validation_alias="serverGroup", serialization_alias="serverGroup" + ) + translation_profile: Optional[RefIdItem] = Field( + default=None, validation_alias="translationProfile", serialization_alias="translationProfile" + ) + translation_rule_direction: Optional[Union[Variable, Global[TranslationRuleDirection], Default[None]]] = Field( + default=None, validation_alias="translationRuleDirection", serialization_alias="translationRuleDirection" + ) + trunk_group: Optional[RefIdItem] = Field( + default=None, validation_alias="trunkGroup", serialization_alias="trunkGroup" + ) + trunk_group_priority: Optional[Union[Variable, Global[int], Default[None]]] = Field( + default=None, validation_alias="trunkGroupPriority", serialization_alias="trunkGroupPriority" + ) + voice_tenant: Optional[RefIdItem] = Field( + default=None, validation_alias="voiceTenant", serialization_alias="voiceTenant" + ) + + +class CallRoutingParcel(_ParcelBase): + type_: Literal["call-routing"] = Field(default="call-routing", exclude=True) + model_config = ConfigDict(populate_by_name=True) + association: Optional[List[Association]] = Field( + default=None, + validation_alias=AliasPath("data", "association"), + description="Association", + ) + dial_peer_tag_prefix: Optional[Union[Variable, Global[int], Default[None]]] = Field( + default=None, + validation_alias=AliasPath("data", "dialPeerTagPrefix"), + ) + fax_protocol: Optional[List[FaxProtocol]] = Field( + default=None, + validation_alias=AliasPath("data", "faxProtocol"), + description="Configure fax protocol", + ) + modem_pass_through: Optional[List[ModemPassThrough]] = Field( + default=None, + validation_alias=AliasPath("data", "modemPassThrough"), + description="Configure of Modem Pass-Through", + ) + port_module_location: Optional[RefIdItem] = Field( + default=None, + validation_alias=AliasPath("data", "portModuleLocation"), + ) + voice: Optional[List[Voice]] = Field( + default=None, + validation_alias=AliasPath("data", "voice"), + description="POTS/Voip voice type", + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/digital_interface.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/digital_interface.py new file mode 100644 index 00000000..6d4817e3 --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/digital_interface.py @@ -0,0 +1,324 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase +from catalystwan.models.configuration.feature_profile.common import RefIdItem + +VoiceInterfaceTemplates = Literal[ + "1 Port E1", + "1 Port T1", + "2 Port E1", + "2 Port T1", + "4 Port E1", + "4 Port T1", + "8 Port E1", + "8 Port T1", +] + +ModuleLocation = Literal[ + "0/1", + "0/2", + "0/3", + "1/0", + "2/0", +] + +ClockType = Literal[ + "line", + "network", + "primary", + "secondary", +] + + +class Interface(BaseModel): + port_id: Union[Variable, Global[int], Default[None]] = Field( + validation_alias="portId", serialization_alias="portId" + ) + clock_type: Optional[Union[Variable, Default[None], Global[ClockType]]] = Field( + default=None, validation_alias="clockType", serialization_alias="clockType" + ) + + +Framing = Literal[ + "crc4", + "no-crc4", + "crc4", + "esf", + "sf", +] + + +LineCode = Literal[ + "ami", + "b8zs", + "hdb3", +] + + +CableLengthType = Literal[ + "long", + "short", +] + + +Value = Literal[ + "-15", + "-22.5", + "-7.5", + "0", +] + +CableLengthValue = Literal[ + "110", + "220", + "330", + "440", + "550", + "660", +] + + +LineTermination = Literal[ + "120-ohm", + "75-ohm", +] + + +SwitchType = Literal[ + "primary-4ess", + "primary-5ess", + "primary-dms100", + "primary-net5", + "primary-ni", + "primary-ntt", + "primary-qsig", +] + + +class BasicSettings(BaseModel): + model_config = ConfigDict(populate_by_name=True) + delay_connect_timer: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="delayConnectTimer", serialization_alias="delayConnectTimer" + ) + framing: Union[Variable, Global[Framing], Default[Framing]] = Field() + line_code: Union[Variable, Default[LineCode], Global[LineCode]] = Field( + validation_alias="lineCode", serialization_alias="lineCode" + ) + network_side: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias="networkSide", serialization_alias="networkSide" + ) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + switch_type: Union[Variable, Global[SwitchType], Default[SwitchType]] = Field( + validation_alias="switchType", serialization_alias="switchType" + ) + cable_length: Optional[Union[Variable, Default[Literal["0"]], Global[CableLengthValue]]] = Field( + default=None, validation_alias="cableLength", serialization_alias="cableLength" + ) + cable_length_type: Optional[Union[Variable, Default[Literal["long"]], Global[CableLengthType]]] = Field( + default=None, validation_alias="cableLengthType", serialization_alias="cableLengthType" + ) + framing_australia: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="framingAustralia", serialization_alias="framingAustralia" + ) + line_termination: Optional[Union[Variable, Default[Literal["120-ohm"]], Global[LineTermination]]] = Field( + default=None, validation_alias="lineTermination", serialization_alias="lineTermination" + ) + timeslots: Optional[Global[str]] = Field(default=None) + + +TypeAndTimerType = Literal[ + "T200", + "T203", + "T301", + "T303", + "T306", + "T309", + "T310", + "T321", +] + + +class TypeAndTimer(BaseModel): + model_config = ConfigDict(populate_by_name=True) + timer: Union[Variable, Global[int]] = Field() + type: Union[Variable, Global[TypeAndTimerType]] = Field() + + +class IsdnTimer(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Optional[Union[Variable, Global[str]]] = Field( + default=None, validation_alias="portRange", serialization_alias="portRange" + ) + type_and_timer: Optional[List[TypeAndTimer]] = Field( + default=None, + validation_alias="typeAndTimer", + serialization_alias="typeAndTimer", + description="list of ISDN Type and Timers", + ) + + +Plan = Literal[ + "data", + "isdn", + "national", + "privacy", + "reserved/10", + "reserved/11", + "reserved/12", + "reserved/13", + "reserved/14", + "reserved/2", + "reserved/5", + "reserved/6", + "reserved/7", + "telex", + "unknown", +] + +IsdnMapType = Literal[ + "abbreviated", + "international", + "national", + "reserved/5", + "subscriber", + "unknown", +] + + +class IsdnMap(BaseModel): + model_config = ConfigDict(populate_by_name=True) + digit_range: Union[Variable, Global[str]] = Field(validation_alias="digitRange", serialization_alias="digitRange") + plan: Union[Variable, Default[None], Global[Plan]] = Field() + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + type: Union[Variable, Default[None], Global[IsdnMapType]] = Field() + + +class Shutdown(BaseModel): + model_config = ConfigDict(populate_by_name=True) + controller: Union[Variable, Default[bool], Global[bool]] = Field() + port_id: Union[Variable, Global[int], Default[None]] = Field( + validation_alias="portId", serialization_alias="portId" + ) + serial: Union[Variable, Default[bool], Global[bool]] = Field() + voice_port: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias="voicePort", serialization_alias="voicePort" + ) + + +CompandType = Literal[ + "a-law", + "u-law", +] + + +class LineParams(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + attenuation: Optional[Union[Variable, Default[int], Global[int]]] = Field(default=None) + call_progress_tone: Optional[Union[Variable, Global[None], Default[None]]] = Field( + default=None, validation_alias="callProgressTone", serialization_alias="callProgressTone" + ) + compand_type: Optional[Union[Variable, Default[CompandType], Global[CompandType]]] = Field( + default=None, validation_alias="compandType", serialization_alias="compandType" + ) + echo_canceller: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="echoCanceller", serialization_alias="echoCanceller" + ) + gain: Optional[Union[Variable, Default[int], Global[int]]] = Field(default=None) + voice_activity_detection: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias="voiceActivityDetection", serialization_alias="voiceActivityDetection" + ) + + +OutgoingIeType = Literal[ + "called-number", + "called-subaddr", + "caller-number", + "caller-subaddr", + "connected-number", + "connected-subaddr", + "display", + "extended-facility", + "facility", + "high-layer-compat", + "low-layer-compat", + "network-facility", + "notify-indicator", + "progress-indicator", + "redirecting-number", + "user-user", +] + + +class OutgoingIe(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + type: Union[Variable, Global[List[OutgoingIeType]]] = Field() + + +TranslationProfileDirection = Literal[ + "incoming", + "outgoing", +] + + +class Association(BaseModel): + model_config = ConfigDict(populate_by_name=True) + port_range: Union[Variable, Global[str]] = Field(validation_alias="portRange", serialization_alias="portRange") + translation_profile: Optional[RefIdItem] = Field( + default=None, validation_alias="translationProfile", serialization_alias="translationProfile" + ) + translation_profile_direction: Optional[ + Union[Variable, Default[None], Global[TranslationProfileDirection]] + ] = Field( + default=None, validation_alias="translationProfileDirection", serialization_alias="translationProfileDirection" + ) + trunk_group: Optional[RefIdItem] = Field( + default=None, validation_alias="trunkGroup", serialization_alias="trunkGroup" + ) + trunk_group_priority: Optional[Union[Variable, Global[int], Default[None]]] = Field( + default=None, validation_alias="trunkGroupPriority", serialization_alias="trunkGroupPriority" + ) + + +class DigitalInterfaceParcel(_ParcelBase): + type_: Literal["digital-interface"] = Field(default="digital-interface", exclude=True) + model_config = ConfigDict(populate_by_name=True) + basic_settings: List[BasicSettings] = Field( + validation_alias=AliasPath("data", "basicSettings"), description="add basic setting" + ) + dsp_hairpin: Union[Variable, Default[bool], Global[bool]] = Field( + validation_alias=AliasPath("data", "dspHairpin"), + ) + interface: List[Interface] = Field( + validation_alias=AliasPath("data", "interface"), description="Configure Digital voice card interface" + ) + isdn_timer: List[IsdnTimer] = Field( + validation_alias=AliasPath("data", "isdnTimer"), description="list of ISDN Timers" + ) + module_location: Union[Variable, Global[ModuleLocation]] = Field( + validation_alias=AliasPath("data", "moduleLocation") + ) + shutdown: List[Shutdown] = Field( + validation_alias=AliasPath("data", "shutdown"), description="list of shutdown options" + ) + association: Optional[List[Association]] = Field( + default=None, + validation_alias=AliasPath("data", "associations"), + description="Select Trunk Group and Translation Profile associations", + ) + isdn_map: Optional[List[IsdnMap]] = Field( + default=None, validation_alias=AliasPath("data", "isdnMap"), description="list of ISDN map" + ) + line_params: Optional[List[LineParams]] = Field( + default=None, validation_alias=AliasPath("data", "lineParams"), description="list of line parameters" + ) + outgoing_ie: Optional[List[OutgoingIe]] = Field( + default=None, validation_alias=AliasPath("data", "outgoingIe"), description="list of outgoing IEs and messages" + ) + voice_interface_templates: Optional[Global[VoiceInterfaceTemplates]] = Field( + default=None, validation_alias="voiceInterfaceTemplates" + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/server_group.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/server_group.py new file mode 100644 index 00000000..b867aa8e --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/server_group.py @@ -0,0 +1,51 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase + + +class AddressList(BaseModel): + model_config = ConfigDict(populate_by_name=True) + address: Union[Variable, Global[str]] = Field() + port: Optional[Union[Variable, Global[int], Default[None]]] = Field(default=None) + preference: Optional[Union[Variable, Global[int], Default[None]]] = Field(default=None) + + +HuntScheme = Literal[ + "none", + "round-robin", +] + + +class HuntStopRule(BaseModel): + model_config = ConfigDict(populate_by_name=True) + response_code_end: Union[Variable, Global[int]] = Field( + validation_alias="responseCodeEnd", serialization_alias="responseCodeEnd" + ) + response_code_start: Union[Variable, Global[int]] = Field( + validation_alias="responseCodeStart", serialization_alias="responseCodeStart" + ) + rule_id: Union[Variable, Global[int]] = Field(validation_alias="ruleId", serialization_alias="ruleId") + + +class ServerGroupParcel(_ParcelBase): + type_: Literal["server-group"] = Field(default="server-group", exclude=True) + model_config = ConfigDict(populate_by_name=True) + server_group_id: Union[Variable, Global[int]] = Field(validation_alias=AliasPath("data", "serverGroupId")) + address_list: Optional[List[AddressList]] = Field( + default=None, + validation_alias=AliasPath("data", "addressList"), + ) + hunt_scheme: Optional[Union[Variable, Global[HuntScheme], Default[Literal["none"]]]] = Field( + default=None, + validation_alias=AliasPath("data", "huntScheme"), + ) + hunt_stop_rule: Optional[List[HuntStopRule]] = Field( + default=None, + validation_alias=AliasPath("data", "huntStopRule"), + ) + shutdown: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias=AliasPath("data", "shutdown") + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/srst.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/srst.py new file mode 100644 index 00000000..38205bb0 --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/srst.py @@ -0,0 +1,90 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase +from catalystwan.models.configuration.feature_profile.common import RefIdItem + + +class Pool(BaseModel): + model_config = ConfigDict(populate_by_name=True) + ipv4_oripv6prefix: Union[Variable, Global[str]] = Field( + validation_alias="ipv4Oripv6prefix", serialization_alias="ipv4Oripv6prefix" + ) + pool_tag: Union[Variable, Global[int]] = Field(validation_alias="poolTag", serialization_alias="poolTag") + + +CallFowardAction = Literal[ + "all", + "busy", + "noan", +] + + +class CallForward(BaseModel): + action: Union[Variable, Global[CallFowardAction]] = Field() + digit_string: Union[Variable, Global[str]] = Field( + validation_alias="digitString", serialization_alias="digitString" + ) + phone_profile: Union[Variable, Global[int]] = Field( + validation_alias="phoneProfile", serialization_alias="phoneProfile" + ) + timeout: Optional[Union[Variable, Default[int], Global[int]]] = Field(default=None) + + +class TranslationProfile(BaseModel): + ref_id: Union[Default[None], Global[str]] = Field(validation_alias="refId", serialization_alias="refId") + + +TranslationProfileDirection = Literal[ + "incoming", + "outgoing", +] + + +class TranslationAndMediaProfile(BaseModel): + model_config = ConfigDict(populate_by_name=True) + media_profile: Optional[RefIdItem] = Field( + default=None, validation_alias="mediaProfile", serialization_alias="mediaProfile" + ) + phone_profile: Optional[Union[Variable, Global[int]]] = Field( + default=None, validation_alias="phoneProfile", serialization_alias="phoneProfile" + ) + translation_profile: Optional[TranslationProfile] = Field( + default=None, validation_alias="translationProfile", serialization_alias="translationProfile" + ) + translation_profile_direction: Optional[ + Union[Variable, Default[None], Global[TranslationProfileDirection]] + ] = Field( + default=None, validation_alias="translationProfileDirection", serialization_alias="translationProfileDirection" + ) + + +class SrstParcel(_ParcelBase): + type_: Literal["srst"] = Field(default="srst", exclude=True) + model_config = ConfigDict(populate_by_name=True) + max_dn: Union[Variable, Global[int]] = Field(validation_alias=AliasPath("data", "maxDn")) + max_phones: Union[Variable, Global[int]] = Field(validation_alias=AliasPath("data", "maxPhones")) + pool: List[Pool] = Field( + validation_alias=AliasPath("data", "pool"), + description="Voice register pool", + ) + call_forward: Optional[List[CallForward]] = Field( + default=None, + validation_alias=AliasPath("data", "callForward"), + description="Call forward option", + ) + filename: Optional[Union[Default[None], Global[str]]] = Field( + default=None, validation_alias=AliasPath("data", "filename") + ) + moh: Optional[Union[Default[bool], Global[bool]]] = Field(default=None, validation_alias=AliasPath("data", "moh")) + system_message: Optional[Union[Default[None], Global[str]]] = Field( + default=None, + validation_alias=AliasPath("data", "systemMessage"), + ) + translation_and_media_profile: Optional[List[TranslationAndMediaProfile]] = Field( + default=None, + validation_alias=AliasPath("data", "translationAndMediaProfile"), + description="translationProfile ID Refs and mediaProfile ID Refs", + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/supervisory_disconnect.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/supervisory_disconnect.py new file mode 100644 index 00000000..2e0f3391 --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/supervisory_disconnect.py @@ -0,0 +1,68 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase + +DualTone = Literal[ + "Busy", + "Disconnect", + "Number Unobtainable", + "Out of Service", + "Reorder", + "Ringback", +] + + +class SupervisoryCustomCPTone(BaseModel): + model_config = ConfigDict(populate_by_name=True) + cadence: Union[Variable, Global[str]] = Field() + dual_tone: Union[Variable, Global[DualTone]] = Field(validation_alias="dualTone", serialization_alias="dualTone") + dualtone_frequency_in: Union[Variable, Global[int]] = Field( + validation_alias="dualtoneFrequencyIn", serialization_alias="dualtoneFrequencyIn" + ) + dualtone_frequency_out: Union[Variable, Global[int]] = Field( + validation_alias="dualtoneFrequencyOut", serialization_alias="dualtoneFrequencyOut" + ) + supervisory_name: Union[Variable, Global[str]] = Field( + validation_alias="supervisoryName", serialization_alias="supervisoryName" + ) + + +class SupervisoryCustomDetectionParams(BaseModel): + model_config = ConfigDict(populate_by_name=True) + cadence_variation: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="cadenceVariation", serialization_alias="cadenceVariation" + ) + max_delay: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="maxDelay", serialization_alias="maxDelay" + ) + max_deviation: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="maxDeviation", serialization_alias="maxDeviation" + ) + max_power: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="maxPower", serialization_alias="maxPower" + ) + min_power: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="minPower", serialization_alias="minPower" + ) + min_power_twist: Union[Variable, Default[int], Global[int]] = Field( + validation_alias="minPowerTwist", serialization_alias="minPowerTwist" + ) + supervisory_number: Optional[Union[Variable, Global[int]]] = Field( + default=None, validation_alias="supervisoryNumber", serialization_alias="supervisoryNumber" + ) + + +class SupervisoryDiconnectParcel(_ParcelBase): + type_: Literal["supervisory-disconnect"] = Field(default="supervisory-disconnect", exclude=True) + model_config = ConfigDict(populate_by_name=True) + supervisory_custom_c_p_tone: Optional[List[SupervisoryCustomCPTone]] = Field( + default=None, + validation_alias=AliasPath("data", "supervisoryCustomCPTone"), + ) + supervisory_custom_detection_params: Optional[List[SupervisoryCustomDetectionParams]] = Field( + default=None, + validation_alias=AliasPath("data", "supervisoryCustomDetectionParams"), + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_profile.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_profile.py index b0b52b1c..9b15051b 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_profile.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_profile.py @@ -1,5 +1,6 @@ # Copyright 2024 Cisco Systems, Inc. and its affiliates from typing import List, Literal, Optional, Union +from uuid import UUID from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -27,3 +28,15 @@ class TranslationProfileParcel(_ParcelBase): validation_alias=AliasPath("data", "translationProfileSettings"), description="Translation Profile configuration", ) + + def set_ref_by_call_type(self, ref: UUID, ct: CallType) -> TranslationProfileSettings: + """Set reference UUID to a calling or called rule item or create one and then set the UUID""" + tps = None + for tps_ in self.translation_profile_settings: + if isinstance(tps_.call_type, Global) and tps_.call_type.value == ct: + tps = tps_ + if tps is None: + tps = TranslationProfileSettings(call_type=Global[CallType](value=ct)) + self.translation_profile_settings.append(tps) + tps.translation_rule = RefIdItem.from_uuid(ref) + return tps diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_rule.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_rule.py index 24faebe0..cc5f965b 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_rule.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/translation_rule.py @@ -28,4 +28,4 @@ class TranslationRuleParcel(_ParcelBase): validation_alias=AliasPath("data", "ruleSettings"), description="Translation Rule configuration", ) - rule_name: Optional[Global[int]] = Field(default=None, validation_alias="ruleName", serialization_alias="ruleName") + rule_name: Optional[Global[int]] = Field(default=None, validation_alias=AliasPath("data", "ruleName")) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_global.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_global.py new file mode 100644 index 00000000..d3120ede --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_global.py @@ -0,0 +1,55 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import List, Literal, Optional, Union + +from pydantic import AliasPath, BaseModel, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase + +IpFormat = Literal[ + "ipv4", + "ipv6", +] + + +class TrustedPrefixList(BaseModel): + model_config = ConfigDict(populate_by_name=True) + ip: Optional[Union[Variable, Default[None], Global[str]]] = Field(default=None) + ip_format: Optional[Union[Variable, Default[None], Global[IpFormat]]] = Field( + default=None, validation_alias="ipFormat", serialization_alias="ipFormat" + ) + + +class ClockPrioritySorting(BaseModel): + model_config = ConfigDict(populate_by_name=True) + clock_priority: Optional[Global[int]] = Field( + default=None, validation_alias="clockPriority", serialization_alias="clockPriority" + ) + clock_priority_sorting_port: Optional[Union[Variable, Default[None], Global[str]]] = Field( + default=None, validation_alias="clockPrioritySortingPort", serialization_alias="clockPrioritySortingPort" + ) + + +class VoiceGlobalParcel(_ParcelBase): + type_: Literal["voice-global"] = Field(default="voice-global", exclude=True) + model_config = ConfigDict(populate_by_name=True) + clock_priority_sorting: Optional[List[ClockPrioritySorting]] = Field( + default=None, + validation_alias=AliasPath("data", "clockPrioritySorting"), + description="Clock Priority Sorting", + ) + source_interface: Optional[Union[Variable, Default[None], Global[str]]] = Field( + default=None, + validation_alias=AliasPath("data", "sourceInterface"), + ) + sync: Optional[Union[Variable, Default[bool], Global[bool]]] = Field( + default=None, validation_alias=AliasPath("data", "sync") + ) + trusted_prefix_list: Optional[List[TrustedPrefixList]] = Field( + default=None, + validation_alias=AliasPath("data", "trustedPrefixList"), + description="Prefix List", + ) + wait_to_restore: Optional[Union[Variable, Default[int], Global[int]]] = Field( + default=None, + validation_alias=AliasPath("data", "waitToRestore"), + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_tenant.py b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_tenant.py new file mode 100644 index 00000000..d8491858 --- /dev/null +++ b/catalystwan/models/configuration/feature_profile/sdwan/uc_voice/voice_tenant.py @@ -0,0 +1,40 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +from typing import Literal, Optional, Union + +from pydantic import AliasPath, ConfigDict, Field + +from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase + +BindInterface = Literal[ + "Both", + "Control", + "Disabled", + "Media", +] + + +TransportType = Literal[ + "TCP", + "TCP TLS", + "UDP", +] + + +class VoiceTenantParcel(_ParcelBase): + type_: Literal["voice-tenant"] = Field(default="voice-tenant", exclude=True) + model_config = ConfigDict(populate_by_name=True) + bind_interface: Union[Variable, Global[BindInterface], Default[Literal["Disabled"]]] = Field( + validation_alias=AliasPath("data", "bindInterface") + ) + transport_type: Union[Variable, Global[TransportType], Default[Literal["UDP"]]] = Field( + validation_alias=AliasPath("data", "transportType") + ) + voice_tenant_tag: Union[Variable, Global[int]] = Field(validation_alias=AliasPath("data", "voiceTenantTag")) + bind_control_interface_name: Optional[Union[Variable, Global[str]]] = Field( + default=None, + validation_alias=AliasPath("data", "bindControlInterfaceName"), + ) + bind_media_interface_name: Optional[Union[Variable, Global[str]]] = Field( + default=None, + validation_alias=AliasPath("data", "bindMediaInterfaceName"), + ) diff --git a/catalystwan/tests/builders/uc_voice.py b/catalystwan/tests/builders/uc_voice.py new file mode 100644 index 00000000..239ba9fe --- /dev/null +++ b/catalystwan/tests/builders/uc_voice.py @@ -0,0 +1,85 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +import unittest +from typing import List +from unittest.mock import MagicMock +from uuid import uuid4 + +from catalystwan.api.builders.feature_profiles.uc_voice import UcVoiceFeatureProfileBuilder +from catalystwan.api.configuration_groups.parcel import as_default, as_global +from catalystwan.models.configuration.feature_profile.common import RefIdItem + + +class BaseModel: + def __init__(self, **kwargs): + self.model_fields_set = set(kwargs.keys()) + for key, value in kwargs.items(): + setattr(self, key, value) + + +class TestUcVoiceFeatureProfileBuilder(unittest.TestCase): + def setUp(self): + self.builder = UcVoiceFeatureProfileBuilder(session=MagicMock()) + self.builder._pushed_associable_parcels = { + "p1_name": uuid4(), + "p2_name": uuid4(), + } + + def test_populate_association_with_matching_fields(self): + association = [ + BaseModel( + media_profile=RefIdItem(ref_id=as_global("p2_name")), + server_group=RefIdItem(ref_id=as_global("p1_name")), + ), + ] + + self.builder._populate_association(association) + + # Assert that matching fields are updated + self.assertEqual( + association[0].media_profile.ref_id.value, str(self.builder._pushed_associable_parcels["p2_name"]) + ) + self.assertEqual( + association[0].server_group.ref_id.value, str(self.builder._pushed_associable_parcels["p1_name"]) + ) + + def test_populate_association_with_no_matching_fields(self): + association = [BaseModel(translation_profile=RefIdItem(ref_id=as_global("non_matching_field")))] + + self.builder._populate_association(association) + + # Assert that no changes are made for non-matching fields + self.assertEqual(association[0].translation_profile.ref_id, as_default(None)) + + def test_populate_association_partial_matching_fields(self): + association = [ + BaseModel( + media_profile=RefIdItem(ref_id=as_global("p2_name")), + supervisory_disconnect=RefIdItem(ref_id=as_global("non_existent_field")), + ) + ] + + self.builder._populate_association(association) + + # Assert that only matching fields are updated + self.assertEqual( + association[0].media_profile.ref_id.value, str(self.builder._pushed_associable_parcels["p2_name"]) + ) + self.assertEqual(association[0].supervisory_disconnect.ref_id, as_default(None)) + + def test_populate_association_with_empty_association(self): + association: List[BaseModel] = [] + + self.builder._populate_association(association) + + # Assert no errors occur and nothing is changed + self.assertEqual(len(association), 0) + + def test_populate_association_with_no_pushed_parcels(self): + self.builder._pushed_associable_parcels = {} + + association = [BaseModel(media_profile=RefIdItem(ref_id=as_global("p3_name")))] + + self.builder._populate_association(association) + + # Assert that fields are changed to default none when the name is missing + self.assertEqual(association[0].media_profile.ref_id, as_default(None)) diff --git a/pyproject.toml b/pyproject.toml index 58a58ffe..b10edae7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "catalystwan" -version = "0.40.0dev2" +version = "0.40.0dev3" description = "Cisco Catalyst WAN SDK for Python" authors = ["kagorski "] readme = "README.md"