Merge pull request #518 from CiscoDevNet/dev/other/ucse

Add UCSE model, converter and integration test

catalystwan/integration_tests/feature_profile/sdwan/other/test_models.py

@@ -2,7 +2,9 @@
 import unittest
 from typing import cast
 
-from catalystwan.models.configuration.feature_profile.sdwan.other import ThousandEyesParcel
+from catalystwan.api.configuration_groups.parcel import Global, as_global
+from catalystwan.models.configuration.feature_profile.sdwan.other import ThousandEyesParcel, UcseParcel
+from catalystwan.models.configuration.feature_profile.sdwan.other.ucse import AccessPort, Imc, LomType, SharedLom
 from catalystwan.session import create_manager_session
 
 
@@ -27,6 +29,26 @@ def test_when_default_values_thousandeyes_parcel_expect_successful_post(self):
         # Assert
         assert parcel_id
 
+    def test_when_default_values_ucse_parcel_expect_successful_post(self):
+        # Arrange
+        ucse_parcel = UcseParcel(
+            parcel_name="UcseDefault",
+            parcel_description="Ucse Parcel",
+            bay=as_global(1),
+            slot=as_global(2),
+            imc=Imc(
+                access_port=AccessPort(
+                    shared_lom=SharedLom(
+                        lom_type=Global[LomType](value="te2"),
+                    )
+                )
+            ),
+        )
+        # Act
+        parcel_id = self.session.api.sdwan_feature_profiles.other.create(self.profile_id, ucse_parcel).id
+        # Assert
+        assert parcel_id
+
     def tearDown(self) -> None:
         self.session.api.sdwan_feature_profiles.other.delete_profile(self.profile_id)
         self.session.close()

catalystwan/models/configuration/feature_profile/sdwan/other/__init__.py

@@ -4,14 +4,16 @@
 from typing_extensions import Annotated
 
 from .thousandeyes import ThousandEyesParcel
+from .ucse import UcseParcel
 
 AnyOtherParcel = Annotated[
-    Union[ThousandEyesParcel,],  # noqa: #231
+    Union[ThousandEyesParcel, UcseParcel],  # noqa: #231
     Field(discriminator="type_"),
 ]
 
 __all__ = [
     "ThousandEyesParcel",
+    "UcseParcel",
     "AnyOtherParcel",
 ]
 

catalystwan/models/configuration/feature_profile/sdwan/other/ucse.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+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, as_default, as_variable
+
+FailOverType = Literal["ge2", "te2"]
+LomType = Literal["ge1", "ge2", "ge3", "te2", "te3", "console", "failover"]
+
+
+class SharedLom(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
+    lom_type: Global[LomType] = Field(..., serialization_alias="lomType", validation_alias="lomType")
+    fail_over_type: Optional[Global[FailOverType]] = Field(
+        default=None, serialization_alias="failOverType", validation_alias="failOverType"
+    )
+
+
+class AccessPort(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+    )
+    dedicated: Union[Global[bool], Default[bool]] = Field(default=as_default(True), description="Dedicated")
+    shared_lom: SharedLom = Field(..., serialization_alias="sharedLom", validation_alias="sharedLom")
+
+
+class Ip(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
+    address: Union[Global[str], Variable] = Field(
+        default=as_variable("{{ipv4Addr}}"), description="Assign IPv4 address"
+    )
+    default_gateway: Union[Global[IPv4Address], Variable, Default[None]] = Field(
+        default=Default[None](value=None),
+        serialization_alias="defaultGateway",
+        validation_alias="defaultGateway",
+        description="Assign default gateway",
+    )
+
+
+class Vlan(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+    )
+    vlan_id: Union[Global[int], Variable, Default[None]] = Field(
+        default=Default[None](value=None),
+        serialization_alias="vlanId",
+        validation_alias="vlanId",
+        description="Assign Vlan Id",
+    )
+    priority: Union[Global[int], Variable, Default[None]] = Field(
+        default=Default[None](value=None), description="Assign priority"
+    )
+
+
+class Imc(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+    )
+    access_port: AccessPort = Field(..., serialization_alias="access-port", validation_alias="access-port")
+    ip: Ip = Field(default_factory=Ip)
+    vlan: Optional[Vlan] = None
+
+
+class InterfaceItem(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
+    if_name: Union[Global[str], Variable, Default[None]] = Field(
+        default=Default[None](value=None),
+        serialization_alias="ifName",
+        validation_alias="ifName",
+        description="Set Inteface name",
+    )
+    l3: Default[bool] = Field(default=as_default(True), description="L3")
+    ucse_interface_vpn: Optional[Union[Global[int], Variable, Default[None]]] = Field(
+        default=Default[None](value=None),
+        serialization_alias="ucseInterfaceVpn",
+        validation_alias="ucseInterfaceVpn",
+        description="UCSE Interface VPN",
+    )
+    address: Optional[Union[Global[str], Variable]] = Field(default=None, description="Assign IPv4 address")
+
+
+class UcseParcel(_ParcelBase):
+    type_: Literal["ucse"] = Field(default="ucse", exclude=True)
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
+    bay: Global[int] = Field(..., validation_alias=AliasPath("data", "bay"), description="Bay")
+    slot: Global[int] = Field(..., validation_alias=AliasPath("data", "slot"), description="Slot")
+    imc: Optional[Imc] = Field(default=None, validation_alias=AliasPath("data", "imc"), description="IMC")
+    interface: Optional[List[InterfaceItem]] = Field(
+        default=None,
+        validation_alias=AliasPath("data", "interface"),
+        description="Interface name: GigabitEthernet0/<>/<> when present",
+    )

catalystwan/utils/config_migration/converters/feature_template/factory_method.py

@@ -20,6 +20,7 @@
 from .omp import OMPTemplateConverter
 from .security import SecurityTemplateConverter
 from .thousandeyes import ThousandEyesTemplateConverter
+from .ucse import UcseTemplateConverter
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +36,7 @@
     NTPTemplateConverter,
     BGPTemplateConverter,
     ThousandEyesTemplateConverter,
+    UcseTemplateConverter,
 ]
 
 

catalystwan/utils/config_migration/converters/feature_template/ucse.py

@@ -0,0 +1,51 @@
+from copy import deepcopy
+
+from catalystwan.api.configuration_groups.parcel import Global
+from catalystwan.models.configuration.feature_profile.sdwan.other import UcseParcel
+from catalystwan.models.configuration.feature_profile.sdwan.other.ucse import LomType
+
+
+class UcseTemplateConverter:
+    """
+    A class for converting template values into a UcseParcel object.
+    """
+
+    supported_template_types = ("ucse",)
+
+    @staticmethod
+    def create_parcel(name: str, description: str, template_values: dict) -> UcseParcel:
+        """
+        Creates a UcseParcel object based on the provided template values.
+
+        Args:
+            name (str): The name of the parcel.
+            description (str): The description of the parcel.
+            template_values (dict): A dictionary containing the template values.
+
+        Returns:
+            UcseParcel: A UcseParcel object with the provided values.
+        """
+        parcel_values = deepcopy(template_values)
+
+        for interface_values in parcel_values.get("interface", []):
+            ip = interface_values.pop("ip", None)
+            if ip:
+                interface_values["address"] = ip.get("static_case", {}).get("address")
+
+        imc = parcel_values.get("imc", {})
+        static_case = imc.get("ip", {}).get("static_case")
+        if static_case:
+            imc["ip"] = static_case
+
+        access_port = imc.get("access_port", {})
+        shared_lom = access_port.get("shared_lom")
+        if shared_lom:
+            lom_type = list(shared_lom.keys())[0]
+            shared_lom.clear()
+            access_port["shared_lom"]["lom_type"] = Global[LomType](value=lom_type)
+
+        for key in ["module_type", "subslot_name"]:
+            parcel_values.pop(key, None)
+
+        parcel_values.update({"parcel_name": name, "parcel_description": description})
+        return UcseParcel(**parcel_values)

catalystwan/workflows/config_migration.py

@@ -35,6 +35,7 @@
     "bgp",
     "cisco_bgp",
     "cisco_thousandeyes",
+    "ucse",
 ]