Add builder

catalystwan/api/builders/feature_profiles/uc_voice.py

@@ -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__)
@@ -22,18 +41,52 @@
     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):
+        return True
+    try:
+        UUID(uuid)
+        return True
+    except ValueError:
+        return False
+
+
 class UcVoiceFeatureProfileBuilder:
     """
     A class for building UcVoice 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.
@@ -45,8 +98,10 @@ def __init__(self, session: ManagerSession) -> None:
         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] = {}  # parcel name: created object uuid
+        self._parcels_with_associations: List[ParcelWithAssociations] = []
 
     def add_profile_name_and_description(self, feature_profile: FeatureProfileCreationPayload) -> None:
         """
@@ -71,7 +126,7 @@ def add_parcel(self, parcel: AnyUcVoiceParcel) -> None:
         Returns:
             None
         """
-        self._independent_items.append(parcel)
+        self._independent_parcels.append(parcel)
 
     def add_translation_profile(
         self,
@@ -83,6 +138,14 @@ def add_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,
+    ):
+        """Items in assotiation list MUST have a ref_id field with
+        a value that matches the parcel name or existing UUID reference"""
+        self._parcels_with_associations.append(parcel)
+
     def build(self) -> FeatureProfileBuildReport:
         """
         Builds the feature profile.
@@ -93,24 +156,52 @@ def build(self) -> FeatureProfileBuildReport:
 
         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)
+        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
+
         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
+
+        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:
         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:
         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,
+    ):
+        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)

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):

catalystwan/models/configuration/feature_profile/sdwan/uc_voice/digital_interface.py

@@ -215,7 +215,7 @@ class OutgoingIe(BaseModel):
 ]
 
 
-class Associations(BaseModel):
+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")
@@ -232,7 +232,7 @@ class DigitalInterfaceParcel(_ParcelBase):
     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")
-    associations: Optional[List[Associations]] = Field(default=None, validation_alias=AliasPath("data", "associations"), description="Select Trunk Group and Translation Profile associations")
+    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")

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))