IdModel concept and wrapping flow

catalystwan/tests/config_migration/test_flatten_general_templates.py

@@ -0,0 +1,49 @@
+import unittest
+
+from catalystwan.api.templates.device_template.device_template import DeviceTemplate, GeneralTemplate
+from catalystwan.utils.config_migration.device_templates import flatten_general_templates
+
+
+class TestFlattenGeneralTemplates(unittest.TestCase):
+    def setUp(self):
+        self.device_template = DeviceTemplate(
+            template_name="DT-example",
+            template_description="DT-example",
+            device_role="None",
+            device_type="None",
+            security_policy_id="None",
+            policy_id="None",
+            generalTemplates=[
+                GeneralTemplate(
+                    name="1level",
+                    templateId="1",
+                    templateType="1",
+                    subTemplates=[
+                        GeneralTemplate(
+                            name="2level",
+                            templateId="2",
+                            templateType="2",
+                            subTemplates=[GeneralTemplate(name="3level", templateId="3", templateType="3")],
+                        )
+                    ],
+                )
+            ],
+        )
+
+    def test_flatten_general_templates(self):
+        self.assertEqual(
+            flatten_general_templates(self.device_template.general_templates),
+            [
+                GeneralTemplate(
+                    name="1level",
+                    templateId="1",
+                    templateType="1",
+                ),
+                GeneralTemplate(
+                    name="2level",
+                    templateId="2",
+                    templateType="2",
+                ),
+                GeneralTemplate(name="3level", templateId="3", templateType="3"),
+            ],
+        )

catalystwan/utils/config_migration/device_templates/__init__.py

@@ -0,0 +1,11 @@
+from typing import List
+
+from .flatten_general_templates import flatten_general_templates
+
+__all__ = [
+    "flatten_general_templates",
+]
+
+
+def __dir__() -> "List[str]":
+    return list(__all__)

catalystwan/utils/config_migration/device_templates/flatten_general_templates.py

@@ -0,0 +1,22 @@
+from typing import List
+
+from catalystwan.api.templates.device_template.device_template import GeneralTemplate
+
+
+def flatten_general_templates(general_templates: List[GeneralTemplate]) -> List[GeneralTemplate]:
+    """
+    Recursively flattens a list of GeneralTemplate objects.
+
+    Args:
+        general_templates (List[GeneralTemplate]): The list of GeneralTemplate objects to flatten.
+
+    Returns:
+        List[GeneralTemplate]: The flattened list of GeneralTemplate objects.
+    """
+    result = []
+    for gt in general_templates:
+        sub_templates = gt.subTemplates
+        gt.subTemplates = []
+        result.append(gt)
+        result.extend(flatten_general_templates(sub_templates))
+    return result

catalystwan/workflows/config_migration.py

@@ -1,13 +1,18 @@
 import logging
-from typing import Callable
+from typing import Any, Callable, List, Optional
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel
 
 from catalystwan.api.policy_api import POLICY_LIST_ENDPOINTS_MAP
-from catalystwan.endpoints.configuration_group import ConfigGroup
+from catalystwan.endpoints.configuration_group import ConfigGroup, ConfigGroupCreationPayload
 from catalystwan.models.configuration.config_migration import UX1Config, UX2Config
+from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
 from catalystwan.session import ManagerSession
 from catalystwan.utils.config_migration.converters.feature_template import create_parcel_from_template
 from catalystwan.utils.config_migration.converters.policy.policy_lists import convert_all as convert_policy_lists
 from catalystwan.utils.config_migration.creators.config_group import ConfigGroupCreator
+from catalystwan.utils.config_migration.device_templates import flatten_general_templates
 
 logger = logging.getLogger(__name__)
 
@@ -41,17 +46,130 @@
     "cisco_dhcp_server",
 ]
 
+FEATURE_PROFILE_SYSTEM = [
+    "cisco_aaa",
+    "cedge_aaa",
+    "aaa",
+    "cisco_banner",
+    "cisco_security",
+    "security",
+    "security-vsmart",
+    "security-vedge",
+    "cisco_system",
+    "system-vsmart",
+    "system-vedge",
+    "cisco_bfd",
+    "bfd-vedge",
+    "cedge_global",
+    "cisco_logging",
+    "logging",
+    "cisco_omp",
+    "omp-vedge",
+    "omp-vsmart",
+    "cisco_ntp",
+    "ntp",
+    "bgp",
+    "cisco_bgp",
+]
+
+FEATURE_PROFILE_TRANSPORT = [
+    "dhcp",
+    "cisco_dhcp_server",
+]
+
+FEATURE_PROFILE_OTHER = [
+    "cisco_thousandeyes",
+    "ucse",
+]
+
 
 def log_progress(task: str, completed: int, total: int) -> None:
     logger.info(f"{task} {completed}/{total}")
 
 
+class IdModel(BaseModel):
+    type: str
+    childs: Optional[List[UUID]] = None
+    id: UUID
+    model: Any
+
+
+def transform2(ux1: UX1Config) -> Any:
+    ux2 = UX2Config()
+    # Create Feature Profiles and Config Group
+    for dt in ux1.templates.device_templates:
+        templates = flatten_general_templates(dt.general_templates)
+
+        # Create Feature Profiles
+        fp_system_uuid = uuid4()
+        fp_system = IdModel(
+            type="feature_profile_system",
+            id=fp_system_uuid,
+            model=FeatureProfileCreationPayload(
+                name="system",
+                description="system",
+            ),
+        )
+        fp_transport_uuid = uuid4()
+        fp_transport = IdModel(
+            type="feature_profile_transport",
+            id=fp_transport_uuid,
+            model=FeatureProfileCreationPayload(
+                name="transport",
+                description="transport",
+            ),
+        )
+
+        fp_other_uuid = uuid4()
+        fp_other = IdModel(
+            type="feature_profile_other",
+            id=fp_other_uuid,
+            model=FeatureProfileCreationPayload(
+                name="other",
+                description="other",
+            ),
+        )
+
+        for template in templates:
+            # Those feature templates IDs are real UUIDs and are used to map to the feature profiles
+            if template.templateType in FEATURE_PROFILE_SYSTEM:
+                fp_system.childs.append(template.templateId)
+            elif template.templateType in FEATURE_PROFILE_TRANSPORT:
+                fp_transport.childs.append(template.templateId)
+            elif template.templateType in FEATURE_PROFILE_OTHER:
+                fp_other.childs.append(template.templateId)
+
+        cg = IdModel(
+            type="config_group",
+            id=uuid4(),
+            childs=[fp_system_uuid, fp_transport_uuid, fp_other_uuid],
+            model=ConfigGroupCreationPayload(
+                name=dt.template_name,
+                description=dt.template_description,
+                solution="sdwan",
+                profiles=[],
+            ),
+        )
+        # Add to UX2
+        ux2.feature_profiles.append(fp_system)
+        ux2.feature_profiles.append(fp_transport)
+        ux2.feature_profiles.append(fp_other)
+        ux2.config_groups.append(cg)
+
+    for ft in ux1.templates.feature_templates:
+        if ft.template_type in SUPPORTED_TEMPLATE_TYPES:
+            model = create_parcel_from_template(ft)
+            # Using real UUIDs for the parcels. So creation overhead is reducted and they aleary mapp to the feature profiles
+            ux2.profile_parcels.append(IdModel(type=model._get_parcel_type(), id=ft.id, model=model))
+
+
 def transform(ux1: UX1Config) -> UX2Config:
     ux2 = UX2Config()
     # Feature Templates
     for ft in ux1.templates.feature_templates:
         if ft.template_type in SUPPORTED_TEMPLATE_TYPES:
-            ux2.profile_parcels.append(create_parcel_from_template(ft))
+            model = create_parcel_from_template(ft)
+            ux2.profile_parcels.append(IdModel(type=model._get_parcel_type(), id=ft.id, model=model))
     # Policy Lists
     ux2.profile_parcels.extend(convert_policy_lists(ux1.policies.policy_lists).output)
     return ux2