Add merge function for pim igmp and multicast, fix nested template da…

…ta structure, add multicast to transform fuction. add tests. This should be followed by PR that sets up logic for assinging subparcels to vpn

catalystwan/api/templates/device_template/device_template.py

@@ -54,26 +54,6 @@ class DeviceTemplate(BaseModel):
     )
     policy_id: str = Field(default="", serialization_alias="policyId", validation_alias="policyId")
 
-    def get_flattened_general_templates(self) -> List[GeneralTemplate]:
-        """
-        Recursively flattens the general templates by removing the sub-templates
-        and returning a list of flattened templates.
-
-        Returns:
-            A list of GeneralTemplate objects representing the flattened templates.
-        """
-
-        def flatten_general_templates(general_templates: List[GeneralTemplate]) -> List[GeneralTemplate]:
-            result = []
-            for gt in general_templates:
-                sub_templates = gt.subTemplates
-                gt.subTemplates = []
-                result.append(gt)
-                result.extend(flatten_general_templates(sub_templates))
-            return result
-
-        return flatten_general_templates(self.general_templates)
-
     def generate_payload(self) -> str:
         env = Environment(
             loader=FileSystemLoader(self.payload_path.parent),

catalystwan/integration_tests/feature_profile/sdwan/test_service.py

@@ -41,8 +41,8 @@
     PimInterfaceParameters,
     RPAnnounce,
     RpDiscoveryScope,
-    SmmFlag,
     SsmAttributes,
+    SsmFlag,
     StaticJoin,
     StaticRpAddress,
 )
@@ -300,7 +300,7 @@ def test_when_fully_specified_values_multicast_expect_successful_post(self):
                 ]
             ),
             pim=PimAttributes(
-                ssm=SsmAttributes(ssm_range_config=SmmFlag(enable_ssm_flag=as_global(True), range=as_global("20"))),
+                ssm=SsmAttributes(ssm_range_config=SsmFlag(enable_ssm_flag=as_global(True), range=as_global("20"))),
                 interface=[
                     PimInterfaceParameters(
                         interface_name=as_global("GigabitEthernet0/0/0"),

catalystwan/models/configuration/config_migration.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 from typing_extensions import Annotated
 
-from catalystwan.api.templates.device_template.device_template import DeviceTemplate
+from catalystwan.api.templates.device_template.device_template import DeviceTemplate, GeneralTemplate
 from catalystwan.endpoints.configuration_group import ConfigGroupCreationPayload
 from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload, ProfileType
 from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel
@@ -43,6 +43,24 @@ def from_merged(template: DeviceTemplate, info: TemplateInformation) -> "DeviceT
             **info_dict
         )
 
+    def get_flattened_general_templates(self) -> List[GeneralTemplate]:
+        """
+        Flatten the representation but leave cisco vpn templates as they are.
+
+        Returns:
+            A list of GeneralTemplate objects representing the flattened templates list.
+        """
+        result = []
+        for template in self.general_templates:
+            subtemplates = template.subTemplates
+            if subtemplates and template.templateType != "cisco_vpn":
+                template.subTemplates = []
+                for subtemplate in subtemplates:
+                    result.append(subtemplate)
+            result.append(template)
+
+        return result
+
 
 class UX1Policies(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
@@ -133,6 +151,8 @@ def insert_parcel_type_from_headers(cls, values: Dict[str, Any]):
         if not profile_parcels:
             profile_parcels = values.get("profile_parcels", [])
         for profile_parcel in profile_parcels:
+            if not isinstance(profile_parcel, dict):
+                break
             profile_parcel["parcel"]["type_"] = profile_parcel["header"]["type"]
         return values
 

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

@@ -12,10 +12,12 @@
 from .lan.ipsec import InterfaceIpsecParcel
 from .lan.svi import InterfaceSviParcel
 from .lan.vpn import LanVpnParcel
+from .multicast import MulticastParcel
 from .ospf import OspfParcel
 from .ospfv3 import Ospfv3IPv4Parcel, Ospfv3IPv6Parcel
 from .route_policy import RoutePolicyParcel
 from .switchport import SwitchportParcel
+from .wireless_lan import WirelessLanParcel
 
 AnyTopLevelServiceParcel = Annotated[
     Union[
@@ -30,6 +32,8 @@
         SwitchportParcel,
         Ipv6AclParcel,
         Ipv4AclParcel,
+        WirelessLanParcel,
+        MulticastParcel,
         # TrackerGroupData,
         # WirelessLanData,
         # SwitchportData
@@ -65,6 +69,8 @@
     "SwitchportParcel",
     "InterfaceSviParcel",
     "InterfaceGreParcel",
+    "WirelessLanParcel",
+    "MulticastParcel",
     "AnyServiceParcel",
     "AnyTopLevelServiceParcel",
     "AnyLanVpnInterfaceParcel",

catalystwan/models/configuration/feature_profile/sdwan/service/multicast.py

@@ -60,7 +60,7 @@ class IgmpAttributes(BaseModel):
     interface: List[IgmpInterfaceParameters]
 
 
-class SmmFlag(BaseModel):
+class SsmFlag(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
     enable_ssm_flag: Global[bool] = Field(
@@ -72,7 +72,7 @@ class SmmFlag(BaseModel):
 class SsmAttributes(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
-    ssm_range_config: SmmFlag = Field(serialization_alias="ssmRangeConfig", validation_alias="ssmRangeConfig")
+    ssm_range_config: SsmFlag = Field(serialization_alias="ssmRangeConfig", validation_alias="ssmRangeConfig")
     spt_threshold: Optional[Union[Global[SptThreshold], Variable, Default[SptThreshold]]] = Field(
         serialization_alias="sptThreshold",
         validation_alias="sptThreshold",

catalystwan/tests/config_migration/test_device_template_with_info.py

@@ -0,0 +1,92 @@
+import unittest
+
+from catalystwan.api.templates.device_template.device_template import GeneralTemplate
+from catalystwan.models.configuration.config_migration import DeviceTemplateWithInfo
+
+
+class TestDeviceTemplate(unittest.TestCase):
+    def setUp(self):
+        self.device_template = DeviceTemplateWithInfo(
+            template_id="1",
+            factory_default=False,
+            devices_attached=2,
+            template_name="DT-example",
+            template_description="DT-example",
+            device_role="None",
+            device_type="None",
+            security_policy_id="None",
+            policy_id="None",
+            generalTemplates=[
+                # Flat - stays flat
+                GeneralTemplate(name="1level", templateId="1", templateType="cedge_aaa", subTemplates=[]),
+                # No relation between system and logging in the UX2 - flatten them
+                GeneralTemplate(
+                    name="1level",
+                    templateId="2",
+                    templateType="cisco_system",
+                    subTemplates=[
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="3",
+                            templateType="cisco_logging",
+                            subTemplates=[],
+                        ),
+                    ],
+                ),
+                # Cisco VPN - keep the structure
+                GeneralTemplate(
+                    name="1level",
+                    templateId="4",
+                    templateType="cisco_vpn",
+                    subTemplates=[
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="5",
+                            templateType="cisco_vpn_interface",
+                            subTemplates=[],
+                        ),
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="6",
+                            templateType="cedge_multicast",
+                            subTemplates=[],
+                        ),
+                    ],
+                ),
+            ],
+        )
+
+    def test_flatten_general_templates(self):
+        self.maxDiff = None
+        self.assertEqual(
+            self.device_template.get_flattened_general_templates(),
+            [
+                GeneralTemplate(name="1level", templateId="1", templateType="cedge_aaa", subTemplates=[]),
+                GeneralTemplate(
+                    name="2level",
+                    templateId="3",
+                    templateType="cisco_logging",
+                    subTemplates=[],
+                ),
+                GeneralTemplate(name="1level", templateId="2", templateType="cisco_system", subTemplates=[]),
+                GeneralTemplate(
+                    name="1level",
+                    templateId="4",
+                    templateType="cisco_vpn",
+                    subTemplates=[
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="5",
+                            templateType="cisco_vpn_interface",
+                            subTemplates=[],
+                        ),
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="6",
+                            templateType="cedge_multicast",
+                            subTemplates=[],
+                        ),
+                    ],
+                ),
+            ],
+        )