Refactor Service VPN converter. Instantiate the converter in the crea…

…te_parcel_from_template function. Convert the create_parcel methods to instance methods, allowing them to access class attributes and methods. Split large create_parcel functions

catalystwan/utils/config_migration/converters/exceptions.py

@@ -0,0 +1,9 @@
+from catalystwan.exceptions import CatalystwanException
+
+
+class CatalystwanConverterCantConvertException(CatalystwanException):
+    """
+    Exception raised when a CatalystwanConverter can't correctly convert a template.
+    """
+
+    pass

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

@@ -8,8 +8,7 @@
 class AAATemplateConverter:
     supported_template_types = ("cisco_aaa", "cedge_aaa", "aaa")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> AAAParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> AAAParcel:
         """
         Creates an AAAParcel object based on the provided template values.
 

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

@@ -7,13 +7,13 @@
     ServiceNodeGroupName,
     ServiceNodeGroupsNames,
 )
+from catalystwan.utils.config_migration.converters.exceptions import CatalystwanConverterCantConvertException
 
 
 class AppqoeTemplateConverter:
     supported_template_types = ("appqoe",)
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> AppqoeParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> AppqoeParcel:
         """
         Create an AppqoeParcel object based on the provided name, description, and template values.
 
@@ -26,7 +26,21 @@ def create_parcel(name: str, description: str, template_values: dict) -> AppqoeP
             AppqoeParcel: The created AppqoeParcel object.
         """
         values = deepcopy(template_values)
-        print(values)
+
+        appnav_controller_group = values.get("appnav_controller_group", [])
+        if not appnav_controller_group:
+            raise CatalystwanConverterCantConvertException("Appnav controller group is required for Appqoe parcel")
+        for appnav in appnav_controller_group:
+            if group_name := appnav.get("group_name"):
+                appnav["group_name"] = as_default(group_name.value, AppnavControllerGroupName)
+            for controller in appnav.get("appnav_controllers", []):
+                if _vpn := controller.get("vpn"):  # noqa: F841
+                    # VPN field is depended on existence of the Service VPN value
+                    # also from UI this list contains only 1 item and should not be a list.
+                    # AppqoeParcel.forwarder.appnav_controller_group.appnav_controllers[0].vpn
+                    # must be populated in the parcel creation process.
+                    pass
+
         for appqoe_item in values.get("service_context", {}).get("appqoe", []):
             if item_name := appqoe_item.get("name"):
                 appqoe_item["name"] = as_default(value=item_name.value)
@@ -46,16 +60,7 @@ def create_parcel(name: str, description: str, template_values: dict) -> AppqoeP
             internal = group.get("internal")
             if internal is not None:
                 group["internal"] = as_default(internal.value)
-        for appnav in values.get("appnav_controller_group", []):
-            if group_name := appnav.get("group_name"):
-                appnav["group_name"] = as_default(group_name.value, AppnavControllerGroupName)
-            for controller in appnav.get("appnav_controllers", []):
-                if _vpn := controller.get("vpn"):  # noqa: F841
-                    # VPN field is depended on existence of the Service VPN value
-                    # also from UI this list contains only 1 item and should not be a list.
-                    # AppqoeParcel.forwarder.appnav_controller_group.appnav_controllers[0].vpn
-                    # must be populated in the parcel creation process.
-                    pass
+
         parcel_values = {
             "parcel_name": name,
             "parcel_description": description,

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

@@ -4,8 +4,7 @@
 class BannerTemplateConverter:
     supported_template_types = ("cisco_banner",)
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> BannerParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> BannerParcel:
         """
         Creates a BannerParcel object based on the provided template values.
 

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

@@ -4,6 +4,5 @@
 
 
 class FeatureTemplateConverter(Protocol):
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> AnySystemParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> AnySystemParcel:
         ...

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

@@ -9,8 +9,7 @@
 class SystemToBasicTemplateConverter:
     supported_template_types = ("cisco_system", "system-vsmart", "system-vedge")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> BasicParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> BasicParcel:
         """
         Converts the provided template values into a BasicParcel object.
 

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

@@ -4,8 +4,7 @@
 class BFDTemplateConverter:
     supported_template_types = ("cisco_bfd", "bfd-vedge")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> BFDParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> BFDParcel:
         """
         Creates a BFDParcel object based on the provided template values.
 

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

@@ -17,8 +17,10 @@
 class BGPTemplateConverter:
     supported_template_types = ("bgp", "cisco_bgp")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> BGPParcel:
+    device_specific_ipv4_neighbor_address = "{{{{lbgp_1_neighbor_{index}_address}}}}"
+    device_specific_ipv6_neighbor_address = "{{{{lbgp_1_ipv6_neighbor_{index}_address}}}}"
+
+    def create_parcel(self, name: str, description: str, template_values: dict) -> BGPParcel:
         """
         Creates a BannerParcel object based on the provided template values.
 
@@ -30,8 +32,6 @@ def create_parcel(name: str, description: str, template_values: dict) -> BGPParc
         Returns:
             BannerParcel: A BannerParcel object with the provided template values.
         """
-        device_specific_ipv4_neighbor_address = "{{{{lbgp_1_neighbor_{index}_address}}}}"
-        device_specific_ipv6_neighbor_address = "{{{{lbgp_1_ipv6_neighbor_{index}_address}}}}"
 
         parcel_values = {"parcel_name": name, "parcel_description": description, **deepcopy(template_values["bgp"])}
 
@@ -48,7 +48,7 @@ def create_parcel(name: str, description: str, template_values: dict) -> BGPParc
                         family_type["family_type"] = as_global(family_type["family_type"].value, FamilyType)
                 if neighbor.get("address") is None:
                     logger.info("Neighbor address is not set, using device specific variable")
-                    neighbor["address"] = as_variable(device_specific_ipv4_neighbor_address.format(index=(i + 1)))
+                    neighbor["address"] = as_variable(self.device_specific_ipv4_neighbor_address.format(index=(i + 1)))
                 if if_name := neighbor.get("update_source", {}).get("if_name"):
                     neighbor["if_name"] = if_name
                     neighbor.pop("update_source")
@@ -73,7 +73,7 @@ def create_parcel(name: str, description: str, template_values: dict) -> BGPParc
                             )
                 if neighbor.get("address") is None:
                     logger.info("Neighbor address is not set, using device specific variable")
-                    neighbor["address"] = as_variable(device_specific_ipv6_neighbor_address.format(index=(i + 1)))
+                    neighbor["address"] = as_variable(self.device_specific_ipv6_neighbor_address.format(index=(i + 1)))
                 if if_name := neighbor.get("update_source", {}).get("if_name"):
                     neighbor["if_name"] = if_name
                     neighbor.pop("update_source")

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

@@ -20,8 +20,7 @@ class DhcpTemplateConverter:
     variable_mac_address = "{{{{dhcp_1_staticLease_{}_macAddress}}}}"
     variable_ip = "{{{{dhcp_1_staticLease_{}_ip}}}}"
 
-    @classmethod
-    def create_parcel(cls, name: str, description: str, template_values: dict) -> LanVpnDhcpServerParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> LanVpnDhcpServerParcel:
         """
         Create a LanVpnDhcpServerParcel object based on the provided parameters.
 
@@ -50,19 +49,19 @@ def create_parcel(cls, name: str, description: str, template_values: dict) -> La
                 "Assiging variable: dhcp_1_addressPool_networkAddress and dhcp_1_addressPool_subnetMask."
             )
             values["address_pool"] = {
-                "network_address": as_variable(cls.variable_address_pool),
-                "subnet_mask": as_variable(cls.variable_subnet_mask),
+                "network_address": as_variable(self.variable_address_pool),
+                "subnet_mask": as_variable(self.variable_subnet_mask),
             }
 
         for entry in values.get("option_code", []):
-            cls._convert_str_list_to_ipv4_list(entry, "ip")
+            self._convert_str_list_to_ipv4_list(entry, "ip")
 
         for key in ("dns_servers", "tftp_servers"):
-            cls._convert_str_list_to_ipv4_list(values, key)
+            self._convert_str_list_to_ipv4_list(values, key)
 
         static_lease = []
         for i, entry in enumerate(values.get("static_lease", [])):
-            mac_address, ip = cls._get_mac_address_and_ip(entry, i)
+            mac_address, ip = self._get_mac_address_and_ip(entry, i)
             static_lease.append(
                 {
                     "mac_address": mac_address,
@@ -79,8 +78,7 @@ def create_parcel(cls, name: str, description: str, template_values: dict) -> La
 
         return LanVpnDhcpServerParcel(**parcel_values)  # type: ignore
 
-    @classmethod
-    def _convert_str_list_to_ipv4_list(cls, d: dict, key: str) -> None:
+    def _convert_str_list_to_ipv4_list(self, d: dict, key: str) -> None:
         """
         Convert a list of strings representing IPv4 addresses to a list of IPv4Address objects.
 
@@ -95,19 +93,18 @@ def _convert_str_list_to_ipv4_list(cls, d: dict, key: str) -> None:
         if str_list := d.get(key, as_global([])).value:
             d[key] = Global[List[IPv4Address]](value=[IPv4Address(ip) for ip in str_list])
 
-    @classmethod
-    def _get_mac_address_and_ip(cls, entry: dict, i: int) -> tuple:
-        mac_address = entry.get("mac_address", as_variable(cls.variable_mac_address.format(i + 1)))
-        ip = entry.get("ip", as_variable(cls.variable_ip.format(i + 1)))
+    def _get_mac_address_and_ip(self, entry: dict, i: int) -> tuple:
+        mac_address = entry.get("mac_address", as_variable(self.variable_mac_address.format(i + 1)))
+        ip = entry.get("ip", as_variable(self.variable_ip.format(i + 1)))
         if isinstance(mac_address, Variable):
             logger.warning(
                 f"No MAC address specified for static lease {i + 1}."
-                f"Assigning variable: {cls.variable_mac_address.format(i + 1)}"
+                f"Assigning variable: {self.variable_mac_address.format(i + 1)}"
             )
         if isinstance(ip, Variable):
             logger.warning(
                 f"No IP address specified for static lease {i + 1}."
-                f"Assigning variable: {cls.variable_ip.format(i + 1)}"
+                f"Assigning variable: {self.variable_ip.format(i + 1)}"
             )
 
         return mac_address, ip

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

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Dict, cast
+from typing import Any, Callable, Dict, cast
 
 from catalystwan.api.template_api import FeatureTemplateInformation
 from catalystwan.exceptions import CatalystwanException
@@ -53,7 +53,7 @@
 }
 
 
-def choose_parcel_converter(template_type: str) -> FeatureTemplateConverter:
+def choose_parcel_converter(template_type: str) -> Callable[..., FeatureTemplateConverter]:
     """
     This function is used to choose the correct parcel factory based on the template type.
 
@@ -87,7 +87,7 @@ def create_parcel_from_template(template: FeatureTemplateInformation) -> AnySyst
     Raises:
         ValueError: If the given template type is not supported.
     """
-    converter = choose_parcel_converter(template.template_type)
+    converter = choose_parcel_converter(template.template_type)()
     template_definition_as_dict = json.loads(cast(str, template.template_definiton))
     template_values = find_template_values(template_definition_as_dict)
     template_values_normalized = template_definition_normalization(template_values)

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

@@ -4,8 +4,7 @@
 class GlobalTemplateConverter:
     supported_template_types = ("cedge_global",)
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> GlobalParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> GlobalParcel:
         """
         Creates an Logging object based on the provided template values.
 

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

@@ -9,8 +9,7 @@
 class LoggingTemplateConverter:
     supported_template_types = ("cisco_logging", "logging")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> LoggingParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> LoggingParcel:
         """
         Creates an Logging object based on the provided template values.
 

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

@@ -3,6 +3,8 @@
 
 from catalystwan.api.configuration_groups.parcel import Global, as_global
 from catalystwan.models.common import TLOCColor
+from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import SubnetMask
+from catalystwan.models.configuration.feature_profile.sdwan.service.lan.vpn import Direction
 from catalystwan.models.configuration.feature_profile.sdwan.system.logging_parcel import (
     AuthType,
     CypherSuite,
@@ -11,7 +13,17 @@
 )
 from catalystwan.models.configuration.feature_profile.sdwan.system.mrf import EnableMrfMigration, Role
 
-CastableLiterals = [Priority, TlsVersion, AuthType, CypherSuite, Role, EnableMrfMigration, TLOCColor]
+CastableLiterals = [
+    Priority,
+    TlsVersion,
+    AuthType,
+    CypherSuite,
+    Role,
+    EnableMrfMigration,
+    TLOCColor,
+    SubnetMask,
+    Direction,
+]
 
 CastedTypes = Union[
     Global[bool],

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

@@ -4,8 +4,7 @@
 class NTPTemplateConverter:
     supported_template_types = ("cisco_ntp", "ntp")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> NTPParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> NTPParcel:
         """
         Creates an Logging object based on the provided template values.
 

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

@@ -1,3 +1,4 @@
+from copy import deepcopy
 from typing import Dict, List
 
 from catalystwan.api.configuration_groups.parcel import Global, as_default, as_global
@@ -7,8 +8,7 @@
 class OMPTemplateConverter:
     supported_template_types = ("cisco_omp", "omp-vedge", "omp-vsmart")
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> OMPParcel:
+    def create_parcel(self, name: str, description: str, template_values: dict) -> OMPParcel:
         """
         Creates an OMPParcel object based on the provided template values.
 
@@ -20,16 +20,15 @@ def create_parcel(name: str, description: str, template_values: dict) -> OMPParc
         Returns:
             OMPParcel: An OMPParcel object with the provided template values.
         """
-
-        def create_advertise_dict(advertise_list: List) -> Dict:
-            return {definition["protocol"].value: Global[bool](value=True) for definition in advertise_list}
-
+        values = deepcopy(template_values)
         parcel_values = {
             "parcel_name": name,
             "parcel_description": description,
-            "ecmp_limit": as_global(float(template_values.get("ecmp_limit", as_default(4)).value)),
-            "advertise_ipv4": create_advertise_dict(template_values.get("advertise", [])),
-            "advertise_ipv6": create_advertise_dict(template_values.get("ipv6_advertise", [])),
+            "ecmp_limit": as_global(float(values.get("ecmp_limit", as_default(4)).value)),
+            "advertise_ipv4": self.create_advertise_dict(values.get("advertise", [])),
+            "advertise_ipv6": self.create_advertise_dict(values.get("ipv6_advertise", [])),
         }
-
         return OMPParcel(**parcel_values)
+
+    def create_advertise_dict(self, advertise_list: List) -> Dict:
+        return {definition["protocol"].value: Global[bool](value=True) for definition in advertise_list}

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

@@ -17,8 +17,9 @@ class SNMPTemplateConverter:
 
     supported_template_types = ("cisco_snmp",)
 
-    @staticmethod
-    def create_parcel(name: str, description: str, template_values: dict) -> SNMPParcel:
+    default_view_oid_id = "{{{{l_snmpView_1_snmpOid_{}_id}}}}"
+
+    def create_parcel(self, name: str, description: str, template_values: dict) -> SNMPParcel:
         """
         Creates a SecurityParcel object based on the provided template values.
 
@@ -31,20 +32,24 @@ def create_parcel(name: str, description: str, template_values: dict) -> SNMPPar
             SecurityParcel: A SecurityParcel object with the provided template values.
         """
         values = deepcopy(template_values)
+        self.configure_community(values)
+        self.configure_target(values)
+        parcel_values = {"parcel_name": name, "parcel_description": description, **values}
+        return SNMPParcel(**parcel_values)  # type: ignore
+
+    def configure_community(self, values: dict):
         for community_item in values.get("community", []):
             if authorization := community_item.get("authorization"):
                 community_item["authorization"] = as_global(authorization.value, Authorization)
 
+    def configure_target(self, values: dict) -> None:
         values["target"] = values.pop("trap", {}).get("target", [])
-        default_view_oid_id = "{{{{l_snmpView_1_snmpOid_{}_id}}}}"
+
         for view in values.get("view", []):
             for i, oid in enumerate(view.get("oid", [])):
-                id_ = oid.get("id", as_variable(default_view_oid_id.format(i + 1)))
+                id_ = oid.get("id", as_variable(self.default_view_oid_id.format(i + 1)))
                 if isinstance(id_, Variable):
                     logger.info(
-                        f"OID ID is not set, using device specific variable {default_view_oid_id.format(i + 1)}"
+                        f"OID ID is not set, using device specific variable {self.default_view_oid_id.format(i + 1)}"
                     )
                 oid["id"] = id_
-
-        parcel_values = {"parcel_name": name, "parcel_description": description, **values}
-        return SNMPParcel(**parcel_values)  # type: ignore