Add tests for normalization and Literal casting

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

@@ -2,15 +2,24 @@
 
 from .aaa import AAAParcel
 from .bfd import BFDParcel
+from .literals import SYSTEM_LITERALS
+from .logging_parcel import LoggingParcel
 
 SYSTEM_PAYLOAD_ENDPOINT_MAPPING: Mapping[type, str] = {
     AAAParcel: "aaa",
     BFDParcel: "bfd",
 }
 
-AnySystemParcel = Union[AAAParcel, BFDParcel]
+AnySystemParcel = Union[AAAParcel, BFDParcel, LoggingParcel]
 
-__all__ = ["AAAParcel", "BFDParcel", "AnySystemParcel", "SYSTEM_PAYLOAD_ENDPOINT_MAPPING"]
+__all__ = [
+    "AAAParcel",
+    "BFDParcel",
+    "LoggingParcel",
+    "AnySystemParcel",
+    "SYSTEM_LITERALS",
+    "SYSTEM_PAYLOAD_ENDPOINT_MAPPING",
+]
 
 
 def __dir__() -> "List[str]":

catalystwan/models/configuration/feature_profile/sdwan/system/aaa.py

@@ -4,12 +4,6 @@
 from pydantic import AliasPath, BaseModel, ConfigDict, Field
 
 from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase, as_default, as_global
-from catalystwan.utils.config_migration.converters.recast import (
-    DefaultGlobalBool,
-    DefaultGlobalIPAddress,
-    DefaultGlobalList,
-    DefaultGlobalStr,
-)
 
 
 class PubkeyChainItem(BaseModel):
@@ -67,9 +61,7 @@ def add_pubkey_chain_item(self, key: str) -> PubkeyChainItem:
 class RadiusServerItem(BaseModel):
     model_config = ConfigDict(extra="forbid", populate_by_name=True)
 
-    address: Union[DefaultGlobalIPAddress, Global[IPv4Address], Global[IPv6Address]] = Field(
-        description="Set IP address of Radius server"
-    )
+    address: Union[Global[IPv4Address], Global[IPv6Address]] = Field(description="Set IP address of Radius server")
     auth_port: Union[Global[int], Default[int], Variable, None] = Field(
         default=as_default(1812),
         validation_alias="authPort",
@@ -101,7 +93,7 @@ class RadiusServerItem(BaseModel):
         description="Set the TACACS server shared type 7 encrypted key",
     )
     # Literal["6", "7"]
-    key_enum: Union[DefaultGlobalStr, Global[str], Default[None], None] = Field(
+    key_enum: Union[Global[str], Default[None], None] = Field(
         default=None,
         validation_alias="keyEnum",
         serialization_alias="keyEnum",
@@ -164,9 +156,7 @@ def generate_radius_server(
 class TacacsServerItem(BaseModel):
     model_config = ConfigDict(extra="forbid", populate_by_name=True)
 
-    address: Union[DefaultGlobalIPAddress, Global[IPv4Address], Global[IPv6Address]] = Field(
-        description="Set IP address of TACACS server"
-    )
+    address: Union[Global[IPv4Address], Global[IPv6Address]] = Field(description="Set IP address of TACACS server")
     port: Union[Variable, Global[int], Default[int], None] = Field(default=None, description="TACACS Port")
     timeout: Union[Variable, Global[int], Default[int], None] = Field(
         default=None,
@@ -186,7 +176,7 @@ class TacacsServerItem(BaseModel):
         description="Set the TACACS server shared type 7 encrypted key",
     )
     # Literal["6", "7"]
-    key_enum: Union[DefaultGlobalStr, Global[str], Default[None], None] = Field(
+    key_enum: Union[Global[str], Default[None], None] = Field(
         default=None,
         validation_alias="keyEnum",
         serialization_alias="keyEnum",
@@ -241,13 +231,13 @@ class AccountingRuleItem(BaseModel):
     method: Global[str] = Field(description="Configure Accounting Method")
     # Literal['1', '15']
     level: Union[Global[str], Default[None], None] = Field(None, description="Privilege level when method is commands")
-    start_stop: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool], None] = Field(
+    start_stop: Union[Variable, Global[bool], Default[bool], None] = Field(
         default=None,
         validation_alias="startStop",
         serialization_alias="startStop",
         description="Record start and stop without waiting",
     )
-    group: Union[DefaultGlobalList, Global[List[str]]] = Field(description="Use Server-group")
+    group: Global[List[str]] = Field(description="Use Server-group")
 
 
 class AuthorizationRuleItem(BaseModel):
@@ -259,8 +249,8 @@ class AuthorizationRuleItem(BaseModel):
     method: Global[str]
     # Literal['1', '15']
     level: Global[str] = Field(description="Privilege level when method is commands")
-    group: Union[DefaultGlobalList, Global[List[str]]] = Field(description="Use Server-group")
-    if_authenticated: Union[DefaultGlobalBool, Global[bool], Default[bool], None] = Field(
+    group: Global[List[str]] = Field(description="Use Server-group")
+    if_authenticated: Union[Global[bool], Default[bool], None] = Field(
         default=None,
         validation_alias="ifAuthenticated",
         serialization_alias="ifAuthenticated",
@@ -270,18 +260,18 @@ class AuthorizationRuleItem(BaseModel):
 
 class AAAParcel(_ParcelBase):
     type_: Literal["aaa"] = Field(default="aaa", exclude=True)
-    authentication_group: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field(
+    authentication_group: Union[Variable, Global[bool], Default[bool]] = Field(
         default=as_default(False),
         validation_alias=AliasPath("data", "authenticationGroup"),
         description="Authentication configurations parameters",
     )
-    accounting_group: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field(
+    accounting_group: Union[Variable, Global[bool], Default[bool]] = Field(
         default=as_default(False),
         validation_alias=AliasPath("data", "accountingGroup"),
         description="Accounting configurations parameters",
     )
     # local, radius, tacacs
-    server_auth_order: Union[DefaultGlobalList, Global[List[str]]] = Field(
+    server_auth_order: Global[List[str]] = Field(
         validation_alias=AliasPath("data", "serverAuthOrder"),
         min_length=1,
         max_length=4,
@@ -299,12 +289,12 @@ class AAAParcel(_ParcelBase):
     accounting_rule: Optional[List[AccountingRuleItem]] = Field(
         default=None, validation_alias=AliasPath("data", "accountingRule"), description="Configure the accounting rules"
     )
-    authorization_console: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field(
+    authorization_console: Union[Variable, Global[bool], Default[bool]] = Field(
         default=as_default(False),
         validation_alias=AliasPath("data", "authorizationConsole"),
         description="For enabling console authorization",
     )
-    authorization_config_commands: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field(
+    authorization_config_commands: Union[Variable, Global[bool], Default[bool]] = Field(
         default=as_default(False),
         validation_alias=AliasPath("data", "authorizationConfigCommands"),
         description="For configuration mode commands.",

catalystwan/models/configuration/feature_profile/sdwan/system/literals.py

@@ -0,0 +1,7 @@
+from typing import Literal
+
+Priority = Literal["information", "debugging", "notice", "warn", "error", "critical", "alert", "emergency"]
+Version = Literal["TLSv1.1", "TLSv1.2"]
+AuthType = Literal["Server", "Mutual"]
+
+SYSTEM_LITERALS = [Priority, Version, AuthType]

catalystwan/models/configuration/feature_profile/sdwan/system/logging_parcel.py

@@ -0,0 +1,82 @@
+from typing import List, Optional, Union
+
+from pydantic import AliasPath, BaseModel, ConfigDict, Field
+
+from catalystwan.api.configuration_groups.parcel import Default, Global, _ParcelBase, as_global
+from catalystwan.models.configuration.feature_profile.sdwan.system.literals import AuthType, Priority, Version
+from catalystwan.utils.pydantic_validators import ConvertBoolToStringModel
+
+
+class TlsProfile(ConvertBoolToStringModel):
+    profile: str
+    version: Optional[Version] = Field(default="TLSv1.1", json_schema_extra={"data_path": ["tls-version"]})
+    auth_type: AuthType = Field(json_schema_extra={"vmanage_key": "auth-type"})
+    ciphersuite_list: Optional[List] = Field(
+        default=None, json_schema_extra={"data_path": ["ciphersuite"], "vmanage_key": "ciphersuite-list"}
+    )
+    model_config = ConfigDict(populate_by_name=True)
+
+
+class Server(BaseModel):
+    name: Global[str]
+    vpn: Optional[Global[str]] = None
+    source_interface: Optional[Global[str]] = Field(
+        default=None, serialization_alias="sourceInterface", validation_alias="sourceInterface"
+    )
+    priority: Optional[Global[Priority]] = Field(default="information")
+    enable_tls: Optional[Global[bool]] = Field(
+        default=as_global(False), serialization_alias="tlsEnable", validation_alias="tlsEnable"
+    )
+    custom_profile: Optional[Global[bool]] = Field(
+        default=as_global(False),
+        serialization_alias="tlsPropertiesCustomProfile",
+        validation_alias="tlsPropertiesCustomProfile",
+    )
+    profile_properties: Optional[Global[str]] = Field(
+        default=None, serialization_alias="tlsPropertiesProfile", validation_alias="tlsPropertiesProfile"
+    )
+    model_config = ConfigDict(populate_by_name=True)
+
+
+class Ipv6Server(BaseModel):
+    name: Global[str]
+    vpn: Optional[Global[str]] = None
+    source_interface: Optional[Global[str]] = Field(
+        default=None, serialization_alias="sourceInterface", validation_alias="sourceInterface"
+    )
+    priority: Optional[Global[Priority]] = Field(default="information")
+    enable_tls: Optional[Global[bool]] = Field(
+        default=as_global(False), serialization_alias="tlsEnable", validation_alias="tlsEnable"
+    )
+    custom_profile: Optional[Global[bool]] = Field(
+        default=as_global(False),
+        serialization_alias="tlsPropertiesCustomProfile",
+        validation_alias="tlsPropertiesCustomProfile",
+    )
+    profile_properties: Optional[Global[str]] = Field(
+        default=None, serialization_alias="tlsPropertiesProfile", validation_alias="tlsPropertiesProfile"
+    )
+    model_config = ConfigDict(populate_by_name=True)
+
+
+class File(BaseModel):
+    disk_file_size: Optional[Union[Global[int], Default[int]]] = Field(
+        default=Default[int](value=10), serialization_alias="diskFileSize", validation_alias="diskFileSize"
+    )
+    disk_file_rotate: Optional[Union[Global[int], Default[int]]] = Field(
+        default=Default[int](value=10), serialization_alias="diskFileRotate", validation_alias="diskFileRotate"
+    )
+
+
+class Disk(BaseModel):
+    disk_enable: Optional[Global[bool]] = Field(
+        default=False, serialization_alias="diskEnable", validation_alias="diskEnable"
+    )
+    file: File
+
+
+class LoggingParcel(_ParcelBase):
+    disk: Optional[Disk] = Field(default=None, validation_alias=AliasPath("data", "disk"))
+    tls_profile: Optional[List[TlsProfile]] = Field(default=[], validation_alias=AliasPath("data", "tlsProfile"))
+    server: Optional[List[Server]] = Field(default=[], validation_alias=AliasPath("data", "server"))
+    ipv6_server: Optional[List[Ipv6Server]] = Field(default=[], validation_alias=AliasPath("data", "ipv6Server"))

catalystwan/tests/config_migration/test_converter_chooser.py

@@ -0,0 +1,26 @@
+import unittest
+
+from parameterized import parameterized  # type: ignore
+
+from catalystwan.exceptions import CatalystwanException
+from catalystwan.utils.config_migration.converters.feature_template import choose_parcel_converter
+from catalystwan.utils.config_migration.converters.feature_template.aaa import AAATemplateConverter
+from catalystwan.utils.config_migration.converters.feature_template.bfd import BFDTemplateConverter
+
+
+class TestParcelConverterChooser(unittest.TestCase):
+    @parameterized.expand(
+        [("cisco_aaa", AAATemplateConverter), ("cedge_aaa", AAATemplateConverter), ("cisco_bfd", BFDTemplateConverter)]
+    )
+    def test_choose_parcel_converter_returns_correct_converter_when_supported(self, template_type, expected):
+        # Arrange, Act
+        converter = choose_parcel_converter(template_type)
+        # Assert
+        self.assertEqual(converter, expected)
+
+    def test_choose_parcel_converter_throws_exception_when_template_type_not_supported(self):
+        # Arrange
+        not_supported_type = "!@#$%^&*()"
+        # Act, Assert
+        with self.assertRaises(CatalystwanException, msg=f"Template type {not_supported_type} not supported"):
+            choose_parcel_converter(not_supported_type)

catalystwan/tests/config_migration/test_normalizer.py

@@ -1,10 +1,13 @@
 import unittest
 from ipaddress import IPv4Address, IPv6Address
-from typing import List
+from typing import List, Literal
+from unittest.mock import patch
 
 from catalystwan.api.configuration_groups.parcel import Global
 from catalystwan.utils.config_migration.converters.feature_template import template_definition_normalization
 
+TestLiteral = Literal["castable_literal"]
+
 
 class TestNormalizer(unittest.TestCase):
     def setUp(self):
@@ -53,15 +56,15 @@ def setUp(self):
             "ipv6addr": Global[IPv6Address](value=IPv6Address("2000:0:2:3::")),
         }
 
-    def test_normalizer(self):
+    def test_normalizer_handles_various_types_of_input(self):
         # Arrange
         expected_result = self.expected_result
         # Act
         returned_result = template_definition_normalization(self.template_values)
         # Assert
-        assert expected_result == returned_result
+        self.assertDictEqual(expected_result, returned_result)
 
-    def test_super_nested(self):
+    def test_normalizer_handles_super_nested_input(self):
         # Arrange
         super_nested_input = {
             "super_nested": {"level1": {"level2": {"level3": {"key_one": "value_one", "key_two": "value_two"}}}}
@@ -80,4 +83,16 @@ def test_super_nested(self):
         returned_result = template_definition_normalization(super_nested_input)
 
         # Assert
-        assert expected_result == returned_result
+        self.assertDictEqual(expected_result, returned_result)
+
+    @patch("catalystwan.models.configuration.feature_profile.sdwan.system.literals.SYSTEM_LITERALS", [TestLiteral])
+    def test_normalizer_literal_casting_when_literal_in_system_literals(self):
+        # Arrange
+        simple_input = {"in": "castable_literal"}
+        expected_result = {"in": Global[TestLiteral](value="castable_literal")}
+
+        # Act
+        returned_result = template_definition_normalization(simple_input)
+
+        # Assert
+        self.assertDictEqual(expected_result, returned_result)