policy-list conversion with transform headers (#525)

* add policy-list item transform with new header support

* fix description

* Fix find_template_values when vipValue not present (#520)

* fix: nested discriminated unions are not allowed in OpenAPI spec

* bump dev version

---------

Co-authored-by: PrzeG <[email protected]>

catalystwan/api/feature_profile_api.py

@@ -11,6 +11,7 @@
 from catalystwan.endpoints.configuration.feature_profile.sdwan.service import ServiceFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile
 from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.url import URLParcel
 from catalystwan.typed_list import DataSequence
 
 if TYPE_CHECKING:
@@ -694,13 +695,9 @@ def get(self, profile_id: UUID, parcel_type: Type[StandardCommunityParcel]) -> D
     def get(self, profile_id: UUID, parcel_type: Type[TlocParcel]) -> DataSequence[Parcel[Any]]:
         ...
 
-    # @overload
-    # def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel]) -> DataSequence[Parcel[Any]]:
-    #     ...
-
-    # @overload
-    # def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel]) -> DataSequence[Parcel[Any]]:
-    #     ...
+    @overload
+    def get(self, profile_id: UUID, parcel_type: Type[URLParcel]) -> DataSequence[Parcel[Any]]:
+        ...
 
     # get by id
 
@@ -820,13 +817,9 @@ def get(
     def get(self, profile_id: UUID, parcel_type: Type[TlocParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
         ...
 
-    # @overload
-    # def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
-    #     ...
-
-    # @overload
-    # def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
-    #     ...
+    @overload
+    def get(self, profile_id: UUID, parcel_type: Type[URLParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
+        ...
 
     def get(
         self,
@@ -954,13 +947,9 @@ def delete(self, profile_id: UUID, parcel_type: Type[StandardCommunityParcel], l
     def delete(self, profile_id: UUID, parcel_type: Type[TlocParcel], list_object_id: UUID) -> None:
         ...
 
-    # @overload
-    # def delete(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], list_object_id: UUID) -> None:
-    #     ...
-
-    # @overload
-    # def delete(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], list_object_id: UUID) -> None:
-    #     ...
+    @overload
+    def delete(self, profile_id: UUID, parcel_type: Type[URLParcel], list_object_id: UUID) -> None:
+        ...
 
     def delete(self, profile_id: UUID, parcel_type: Type[AnyPolicyObjectParcel], list_object_id: UUID) -> None:
         """

catalystwan/models/configuration/config_migration.py

@@ -63,8 +63,11 @@ class UX1Config(BaseModel):
 
 
 class TransformHeader(BaseModel):
-    type: str
-    id: UUID
+    type: str = Field(
+        description="Needed to push item to specific endpoint."
+        "Type discriminator is not present in many UX2 item payloads"
+    )
+    origin: UUID = Field(description="Original UUID of converted item")
     subelements: List[UUID] = []
 
 

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

@@ -31,20 +31,12 @@
 from .security.local_domain import LocalDomainListEntry, LocalDomainParcel
 from .security.protocol_list import ProtocolListEntry, ProtocolListParcel
 from .security.security_port import SecurityPortListEntry, SecurityPortParcel
-from .security.url import BaseURLListEntry, URLAllowParcel, URLBlockParcel
+from .security.url import BaseURLListEntry, URLAllowParcel, URLBlockParcel, URLParcel
 from .security.zone import SecurityZoneListEntry, SecurityZoneListParcel
 
-AnyURLParcel = Annotated[
-    Union[
-        URLAllowParcel,
-        URLBlockParcel,
-    ],
-    Field(discriminator="parcel_type"),
-]
-
 AnyPolicyObjectParcel = Annotated[
     Union[
-        AnyURLParcel,
+        URLParcel,
         ApplicationListParcel,
         AppProbeParcel,
         ColorParcel,
@@ -127,6 +119,7 @@
     "StandardCommunityParcel",
     "TlocEntry",
     "TlocParcel",
+    "URLParcel",
     "URLAllowParcel",
     "URLBlockParcel",
 )

catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/url.py

@@ -12,22 +12,23 @@ class BaseURLListEntry(BaseModel):
     pattern: Global[str]
 
 
-class BaseURLParcel(_ParcelBase):
+class URLParcel(_ParcelBase):
     type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True)
+    parcel_type: Literal["urlallowed", "urlblocked"]
     entries: List[BaseURLListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
     def add_url(self, pattern: str):
         self.entries.append(BaseURLListEntry(pattern=as_global(pattern)))
 
 
-class URLAllowParcel(BaseURLParcel):
+class URLAllowParcel(URLParcel):
     type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True)
     parcel_type: Literal["urlallowed"] = Field(
         default="urlallowed", validation_alias="type", serialization_alias="type"
     )
 
 
-class URLBlockParcel(BaseURLParcel):
+class URLBlockParcel(URLParcel):
     type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True)
     parcel_type: Literal["urlblocked"] = Field(
         default="urlblocked", validation_alias="type", serialization_alias="type"

catalystwan/utils/feature_template/find_template_values.py

@@ -37,14 +37,18 @@ def find_template_values(
             return templated_values
 
         value = template_definition[target_key]
-        template_value = template_definition[target_key_for_template_value]
+        template_value = template_definition.get(target_key_for_template_value)
 
         field_key = path[-1]
         # TODO: Handle nested DeviceVariable
         if value == "variableName":
             if device_specific_variables is not None:
                 device_specific_variables[field_key] = DeviceVariable(name=template_definition["vipVariableName"])
-        elif template_definition["vipType"] == "variable":
+            return template_definition
+        if template_value is None:
+            return template_definition
+
+        if template_definition["vipType"] == "variable":
             if device_specific_variables is not None and template_value:
                 device_specific_variables[field_key] = DeviceVariable(name=template_value)
         elif template_definition["vipObjectType"] == "list":

catalystwan/workflows/config_migration.py

@@ -15,6 +15,7 @@
 from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload
 from catalystwan.session import ManagerSession
 from catalystwan.utils.config_migration.converters.feature_template import create_parcel_from_template
+from catalystwan.utils.config_migration.converters.policy.policy_lists import convert as convert_policy_list
 from catalystwan.utils.config_migration.creators.config_group import ConfigGroupCreator
 from catalystwan.utils.config_migration.device_templates import flatten_general_templates
 
@@ -99,7 +100,7 @@ def transform(ux1: UX1Config) -> UX2Config:
         transformed_fp_system = TransformedFeatureProfile(
             header=TransformHeader(
                 type="system",
-                id=fp_system_uuid,
+                origin=fp_system_uuid,
             ),
             feature_profile=FeatureProfileCreationPayload(
                 name=f"{dt.template_name}_system",
@@ -110,7 +111,7 @@ def transform(ux1: UX1Config) -> UX2Config:
         transformed_fp_transport = TransformedFeatureProfile(
             header=TransformHeader(
                 type="transport",
-                id=fp_transport_uuid,
+                origin=fp_transport_uuid,
             ),
             feature_profile=FeatureProfileCreationPayload(
                 name=f"{dt.template_name}_transport",
@@ -121,7 +122,7 @@ def transform(ux1: UX1Config) -> UX2Config:
         transformed_fp_other = TransformedFeatureProfile(
             header=TransformHeader(
                 type="other",
-                id=fp_other_uuid,
+                origin=fp_other_uuid,
             ),
             feature_profile=FeatureProfileCreationPayload(
                 name=f"{dt.template_name}_other",
@@ -141,7 +142,7 @@ def transform(ux1: UX1Config) -> UX2Config:
         transformed_cg = TransformedConfigGroup(
             header=TransformHeader(
                 type="config_group",
-                id=uuid4(),
+                origin=uuid4(),
                 subelements=[fp_system_uuid, fp_transport_uuid, fp_other_uuid],
             ),
             config_group=ConfigGroupCreationPayload(
@@ -163,13 +164,21 @@ def transform(ux1: UX1Config) -> UX2Config:
             transformed_parcel = TransformedParcel(
                 header=TransformHeader(
                     type=parcel._get_parcel_type(),
-                    id=UUID(ft.id),
+                    origin=UUID(ft.id),
                 ),
                 parcel=parcel,
             )
             # Add to UX2. We can indentify the parcels as subelements of the feature profiles by the UUIDs
             ux2.profile_parcels.append(transformed_parcel)
 
+    # Policy Lists
+    for policy_list in ux1.policies.policy_lists:
+        policy_parcel = convert_policy_list(policy_list)
+        if policy_parcel is not None:
+            header = TransformHeader(type=policy_parcel._get_parcel_type(), origin=policy_list.list_id)
+            ux2.profile_parcels.append(TransformedParcel(header=header, parcel=policy_parcel))
+        else:
+            logger.warning(f"{policy_list.type} {policy_list.list_id} {policy_list.name} was not converted")
     return ux2
 
 

pyproject.toml

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