Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Refactor code for template definition normalization
Browse files Browse the repository at this point in the history
This commit refactors the code responsible for normalizing template definitions by implementing improved type annotations, better function organization, and enhanced error handling. Key changes include:

1. Refactoring the to_snake_case function to use the str.replace method for kebab-case to snake_case conversion.
2. Improving the cast_value_to_global function to handle different types of input values, including lists and IP addresses, and utilizing IPv4Address and IPv6Address from the ipaddress module.
3. Enhancing the transform_dict function to handle nested dictionaries and lists, casting leaf values to global types recursively.
4. Updating the template_definition_normalization function to utilize the refactored transform_dict function for key transformation and value normalization.

These changes improve code readability, maintainability, and robustness, providing a more efficient and reliable solution for template definition normalization.
  • Loading branch information
jpkrajewski committed Feb 25, 2024
1 parent 146508f commit 88a53e1
Show file tree
Hide file tree
Showing 16 changed files with 307 additions and 237 deletions.
124 changes: 3 additions & 121 deletions catalystwan/models/configuration/config_migration.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,9 @@
import logging
from datetime import datetime
from typing import List, Literal, Union
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field
from typing_extensions import Annotated

from catalystwan.api.template_api import DeviceTemplateInformation, FeatureTemplateInformation
from catalystwan.endpoints.configuration_feature_profile import ConfigurationFeatureProfile
from catalystwan.endpoints.configuration_group import ConfigGroup
from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel
from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel
from catalystwan.models.policy import (
Expand All @@ -19,7 +13,6 @@
LocalizedPolicy,
SecurityPolicy,
)
from catalystwan.session import ManagerSession

AnyParcel = Annotated[
Union[
Expand Down Expand Up @@ -57,7 +50,9 @@ class UX1Templates(BaseModel):
class ConfigGroupPreset(BaseModel):
config_group_name: str = Field(serialization_alias="name", validation_alias="name")
solution: Literal["sdwan"] = "sdwan"
profile_parcels: List[AnyParcel] = Field(serialization_alias="profileParcels", validation_alias="profileParcels")
profile_parcels: List[AnyParcel] = Field(
default=[], serialization_alias="profileParcels", validation_alias="profileParcels"
)


class UX1Config(BaseModel):
Expand All @@ -74,116 +69,3 @@ class UX2Config(BaseModel):
config_group_presets: List[ConfigGroupPreset] = Field(
default=[], serialization_alias="configGroupPresets", validation_alias="configGroupPresets"
)


class ConfigGroupCreator:
"""
Creates a configuration group and attach feature profiles for migrating UX1 templates to UX2.
"""

def __init__(self, session: ManagerSession, config: UX2Config, logger: logging.Logger):
"""
Args:
session (ManagerSession): A valid Manager API session.
config (UX2Config): The UX2 configuration to migrate.
logger (logging.Logger): A logger for logging messages.
"""
self.session = session
self.config = config
self.logger = logger
self.profile_ids: List[UUID] = []

def create(self) -> ConfigGroup:
"""
Creates a configuration group and attach feature profiles for migrating UX1 templates to UX2.
Returns:
ConfigGroup: The created configuration group.
"""
self.created_at = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
self._create_sdwan_system_feature_profile()
self._create_sdwan_policy_objects_feature_profile()
config_group_id = self._create_configuration_group()
return self.session.api.config_group.get(config_group_id) # type: ignore[return-value]

def _create_sdwan_system_feature_profile(self):
"""
Creates a SDWAN System Feature Profile for migrating UX1 Templates to UX2.
Args:
session (ManagerSession): A valid Manager API session.
name (str): The name of the SDWAN System Feature Profile.
Returns:
UUID: The ID of the created SDWAN System Feature Profile.
Raises:
ManagerHTTPError: If the SDWAN System Feature Profile cannot be created.
"""
system_name = f"MIGRATION_SDWAN_SYSTEM_FEATURE_PROFILE_{self.created_at}"
profile_system = FeatureProfileCreationPayload(
name=system_name, description="Profile for migrating UX1 Templates to UX2"
)
system_id = self.session.endpoints.configuration_feature_profile.create_sdwan_system_feature_profile(
profile_system
).id
self.logger.info(f"Created SDWAN System Feature Profile {system_name} with ID: {system_id}")
self.profile_ids.append(system_id)

def _create_sdwan_policy_objects_feature_profile(self):
"""
Creates a SDWAN Policy Objects Feature Profile for migrating UX1 Policies to UX2.
Args:
session (ManagerSession): A valid Manager API session.
name (str): The name of the SDWAN Policy Objects Feature Profile.
Returns:
UUID: The ID of the created SDWAN Policy Objects Feature Profile.
Raises:
ManagerHTTPError: If the SDWAN Policy Objects Feature Profile cannot be created.
"""
policy_objects_name = f"MIGRATION_SDWAN_POLICY_OBJECTS_FEATURE_PROFILE_{self.created_at}"
# TODO: Find a way to create a policy object profile
# for now there is no API or UI for creating a policy object profile
profile_policy_objects = FeatureProfileCreationPayload( # noqa: F841
name=policy_objects_name, description="Profile for migrating UX1 Policies to UX2"
)

# Using default profile name for SDWAN Policy Objects Feature Profile
policy_object_id = (
ConfigurationFeatureProfile(self.session)
.get_sdwan_feature_profiles()
.filter(profile_name="Default_Policy_Object_Profile")
.single_or_default()
).profile_id
self.logger.info(
f"Created SDWAN Policy Object Feature Profile {policy_objects_name} with ID: {policy_object_id}"
)
self.profile_ids.append(policy_object_id)

def _create_configuration_group(self):
"""
Creates a configuration group and attach feature profiles for migrating UX1 templates to UX2.
Args:
session (ManagerSession): A valid Manager API session.
name (str): The name of the configuration group.
profile_ids (List[UUID]): The IDs of the feature profiles to include in the configuration group.
Returns:
UUID: The ID of the created configuration group.
Raises:
ManagerHTTPError: If the configuration cannot be pushed.
"""
config_group_name = f"SDWAN_CONFIG_GROUP_{self.created_at}"
config_group_id = self.session.api.config_group.create(
name=config_group_name,
description="SDWAN Config Group created for migrating UX1 Templates to UX2",
solution="sdwan",
profile_ids=self.profile_ids,
).id
self.logger.info(f"Created SDWAN Configuration Group {config_group_name} with ID: {config_group_id}")
return config_group_id

This file was deleted.

Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
from typing import List, Mapping, Union

from .aaa import AAA
from .bfd import BFD
from .aaa import AAAParcel
from .bfd import BFDParcel

SYSTEM_PAYLOAD_ENDPOINT_MAPPING: Mapping[type, str] = {
AAA: "aaa",
BFD: "bfd",
AAAParcel: "aaa",
BFDParcel: "bfd",
}

AnySystemParcel = Union[AAA, BFD]
AnySystemParcel = Union[AAAParcel, BFDParcel]

__all__ = ["AAA", "BFD", "AnySystemParcel", "SYSTEM_PAYLOAD_ENDPOINT_MAPPING"]
__all__ = ["AAAParcel", "BFDParcel", "AnySystemParcel", "SYSTEM_PAYLOAD_ENDPOINT_MAPPING"]


def __dir__() -> "List[str]":
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from pydantic import AliasPath, BaseModel, ConfigDict, Field

from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase, as_default, as_global
from catalystwan.models.configuration.feature_profile.converters.recast import (
from catalystwan.utils.config_migration.converters.recast import (
DefaultGlobalBool,
DefaultGlobalIPAddress,
DefaultGlobalList,
Expand Down Expand Up @@ -268,7 +268,7 @@ class AuthorizationRuleItem(BaseModel):
)


class AAA(_ParcelBase):
class AAAParcel(_ParcelBase):
type_: Literal["aaa"] = Field(default="aaa", exclude=True)
authentication_group: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field(
default=as_default(False),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,7 @@

from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global
from catalystwan.models.common import TLOCColor
from catalystwan.models.configuration.feature_profile.converters.recast import (
DefaultGlobalBool,
DefaultGlobalColorLiteral,
)
from catalystwan.utils.config_migration.converters.recast import DefaultGlobalBool, DefaultGlobalColorLiteral

DEFAULT_BFD_COLOR_MULTIPLIER = as_global(7)
DEFAULT_BFD_DSCP = as_global(48)
Expand All @@ -29,7 +26,7 @@ class Color(BaseModel):
model_config = ConfigDict(populate_by_name=True)


class BFD(_ParcelBase):
class BFDParcel(_ParcelBase):
type_: Literal["bfd"] = Field(default="bfd", exclude=True)
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
# from catalystwan.api.configuration_groups.parcel import _ParcelBase, Global, as_global
# from catalystwan.models.configuration.common import AuthType, Priority, Version
# from pydantic import AliasPath, BaseModel, ConfigDict, Field
# from catalystwan.models.configuration.feature_profile.converters.recast import (
# from catalystwan.utils.config_migration.converters.recast import (
# DefaultGlobalBool,
# DefaultGlobalStr,
# )
Expand Down
83 changes: 83 additions & 0 deletions catalystwan/tests/config_migration/test_normalizer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
import unittest
from ipaddress import IPv4Address, IPv6Address
from typing import List

from catalystwan.api.configuration_groups.parcel import Global
from catalystwan.utils.config_migration.converters.feature_template import template_definition_normalization


class TestNormalizer(unittest.TestCase):
def setUp(self):
self.template_values = {
"key-one": "Simple string !@#$%^&*()-=[';/.,`~]",
"keyone": "Simplestring!@#$%^&*()-=[';/.,`~]",
"bool-value-as-string": "true",
"boolvalueasstring": "false",
"simple-int": 1,
"simpleint": 333333331231,
"simple-ints-in-list": [1, 2, 4, 5, 6, 7, 8, 9],
"simple-int-in-list": [1],
"simplestringsinlist": ["1232132", "!@#$%^&*()-=[';/.,`~]", ""],
"objects-in-list": [
{"color": "lte", "hello-interval": 300000, "pmtu-discovery": "false"},
{"color": "mpls", "pmtu-discovery": "false"},
{"color": "biz-internet"},
{"color": "public-internet"},
],
"nested-objects": [{"next-hop": [{"distance": 1}]}],
"ipv4-address": "10.0.0.2",
"ipv6addr": "2000:0:2:3::",
}
self.expected_result = {
"key_one": Global[str](value="Simple string !@#$%^&*()-=[';/.,`~]"),
"keyone": Global[str](value="Simplestring!@#$%^&*()-=[';/.,`~]"),
"bool_value_as_string": Global[bool](value=True),
"boolvalueasstring": Global[bool](value=False),
"simple_int": Global[int](value=1),
"simpleint": Global[int](value=333333331231),
"simple_ints_in_list": Global[List[int]](value=[1, 2, 4, 5, 6, 7, 8, 9]),
"simple_int_in_list": Global[List[int]](value=[1]),
"simplestringsinlist": Global[List[str]](value=["1232132", "!@#$%^&*()-=[';/.,`~]", ""]),
"objects_in_list": [
{
"color": Global[str](value="lte"),
"hello_interval": Global[int](value=300000),
"pmtu_discovery": Global[bool](value=False),
},
{"color": Global[str](value="mpls"), "pmtu_discovery": Global[bool](value=False)},
{"color": Global[str](value="biz-internet")},
{"color": Global[str](value="public-internet")},
],
"nested_objects": [{"next_hop": [{"distance": Global[int](value=1)}]}],
"ipv4_address": Global[IPv4Address](value=IPv4Address("10.0.0.2")),
"ipv6addr": Global[IPv6Address](value=IPv6Address("2000:0:2:3::")),
}

def test_normalizer(self):
# Arrange
expected_result = self.expected_result
# Act
returned_result = template_definition_normalization(self.template_values)
# Assert
assert expected_result == returned_result

def test_super_nested(self):
# Arrange
super_nested_input = {
"super_nested": {"level1": {"level2": {"level3": {"key_one": "value_one", "key_two": "value_two"}}}}
}
expected_result = {
"super_nested": {
"level1": {
"level2": {
"level3": {"key_one": Global[str](value="value_one"), "key_two": Global[str](value="value_two")}
}
}
}
}

# Act
returned_result = template_definition_normalization(super_nested_input)

# Assert
assert expected_result == returned_result
Loading

0 comments on commit 88a53e1

Please sign in to comment.