Add suffix to config groups and feature profiles (#690)

* Add suffix to config groups and feature templates

* change to default_factory

* Fix Issue646 Not content in response

* restore tenant model validate during login

* bump dev version

* add topology profile type literal

* fix models

---------

Co-authored-by: sbasan <sbasan@ams3-vpn-dhcp1956.cisco.com>
Co-authored-by: Szymon Basan <szymon.basan@gmail.com>
Co-authored-by: Szymon Basan <sbasan@cisco.com>

catalystwan/models/configuration/config_migration.py

@@ -1,7 +1,7 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
-from uuid import UUID
+from uuid import UUID, uuid4
 
 from packaging.version import Version
 from pydantic import BaseModel, ConfigDict, Field, model_validator
@@ -210,13 +210,22 @@ def insert_parcel_type_from_headers(cls, values: Dict[str, Any]):
 
 
 class ConfigTransformResult(BaseModel):
+    # https://docs.pydantic.dev/2.0/usage/models/#fields-with-dynamic-default-values
+    uuid: UUID = Field(default_factory=uuid4)
     ux2_config: UX2Config = Field(
         default_factory=lambda: UX2Config(), serialization_alias="ux2Config", validation_alias="ux2Config"
     )
     failed_items: List[FailedConversionItem] = Field(
         default_factory=list, serialization_alias="failedConversionItems", validation_alias="failedConversionItems"
     )
 
+    def add_suffix_to_names(self):
+        suffix = str(self.uuid)[:7]
+        for config_group in self.ux2_config.config_groups:
+            config_group.config_group.name = f"{config_group.config_group.name}_{suffix}"
+        for feature_profile in self.ux2_config.feature_profiles:
+            feature_profile.feature_profile.name = f"{feature_profile.feature_profile.name}_{suffix}"
+
     def add_failed_conversion_parcel(
         self,
         exception_message: str,

catalystwan/models/configuration/feature_profile/common.py

@@ -36,12 +36,13 @@
     "cli",
     "service",
     "application-priority",
-    "policy-object",
+    "policy-object",  # automatically created default policy object feature profile
     "embedded-security",
     "other",
     "uc-voice",
     "global",  # automatically created global cellulargateway feature profile
     "sig-security",
+    "topology",
 ]
 
 SchemaType = Literal[

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefered_group_color.py

@@ -16,7 +16,9 @@
 
 class Preference(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
-    color_preference: Global[list] = Field(serialization_alias="colorPreference", validation_alias="colorPreference")
+    color_preference: Global[List[TLOCColor]] = Field(
+        serialization_alias="colorPreference", validation_alias="colorPreference"
+    )
     path_preference: Global[PathPreference] = Field(
         serialization_alias="pathPreference", validation_alias="pathPreference"
     )
@@ -49,7 +51,7 @@ def add_primary(self, color_preference: List[TLOCColor], path_preference: PathPr
         self.entries.append(
             PreferredColorGroupEntry(
                 primary_preference=Preference(
-                    color_preference=as_global(color_preference),
+                    color_preference=Global[List[TLOCColor]](value=color_preference),
                     path_preference=as_global(path_preference, PathPreference),
                 ),
                 secondary_preference=None,
@@ -60,11 +62,13 @@ def add_primary(self, color_preference: List[TLOCColor], path_preference: PathPr
     def add_secondary(self, color_preference: List[TLOCColor], path_preference: PathPreference):
         preferred_color = self.entries[0]
         preferred_color.secondary_preference = Preference(
-            color_preference=as_global(color_preference), path_preference=as_global(path_preference, PathPreference)
+            color_preference=Global[List[TLOCColor]](value=color_preference),
+            path_preference=as_global(path_preference, PathPreference),
         )
 
     def add_tertiary(self, color_preference: List[TLOCColor], path_preference: PathPreference):
         preferred_color = self.entries[0]
         preferred_color.tertiary_preference = Preference(
-            color_preference=as_global(color_preference), path_preference=as_global(path_preference, PathPreference)
+            color_preference=Global[List[TLOCColor]](value=color_preference),
+            path_preference=as_global(path_preference, PathPreference),
         )

catalystwan/models/policy/definition/security_group.py

@@ -66,10 +66,8 @@ class SecurityGroup(PolicyDefinitionBase):
 
     @model_validator(mode="after")
     def validate_by_sequence_ip_type(self):
-        if (
-            self.sequence_ip_type == SequenceIPType.IPV4 and isinstance(self.definition, SecurityGroupIPv6Definition)
-        ) or (
-            self.sequence_ip_type == SequenceIPType.IPV6 and isinstance(self.definition, SecurityGroupIPv4Definition)
+        if (self.sequence_ip_type == "ipv6" and not isinstance(self.definition, SecurityGroupIPv6Definition)) or (
+            self.sequence_ip_type == "ipv4" and not isinstance(self.definition, SecurityGroupIPv4Definition)
         ):
             raise ValueError(f"Incompatible definition for {self.sequence_ip_type} sequence")
         return self

catalystwan/response.py

@@ -120,8 +120,9 @@ def parse_cookies_to_dict(cookies: str) -> Dict[str, str]:
 
 
 class JsonPayload:
-    def __init__(self, json: Any = None):
+    def __init__(self, json: Any = None, empty: bool = False):
         self.json = json
+        self.empty = empty
         self.data = None
         self.error = None
         self.headers = None
@@ -141,7 +142,7 @@ def __init__(self, response: Response):
         try:
             self.payload = JsonPayload(response.json())
         except JSONDecodeError:
-            self.payload = JsonPayload()
+            self.payload = JsonPayload(empty=True)
 
     def _detect_expired_jsessionid(self) -> bool:
         """Determines if server sent expired JSESSIONID"""
@@ -184,6 +185,9 @@ def dataseq(self, cls: Type[T], sourcekey: Optional[str] = "data", validate: boo
             DataSequence[T] of given type T which is subclassing from Dataclass/BaseModel,
             in case JSON payload was containing a single Object - sequence with one element is returned
         """
+        if self.payload.empty:
+            return DataSequence(cls, [])
+
         if sourcekey is None:
             data = self.payload.json
         else:

catalystwan/session.py

@@ -421,7 +421,7 @@ def get_tenant_id(self) -> str:
         Returns:
             Tenant UUID.
         """
-        tenants = self.get("dataservice/tenant").dataseq(Tenant, validate=False)
+        tenants = self.get("dataservice/tenant").dataseq(Tenant)
         tenant = tenants.filter(subdomain=self.subdomain).single_or_default()
 
         if not tenant or not tenant.tenant_id:

catalystwan/tests/config_migration/test_transform.py

@@ -153,7 +153,7 @@ def test_when_many_cisco_vpn_feature_templates_expect_assign_to_correct_feature_
         )
     )
     # Act
-    ux2_config = transform(ux1_config).ux2_config
+    ux2_config = transform(ux1_config, add_suffix=False).ux2_config
     # There must be feature profiles named DeviceTemplate_service and DeviceTemplate_transport_and_management
     service_profile = None
     transport_and_management_profile = None
@@ -246,7 +246,7 @@ def test_when_one_feature_template_with_invalid_payload_expect_one_failed_item_i
         )
     )
     # Act
-    transform_result = transform(ux1_config)
+    transform_result = transform(ux1_config, add_suffix=False)
     # Assert
     assert len(transform_result.failed_items) == 1
     assert transform_result.failed_items[0].feature_template == malformed_logging

catalystwan/tests/test_response.py

@@ -7,6 +7,7 @@
 from attr import define, field  # type: ignore
 from parameterized import parameterized  # type: ignore
 from pydantic import BaseModel, Field, ValidationError
+from requests.exceptions import JSONDecodeError
 
 from catalystwan.dataclasses import DataclassBase
 from catalystwan.response import ManagerErrorInfo, ManagerResponse
@@ -173,3 +174,12 @@ def test_dataseq_optional_validate(self):
             assert data.important == VALIDATE_DATASEQ_TEST_DATA[i]["important"]
         with self.assertRaises(ValidationError):
             vmng_response.dataseq(DataForValidateTest, sourcekey=None, validate=True)
+
+    def test_dataseq_with_misisng_data(self):
+        self.response_mock.json.side_effect = JSONDecodeError("test", "test", 1)
+        vmng_response = ManagerResponse(self.response_mock)
+        # Act
+        dataseq = vmng_response.dataseq(DataForValidateTest, sourcekey="data", validate=True)
+        # Assert
+        assert isinstance(dataseq, DataSequence)
+        assert len(dataseq) == 0

catalystwan/workflows/config_migration.py

@@ -191,7 +191,7 @@ def get_version_info(session: ManagerSession) -> VersionInfo:
     return VersionInfo(platform=session._platform_version, sdk=PACKAGE_VERSION)
 
 
-def transform(ux1: UX1Config) -> ConfigTransformResult:
+def transform(ux1: UX1Config, add_suffix: bool = True) -> ConfigTransformResult:
     transform_result = ConfigTransformResult()
     ux2 = UX2Config(version=ux1.version)
     subtemplates_mapping = defaultdict(set)
@@ -353,7 +353,10 @@ def transform(ux1: UX1Config) -> ConfigTransformResult:
             )
 
     ux2 = merge_parcels(ux2)
-    transform_result.ux2_config = UX2Config.model_validate(ux2)
+    transform_result.ux2_config = ux2
+    if add_suffix:
+        transform_result.add_suffix_to_names()
+    transform_result.ux2_config = UX2Config.model_validate(transform_result.ux2_config)
     return transform_result
 
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "catalystwan"
-version = "0.33.6dev1"
+version = "0.33.7dev0"
 description = "Cisco Catalyst WAN SDK for Python"
 authors = ["kagorski <kagorski@cisco.com>"]
 readme = "README.md"