Add function to transform header. Add network type to feature profile…

…. Add unittest for qos map and settings for app prio. Add converter for settings. Move logic to localized pusher

catalystwan/models/configuration/config_migration.py

@@ -175,6 +175,12 @@ class TransformHeader(BaseModel):
     status: ConvertOutputStatus = Field(default="complete")
     info: List[str] = Field(default_factory=list)
 
+    def add_localized_policy_subelement(self, subelement: UUID) -> None:
+        if self.localized_policy_subelements is None:
+            self.localized_policy_subelements = {subelement}
+        else:
+            self.localized_policy_subelements.add(subelement)
+
 
 class TransformedTopologyGroup(BaseModel):
     header: TransformHeader

catalystwan/models/configuration/feature_profile/common.py

@@ -30,19 +30,20 @@
 from catalystwan.models.configuration.common import Solution
 
 ProfileType = Literal[
-    "transport",
-    "system",
-    "cli",
-    "service",
     "application-priority",
-    "policy-object",  # automatically created default policy object feature profile
+    "cli",
+    "dns-security",
     "embedded-security",
-    "other",
-    "uc-voice",
     "global",  # automatically created global cellulargateway feature profile
+    "networks",
+    "other",
+    "policy-object",  # automatically created default policy object feature profile
+    "service",
     "sig-security",
+    "system",
     "topology",
-    "dns-security",
+    "transport",
+    "uc-voice",
 ]
 
 SchemaType = Literal[

catalystwan/models/policy/localized.py

@@ -1,4 +1,4 @@
-# Copyright 2023 Cisco Systems, Inc. and its affiliates
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 
 from typing import Any, List, Literal, Optional, Union
 from uuid import UUID
@@ -117,6 +117,9 @@ def add_device_access_policy_ipv6(self, definition_id: UUID) -> None:
     def add_route_policy(self, definition_id: UUID) -> None:
         self._add_item("vedgeRoute", definition_id)
 
+    def set_definition(self, assembly: List[LocalizedPolicyAssemblyItem], settings: LocalizedPolicySettings) -> None:
+        self.policy_definition = LocalizedPolicyDefinition(assembly=assembly, settings=settings)
+
     @model_validator(mode="before")
     @classmethod
     def try_parse_policy_definition_string(cls, values):

catalystwan/tests/config_migration/test_data/__init__.py

@@ -1,10 +1,13 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
 from typing import List
 
 from .feature_templates.dhcp import dhcp_server
 from .feature_templates.interface import interface_ethernet, interface_gre, interface_ipsec, interface_multilink
 from .feature_templates.ospfv3 import ospfv3
 from .feature_templates.vpn import vpn_management, vpn_service, vpn_transport
+from .localized_policies.localized_policy import create_localized_policy_info
+from .policy_definitions.qos_map import create_qos_map_policy
 
 __all__ = [
     "interface_ethernet",
@@ -16,6 +19,8 @@
     "ospfv3",
     "dhcp_server",
     "interface_multilink",
+    "create_qos_map_policy",
+    "create_localized_policy_info",
 ]
 
 

catalystwan/tests/config_migration/test_data/localized_policies/localized_policy.py

@@ -0,0 +1,32 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+from datetime import datetime
+from uuid import uuid4
+
+from catalystwan.models.policy.localized import LocalizedPolicyInfo, LocalizedPolicySettings
+
+
+def create_localized_policy_info(name: str) -> LocalizedPolicyInfo:
+    policy = LocalizedPolicyInfo(
+        policy_type="feature",
+        policy_id=uuid4(),
+        policy_name=name,
+        created_by="tester",
+        created_on=datetime.now(),
+        last_updated_by="tester",
+        last_updated_on=datetime.now(),
+        policy_version=None,
+    )
+    settings = LocalizedPolicySettings(
+        flow_visibility=True,
+        flow_visibility_ipv6=True,
+        app_visibility=True,
+        app_visibility_ipv6=True,
+        cloud_qos=True,
+        cloud_qos_service_side=True,
+        implicit_acl_logging=True,
+        log_frequency=10,
+        ip_visibility_cache_entries=100,
+        ip_v6_visibility_cache_entries=200,
+    )
+    policy.set_definition([], settings)
+    return policy

catalystwan/tests/config_migration/test_data/policy_definitions/qos_map.py

@@ -0,0 +1,23 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+from datetime import datetime
+from uuid import uuid4
+
+from catalystwan.models.policy.definition.qos_map import QoSMapPolicyGetResponse
+
+
+def create_qos_map_policy(name: str) -> QoSMapPolicyGetResponse:
+    policy = QoSMapPolicyGetResponse(
+        definition_id=uuid4(),
+        name=name,
+        description=f"{name} description",
+        last_updated=datetime.now(),
+        owner="tester",
+        reference_count=0,
+        references=[],
+        is_activated_by_vsmart=False,
+    )
+    policy.add_scheduler(queue=1, class_map_ref=uuid4(), scheduling="llq", drops="red-drop", bandwidth=10)
+    policy.add_scheduler(queue=2, class_map_ref=uuid4(), scheduling="wrr", drops="red-drop", bandwidth=20)
+    policy.add_scheduler(queue=3, class_map_ref=uuid4(), scheduling="llq", drops="tail-drop", bandwidth=30)
+    policy.add_scheduler(queue=4, class_map_ref=uuid4(), scheduling="wrr", drops="tail-drop", bandwidth=40)
+    return policy

catalystwan/tests/config_migration/test_transform.py

@@ -1,3 +1,4 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
 from typing import List, Optional, TypeVar
 from uuid import UUID, uuid4
@@ -8,10 +9,13 @@
     DeviceTemplateWithInfo,
     TransformedParcel,
     UX1Config,
+    UX1Policies,
     UX1Templates,
 )
 from catalystwan.models.templates import FeatureTemplateInformation
 from catalystwan.tests.config_migration.test_data import (
+    create_localized_policy_info,
+    create_qos_map_policy,
     dhcp_server,
     interface_ethernet,
     interface_gre,
@@ -514,3 +518,51 @@ def test_when_transform_expect_removed_copies():
     assert removed_vpn_service is None
     assert removed_vpn_standalone is None
     assert removed_ethernet is None
+
+
+def test_when_localized_policy_with_qos_expect_application_priority_feature_profile_with_qos_parcels():
+    """Localized Policy can have QoS Maps as a subelements. This test checks if the transformed
+    Localized Policy with QoS Map subelements produces the correct QoS parcels and if they are
+    correctly assigned to the appropriate feature profile after the transformation from UX1 to UX2."""
+
+    # Arrange
+    localized_policy = create_localized_policy_info("LocalizedPolicy1")
+    qos_map_1 = create_qos_map_policy("QoSMap1")
+    qos_map_2 = create_qos_map_policy("QoSMap2")
+    localized_policy.add_qos_map(qos_map_1.definition_id)
+    localized_policy.add_qos_map(qos_map_2.definition_id)
+    ux1_config = UX1Config(
+        policies=UX1Policies(
+            localized_policies=[localized_policy],
+            policy_definitions=[qos_map_1, qos_map_2],
+        )
+    )
+    # Act
+    ux2_config = transform(ux1_config).ux2_config
+    # Find application priority feature profile
+    application_priority_profile = next(
+        (
+            p
+            for p in ux2_config.feature_profiles
+            if p.feature_profile.name == f"FROM_{localized_policy.policy_name}"
+            and p.header.type == "application-priority"
+        ),
+        None,
+    )
+    qos_map_1_parcel = next((p for p in ux2_config.profile_parcels if p.parcel.parcel_name == qos_map_1.name), None)
+    qos_map_2_parcel = next((p for p in ux2_config.profile_parcels if p.parcel.parcel_name == qos_map_2.name), None)
+    settings = next((p for p in ux2_config.profile_parcels if p.parcel.parcel_name.endswith("_Settings")), None)
+    # Assert
+    assert application_priority_profile is not None
+    # Feature profile shoulde have 3 subelements: QoS Map 1, QoS Map 2 and
+    # Settings with uuid derived from Localized Policy
+    assert application_priority_profile.header.localized_policy_subelements == {
+        qos_map_1.definition_id,
+        qos_map_2.definition_id,
+        localized_policy.policy_id,
+    }
+    # QoS Map 1 and QoS Map 2 should be in the list of parcels
+    assert qos_map_1_parcel is not None
+    assert qos_map_2_parcel is not None
+    # Settings should be in the list of parcels
+    assert settings is not None

catalystwan/utils/config_migration/converters/policy/policy_settings.py

@@ -0,0 +1,39 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+from uuid import UUID
+
+from catalystwan.api.configuration_groups.parcel import as_optional_global
+from catalystwan.models.configuration.config_migration import ConvertResult, PolicyConvertContext
+from catalystwan.models.configuration.feature_profile.sdwan.application_priority.policy_settings import (
+    PolicySettingsParcel,
+)
+from catalystwan.models.policy.localized import LocalizedPolicyInfo
+
+
+def convert_localized_policy_settings(
+    policy: LocalizedPolicyInfo, uuid: UUID, context: PolicyConvertContext
+) -> ConvertResult[PolicySettingsParcel]:
+    if isinstance(policy.policy_definition, str):
+        return ConvertResult(
+            output=None,
+            status="unsupported",
+            info=[
+                f"Localized policy {policy.policy_name} has a string settings definition. "
+                "This is not supported by the converter."
+            ],
+        )
+    settings = policy.policy_definition.settings
+    if settings is None:
+        parcel = PolicySettingsParcel(
+            parcel_name=f"{policy.policy_name}_Settings",
+            parcel_description=policy.policy_description,
+        )
+        return ConvertResult(output=parcel, status="complete")
+    parcel = PolicySettingsParcel(
+        parcel_name=f"{policy.policy_name}_Settings",
+        parcel_description=policy.policy_description,
+        app_visibility=as_optional_global(settings.app_visibility),
+        app_visibility_ipv6=as_optional_global(settings.app_visibility_ipv6),
+        flow_visibility=as_optional_global(settings.flow_visibility),
+        flow_visibility_ipv6=as_optional_global(settings.app_visibility_ipv6),
+    )
+    return ConvertResult(output=parcel, status="partial", info=["cflowd is not supported yet"])

catalystwan/utils/config_migration/creators/localized_policy_pusher.py

@@ -1,3 +1,4 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 import logging
 from typing import Callable, Dict, List, Literal, Tuple, Union, cast
 from uuid import UUID
@@ -10,19 +11,29 @@
 from catalystwan.exceptions import ManagerErrorInfo, ManagerHTTPError
 from catalystwan.models.configuration.config_migration import (
     PushContext,
+    TransformedFeatureProfile,
     TransformedParcel,
     UX2Config,
     UX2ConfigPushResult,
 )
 from catalystwan.models.configuration.feature_profile.parcel import list_types
 from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv4acl import Ipv4AclParcel
 from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv6acl import Ipv6AclParcel
+from catalystwan.models.configuration.feature_profile.sdwan.application_priority.policy_settings import (
+    PolicySettingsParcel,
+)
+from catalystwan.models.configuration.feature_profile.sdwan.application_priority.qos_policy import QosPolicyParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.route_policy import RoutePolicyParcel
 from catalystwan.models.configuration.feature_profile.sdwan.system.device_access import DeviceAccessIPv4Parcel
 from catalystwan.models.configuration.feature_profile.sdwan.system.device_access_ipv6 import DeviceAccessIPv6Parcel
 from catalystwan.session import ManagerSession
 from catalystwan.utils.config_migration.creators.references_updater import update_parcel_references
 
+_AnyApplicationPriorityPolicyParcel = Annotated[
+    Union[QosPolicyParcel, PolicySettingsParcel],
+    Field(discriminator="type_"),
+]
+
 _AnyTransportPolicyFeatureParcel = Annotated[
     Union[Ipv4AclParcel, Ipv6AclParcel, RoutePolicyParcel],
     Field(discriminator="type_"),
@@ -36,7 +47,12 @@
     Field(discriminator="type_"),
 ]
 _AnyLocalizedPolicyParcel = Annotated[
-    Union[_AnyTransportPolicyFeatureParcel, _AnyServicePolicyFeatureParcel, _AnySystemPolicyFeatureParcel],
+    Union[
+        _AnyTransportPolicyFeatureParcel,
+        _AnyServicePolicyFeatureParcel,
+        _AnySystemPolicyFeatureParcel,
+        _AnyApplicationPriorityPolicyParcel,
+    ],
     Field(discriminator="type_"),
 ]
 _LocalizedPolicyProfileTypes = Literal["transport", "service", "system"]
@@ -66,6 +82,7 @@ def __init__(
         self._ux2_config = ux2_config
         self._session = session
         self._cg_api = session.api.config_group
+        self._app_prio_api = session.api.sdwan_feature_profiles.application_priority
         self._push_result: UX2ConfigPushResult = push_result
         self._push_context = push_context
         self._progress: Callable[[str, int, int], None] = progress
@@ -146,6 +163,13 @@ def _update_system_profile(
         parcel.parcel_name += "_system"
         return api.create_parcel(profile_id=profile_id, payload=parcel).id
 
+    def _get_all_application_priority_profiles_with_subelements(self) -> List[TransformedFeatureProfile]:
+        return [
+            p
+            for p in self._ux2_config.feature_profiles
+            if p.header.type == "application-priority" and p.header.localized_policy_subelements is not None
+        ]
+
     def associate_config_groups_with_default_policy_object_profile(self):
         for cg_id, cg in self._get_config_group_contents(self._find_config_groups_to_update()).items():
             profile_ids = [p.id for p in cg.profiles]
@@ -166,6 +190,7 @@ def push(self):
         self._progress("Associating Config Groups with Default Policy Object Profile", 0, 1)
         self.associate_config_groups_with_default_policy_object_profile()
         self._progress("Associating Config Groups with Default Policy Object Profile", 1, 1)
+        # ----- ACLs, Route Policy, Device Access -----
         profile_infos = self._find_profiles_to_update()
         profile_ids = [t[1] for t in profile_infos]
         profile_reports = self._create_profile_report_by_id_lookup(profile_ids)
@@ -194,3 +219,30 @@ def push(self):
                     report.add_failed_parcel(
                         parcel_name=error_parcel.parcel_name, parcel_type=error_parcel.type_, error_info=error_info
                     )
+        # ----- QoSMap, Settings -----
+        app_prio_profiles = self._get_all_application_priority_profiles_with_subelements()
+        for i, app_prio_profile in enumerate(app_prio_profiles):
+            self._progress("Creating Application Priority profile with policy parcels", i + 1, len(app_prio_profiles))
+            try:
+                profile_id = self._app_prio_api.create_profile(
+                    app_prio_profile.feature_profile.name, app_prio_profile.feature_profile.description
+                ).id
+            except ManagerHTTPError as e:
+                logger.error(f"Error occured during Application Priority profile creation: {e.info}")
+                continue
+            report = FeatureProfileBuildReport(
+                profile_name=app_prio_profile.feature_profile.name, profile_uuid=profile_id
+            )
+            self._push_result.report.add_feature_profiles_not_assosiated_with_config_group(report)
+            self._push_result.rollback.add_feature_profile(profile_id, "application-priority")
+            transformed_parcels = self._get_parcels_to_push(list(app_prio_profile.header.localized_policy_subelements))
+            for tp in transformed_parcels:
+                parcel = tp.parcel
+                try:
+                    parcel_id = self._app_prio_api.create_parcel(profile_id=profile_id, payload=parcel).id
+                    report.add_created_parcel(parcel_name=parcel.parcel_name, parcel_uuid=parcel_id)
+                except ManagerHTTPError as e:
+                    logger.error(f"Error occured during Application Priority parcel creation: {e.info}")
+                    report.add_failed_parcel(
+                        parcel_name=parcel.parcel_name, parcel_type=parcel._get_parcel_type(), error_info=e.info
+                    )