Dev/migration templates (#487)

* draft - save work * draft * bump dev version * migrate ConfigGroup to pydantic v2 * bump dev version * Created Factory and adapters for model migration. Modeled templates for UX2 - BFD and AAA * use typing.List * Refactor. Add Logging model * Plural form in features * fix variable --------- Co-authored-by: sbasan <[email protected]> Co-authored-by: Kuba <[email protected]>
CiscoDevNet · Feb 23, 2024 · 2239e37 · 2239e37
1 parent 671370a
commit 2239e37
Show file tree

Hide file tree

Showing 25 changed files with 641 additions and 45 deletions.
diff --git a/catalystwan/api/configuration_groups/parcel.py b/catalystwan/api/configuration_groups/parcel.py
@@ -21,10 +21,10 @@ class _ParcelBase(BaseModel):
         validation_alias="description",
         description="Set the parcel description",
     )
-    data: Optional[Any] = None
+    # data: Optional[Any] = None
     _parcel_data_key: str = PrivateAttr(default="data")
 
-    @model_serializer(mode="wrap")
+    @model_serializer(mode="wrap", when_used="json")
     def envelope_parcel_data(self, handler) -> Dict[str, Any]:
         model_dict = handler(self)
         model_dict[self._parcel_data_key] = {}
@@ -49,7 +49,7 @@ class OptionType(str, Enum):
 
 
 class ParcelAttribute(BaseModel):
-    model_config = ConfigDict(extra="forbid")
+    model_config = ConfigDict(extra="forbid", populate_by_name=True)
     option_type: OptionType = Field(serialization_alias="optionType", validation_alias="optionType")
 
 

diff --git a/catalystwan/api/policy_api.py b/catalystwan/api/policy_api.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Type, overload
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Tuple, Type, overload
 from uuid import UUID
 
 from catalystwan.api.task_status_api import Task
@@ -637,6 +637,12 @@ def get(self, type: Type[AnyPolicyList], id: Optional[UUID] = None) -> Any:
             return endpoints.get_lists_by_id(id=id)
         return endpoints.get_policy_lists()
 
+    def get_all(self) -> List[AnyPolicyList]:
+        infos: List[AnyPolicyList] = []
+        for list_type, _ in POLICY_LIST_ENDPOINTS_MAP.items():
+            infos.extend(self.get(list_type))
+        return infos
+
 
 class PolicyDefinitionsAPI:
     def __init__(self, session: ManagerSession):
@@ -780,6 +786,12 @@ def get(self, type: Type[AnyPolicyDefinition], id: Optional[UUID] = None) -> Any
             return endpoints.get_policy_definition(id=id)
         return endpoints.get_definitions()
 
+    def get_all(self) -> List[Tuple[type, PolicyDefinitionInfo]]:
+        all_items: List[Tuple[type, PolicyDefinitionInfo]] = []
+        for definition_type, _ in POLICY_DEFINITION_ENDPOINTS_MAP.items():
+            all_items.extend([(definition_type, info) for info in self.get(definition_type)])
+        return all_items
+
 
 class PolicyAPI:
     """This is exposing so called 'UX 1.0' API"""

diff --git a/catalystwan/api/template_api.py b/catalystwan/api/template_api.py
@@ -1,11 +1,13 @@
 from __future__ import annotations
 
+import datetime as dt
 import json
 import logging
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Optional, Type, overload
+from typing import TYPE_CHECKING, Any, List, Optional, Type, overload
 
 from ciscoconfparse import CiscoConfParse  # type: ignore
+from pydantic import BaseModel, ConfigDict, Field
 
 from catalystwan.api.task_status_api import Task
 from catalystwan.api.templates.cli_template import CLITemplate
@@ -67,6 +69,41 @@ class DeviceTemplateFeature(Enum):
     ALL = "all"
 
 
+class TemplateInformation(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    last_updated_by: str = Field(serialization_alias="lastUpdatedBy", validation_alias="lastUpdatedBy")
+    id: str = Field(serialization_alias="templateId", validation_alias="templateId")
+    factory_default: bool = Field(serialization_alias="factoryDefault", validation_alias="factoryDefault")
+    name: str = Field(serialization_alias="templateName", validation_alias="templateName")
+    devices_attached: int = Field(serialization_alias="devicesAttached", validation_alias="devicesAttached")
+    description: str = Field(serialization_alias="templateDescription", validation_alias="templateDescription")
+    last_updated_on: dt.datetime = Field(serialization_alias="lastUpdatedOn", validation_alias="lastUpdatedOn")
+    resource_group: Optional[str] = Field(None, serialization_alias="resourceGroup", validation_alias="resourceGroup")
+
+
+class FeatureTemplateInformation(TemplateInformation):
+    model_config = ConfigDict(populate_by_name=True)
+
+    template_type: str = Field(serialization_alias="templateType", validation_alias="templateType")
+    device_type: List[str] = Field(serialization_alias="deviceType", validation_alias="deviceType")
+    version: str = Field(serialization_alias="templateMinVersion", validation_alias="templateMinVersion")
+    template_definiton: Optional[str] = Field(
+        None, serialization_alias="templateDefinition", validation_alias="templateDefinition"
+    )
+
+
+class DeviceTemplateInformation(TemplateInformation):
+    model_config = ConfigDict(populate_by_name=True)
+
+    device_type: str = Field(serialization_alias="deviceType", validation_alias="deviceType")
+    template_class: str = Field(serialization_alias="templateClass", validation_alias="templateClass")
+    config_type: str = Field(serialization_alias="configType", validation_alias="configType")
+    template_attached: int = Field(serialization_alias="templateAttached", validation_alias="templateAttached")
+    draft_mode: Optional[str] = Field(None, serialization_alias="draftMode", validation_alias="draftMode")
+    device_role: Optional[str] = Field(None, serialization_alias="deviceRole", validation_alias="deviceRole")
+
+
 class TemplatesAPI:
     def __init__(self, session: ManagerSession) -> None:
         self.session = session
@@ -691,3 +728,14 @@ def load_running(self, device: Device) -> CiscoConfParse:
         config = CiscoConfParse(response["config"].splitlines())
         logger.debug(f"Template loaded from {device.hostname}.")
         return config
+
+    def get_feature_templates(self) -> DataSequence[FeatureTemplateInformation]:
+        endpoint = "/dataservice/template/feature"
+        fr_templates = self.session.get(endpoint)
+        return fr_templates.dataseq(FeatureTemplateInformation)
+
+    def get_device_templates(self) -> DataSequence[DeviceTemplateInformation]:
+        endpoint = "/dataservice/template/device"
+        params = {"feature": "all"}
+        templates = self.session.get(url=endpoint, params=params)
+        return templates.dataseq(DeviceTemplateInformation)
diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/system.py
@@ -44,6 +44,11 @@ def delete_sdwan_system_feature_profile(self, system_id: str) -> None:
     def create_aaa_profile_parcel_for_system(self, system_id: str, payload: _ParcelBase) -> ParcelId:
         ...
 
+    @versions(supported_versions=(">=20.9"), raises=False)
+    @post("/v1/feature-profile/sdwan/system/{system_id}/bfd")
+    def create_bfd_profile_parcel_for_system(self, system_id: str, payload: _ParcelBase) -> ParcelId:
+        ...
+
     @versions(supported_versions=(">=20.9"), raises=False)
     @put("/v1/feature-profile/sdwan/system/{system_id}/aaa/{parcel_id}")
     def edit_aaa_profile_parcel_for_system(self, system_id: str, parcel_id: str, payload: _ParcelBase) -> ParcelId:

diff --git a/catalystwan/endpoints/configuration_group.py b/catalystwan/endpoints/configuration_group.py
@@ -2,7 +2,7 @@
 from datetime import datetime
 from typing import List, Optional
 
-from pydantic.v1 import BaseModel, Field
+from pydantic import BaseModel, Field
 
 from catalystwan.endpoints import APIEndpoints, delete, get, post, put, versions
 from catalystwan.models.configuration.common import Solution

diff --git a/catalystwan/models/configuration/config_migration.py b/catalystwan/models/configuration/config_migration.py
@@ -1,7 +1,11 @@
 from typing import List
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
+from catalystwan.api.configuration_groups.parcel import _ParcelBase
+from catalystwan.api.template_api import DeviceTemplateInformation, FeatureTemplateInformation
+from catalystwan.endpoints.configuration_group import ConfigGroup
+from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
 from catalystwan.models.policy import (
     AnyPolicyDefinition,
     AnyPolicyList,
@@ -12,23 +16,48 @@
 
 
 class UX1Policies(BaseModel):
-    centralized_policies: List[CentralizedPolicy] = Field(default=[], serialization_alias="centralizedPolicies")
-    localized_policies: List[LocalizedPolicy] = Field(default=[], serialization_alias="localizedPolicies")
-    security_policies: List[SecurityPolicy] = Field(default=[], serialization_alias="securityPolicies")
-    policy_definitions: List[AnyPolicyDefinition] = Field(default=[], serialization_alias="policyDefinitions")
-    policy_lists: List[AnyPolicyList] = Field(default=[], serialization_alias="policyLists")
+    model_config = ConfigDict(populate_by_name=True)
+    centralized_policies: List[CentralizedPolicy] = Field(
+        default=[], serialization_alias="centralizedPolicies", validation_alias="centralizedPolicies"
+    )
+    localized_policies: List[LocalizedPolicy] = Field(
+        default=[], serialization_alias="localizedPolicies", validation_alias="localizedPolicies"
+    )
+    security_policies: List[SecurityPolicy] = Field(
+        default=[], serialization_alias="securityPolicies", validation_alias="securityPolicies"
+    )
+    policy_definitions: List[AnyPolicyDefinition] = Field(
+        default=[], serialization_alias="policyDefinitions", validation_alias="policyDefinitions"
+    )
+    policy_lists: List[AnyPolicyList] = Field(
+        default=[], serialization_alias="policyLists", validation_alias="policyLists"
+    )
 
 
 class UX1Templates(BaseModel):
-    pass
+    features: List[FeatureTemplateInformation] = Field(default=[])
+    devices: List[DeviceTemplateInformation] = Field(default=[])
 
 
 class UX1Config(BaseModel):
     # All UX1 Configuration items - Mega Model
-    policies: UX1Policies
-    templates: UX1Templates
+    model_config = ConfigDict(populate_by_name=True)
+    policies: UX1Policies = UX1Policies()
+    templates: UX1Templates = UX1Templates()
 
 
 class UX2Config(BaseModel):
     # All UX2 Configuration items - Mega Model
-    pass
+    model_config = ConfigDict(populate_by_name=True)
+    config_groups: List[ConfigGroup] = Field(
+        default=[], serialization_alias="configurationGroups", validation_alias="configurationGroups"
+    )
+    policy_groups: List[ConfigGroup] = Field(
+        default=[], serialization_alias="policyGroups", validation_alias="policyGroups"
+    )
+    feature_profiles: List[FeatureProfileCreationPayload] = Field(
+        default=[], serialization_alias="featureProfiles", validation_alias="featureProfiles"
+    )
+    profile_parcels: List[_ParcelBase] = Field(
+        default=[], serialization_alias="profileParcels", validation_alias="profileParcels"
+    )
diff --git a/catalystwan/models/configuration/feature_profile/converters/feature_template/__init__.py b/catalystwan/models/configuration/feature_profile/converters/feature_template/__init__.py
@@ -0,0 +1,58 @@
+from typing import Any, Dict, List
+
+from catalystwan.api.template_api import FeatureTemplateInformation
+from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel
+
+from .aaa import AAATemplateConverter
+from .base import FeatureTemplateConverter
+from .bfd import BFDTemplateConverter
+from .normalizator import template_definition_normalization
+
+supported_parcel_converters: Dict[Any, FeatureTemplateConverter] = {
+    ("cisco_aaa", "cedge_aaa"): AAATemplateConverter,  # type: ignore[dict-item]
+    ("cisco_bfd",): BFDTemplateConverter,  # type: ignore[dict-item]
+}
+
+
+def choose_parcel_converter(template_type: str) -> FeatureTemplateConverter:
+    """
+    This function is used to choose the correct parcel factory based on the template type.
+
+    Args:
+        template_type (str): The template type used to determine the correct factory.
+
+    Returns:
+        BaseFactory: The chosen parcel factory.
+
+    Raises:
+        ValueError: If the template type is not supported.
+    """
+    for key in supported_parcel_converters.keys():
+        if template_type in key:
+            return supported_parcel_converters[key]
+    raise ValueError(f"Template type {template_type} not supported")
+
+
+def create_parcel_from_template(template: FeatureTemplateInformation) -> AnySystemParcel:
+    """
+    Creates a new instance of a _ParcelBase based on the given template.
+
+    Args:
+        template (FeatureTemplateInformation): The template to use for creating the _ParcelBase instance.
+
+    Returns:
+        _ParcelBase: The created _ParcelBase instance.
+
+    Raises:
+        ValueError: If the given template type is not supported.
+    """
+    converter = choose_parcel_converter(template.template_type)
+    template_values = template_definition_normalization(template.template_definiton)
+    return converter.create_parcel(template.name, template.description, template_values)
+
+
+__all__ = ["create_parcel_from_template"]
+
+
+def __dir__() -> "List[str]":
+    return list(__all__)
diff --git a/catalystwan/models/configuration/feature_profile/converters/feature_template/aaa.py b/catalystwan/models/configuration/feature_profile/converters/feature_template/aaa.py
@@ -0,0 +1,30 @@
+from catalystwan.models.configuration.feature_profile.sdwan.system import AAA
+
+
+class AAATemplateConverter:
+    @staticmethod
+    def create_parcel(name: str, description: str, template_values: dict) -> AAA:
+        """
+        Creates an AAA object based on the provided template values.
+
+        Returns:
+            AAA: An AAA object with the provided template values.
+        """
+        template_values["name"] = name
+        template_values["description"] = description
+
+        delete_properties = (
+            "radius_client",
+            "radius_trustsec_group",
+            "rda_server_key",
+            "domain_stripping",
+            "auth_type",
+            "port",
+            "cts_auth_list",
+        )
+
+        for prop in delete_properties:
+            if template_values.get(prop) is not None:
+                del template_values[prop]
+
+        return AAA(**template_values)
diff --git a/catalystwan/models/configuration/feature_profile/converters/feature_template/base.py b/catalystwan/models/configuration/feature_profile/converters/feature_template/base.py
@@ -0,0 +1,9 @@
+from typing_extensions import Protocol
+
+from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel
+
+
+class FeatureTemplateConverter(Protocol):
+    @staticmethod
+    def create_parcel(name: str, description: str, template_values: dict) -> AnySystemParcel:
+        ...
diff --git a/catalystwan/models/configuration/feature_profile/converters/feature_template/bfd.py b/catalystwan/models/configuration/feature_profile/converters/feature_template/bfd.py
@@ -0,0 +1,20 @@
+from catalystwan.models.configuration.feature_profile.sdwan.system import BFD
+
+
+class BFDTemplateConverter:
+    @staticmethod
+    def create_parcel(name: str, description: str, template_values: dict) -> BFD:
+        """
+        Creates an BFD object based on the provided template values.
+
+        Returns:
+            BFD: An BFD object with the provided template values.
+        """
+        template_values["name"] = name
+        template_values["description"] = description
+
+        if template_values.get("color") is not None:
+            template_values["colors"] = template_values["color"]
+            del template_values["color"]
+
+        return BFD(**template_values)
diff --git a/catalystwan/models/configuration/feature_profile/converters/feature_template/logging.py b/catalystwan/models/configuration/feature_profile/converters/feature_template/logging.py
@@ -0,0 +1,27 @@
+# from catalystwan.models.configuration.feature_profile.sdwan.system import Logging
+
+# class LoggingTemplateConverter:
+
+#     @staticmethod
+#     def create_parcel(name, description, template_values: dict):
+#         """
+#         Creates an Logging object based on the provided template values.
+
+#         Returns:
+#             Logging: An Logging object with the provided template values.
+#         """
+#         template_values["name"] = name
+#         template_values["description"] = description
+
+#         template_values["disk"] = {
+#             "disk_enable": template_values["enable"],
+#             "file": {
+#                 "disk_file_size": template_values["size"],
+#                 "disk_file_rotate": template_values["rotate"]
+#             }
+#         }
+#         del template_values["enable"]
+#         del template_values["size"]
+#         del template_values["rotate"]
+
+#         return Logging(**template_values)