diff --git a/catalystwan/models/configuration/config_migration.py b/catalystwan/models/configuration/config_migration.py index f59a69fc..edfa7cb2 100644 --- a/catalystwan/models/configuration/config_migration.py +++ b/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 diff --git a/catalystwan/models/configuration/feature_profile/common.py b/catalystwan/models/configuration/feature_profile/common.py index d97ca6e9..0081f1be 100644 --- a/catalystwan/models/configuration/feature_profile/common.py +++ b/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[ diff --git a/catalystwan/models/policy/localized.py b/catalystwan/models/policy/localized.py index 4538abff..127b9540 100644 --- a/catalystwan/models/policy/localized.py +++ b/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): diff --git a/catalystwan/tests/config_migration/test_data/__init__.py b/catalystwan/tests/config_migration/test_data/__init__.py index 9666794b..161dad78 100644 --- a/catalystwan/tests/config_migration/test_data/__init__.py +++ b/catalystwan/tests/config_migration/test_data/__init__.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 @@ -5,6 +6,8 @@ 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", ] diff --git a/catalystwan/tests/config_migration/test_data/localized_policies/localized_policy.py b/catalystwan/tests/config_migration/test_data/localized_policies/localized_policy.py new file mode 100644 index 00000000..65b9a09e --- /dev/null +++ b/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 diff --git a/catalystwan/tests/config_migration/test_data/policy_definitions/__init__.py b/catalystwan/tests/config_migration/test_data/policy_definitions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/catalystwan/tests/config_migration/test_data/policy_definitions/qos_map.py b/catalystwan/tests/config_migration/test_data/policy_definitions/qos_map.py new file mode 100644 index 00000000..602e24b9 --- /dev/null +++ b/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 diff --git a/catalystwan/tests/config_migration/test_transform.py b/catalystwan/tests/config_migration/test_transform.py index c04cf3f8..3071d992 100644 --- a/catalystwan/tests/config_migration/test_transform.py +++ b/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 diff --git a/catalystwan/utils/config_migration/converters/policy/policy_settings.py b/catalystwan/utils/config_migration/converters/policy/policy_settings.py new file mode 100644 index 00000000..3ce3a9e9 --- /dev/null +++ b/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"]) diff --git a/catalystwan/utils/config_migration/creators/localized_policy_pusher.py b/catalystwan/utils/config_migration/creators/localized_policy_pusher.py index a07a0c36..0357fa25 100644 --- a/catalystwan/utils/config_migration/creators/localized_policy_pusher.py +++ b/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,6 +11,7 @@ from catalystwan.exceptions import ManagerErrorInfo, ManagerHTTPError from catalystwan.models.configuration.config_migration import ( PushContext, + TransformedFeatureProfile, TransformedParcel, UX2Config, UX2ConfigPushResult, @@ -17,12 +19,21 @@ 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 + ) diff --git a/catalystwan/utils/config_migration/creators/security_policy_pusher.py b/catalystwan/utils/config_migration/creators/security_policy_pusher.py index 0119b5d9..9d655e32 100644 --- a/catalystwan/utils/config_migration/creators/security_policy_pusher.py +++ b/catalystwan/utils/config_migration/creators/security_policy_pusher.py @@ -1,9 +1,9 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates import logging from typing import Callable, cast from uuid import UUID from catalystwan.api.builders.feature_profiles.report import FeatureProfileBuildReport -from catalystwan.api.configuration_groups.parcel import Global from catalystwan.exceptions import ManagerHTTPError from catalystwan.models.configuration.config_migration import ( PushContext, @@ -12,11 +12,6 @@ UX2ConfigPushResult, ) from catalystwan.models.configuration.feature_profile.parcel import AnyDnsSecurityParcel, list_types -from catalystwan.models.configuration.feature_profile.sdwan.application_priority import ( - Cflowd, - PolicySettingsParcel, - QosPolicyParcel, -) from catalystwan.models.configuration.feature_profile.sdwan.embedded_security import NgfirewallParcel, PolicyParcel from catalystwan.models.configuration.feature_profile.sdwan.embedded_security.policy import NgFirewallContainer from catalystwan.session import ManagerSession @@ -46,19 +41,6 @@ def __init__( def push(self) -> None: self.push_dns_security_policies() self.push_embedded_security_policies() - self.push_app_priority_and_sla_policies() - - def _push_app_priority_and_sla_profile(self, name: str, description: str) -> FeatureProfileBuildReport: - try: - profile_id = self._app_profile_api.create_profile(name, description).id - self._push_result.rollback.add_feature_profile(profile_id, "application-priority") - feature_profile_report = FeatureProfileBuildReport(profile_name=name, profile_uuid=profile_id) - except ManagerHTTPError as e: - logger.error(f"Error occured during App priority and SLAprofile creation: {e.info}") - feature_profile_report = FeatureProfileBuildReport(profile_name=name, profile_uuid=UUID(int=0)) - - self._push_result.report.security_policies.append(feature_profile_report) - return feature_profile_report def _push_embedded_security_profile(self, name: str, description: str) -> FeatureProfileBuildReport: try: @@ -100,61 +82,6 @@ def _push_policy_parcel( logger.error(f"Error occured during creating PolicyParcel in embedded security profile: {e.info}") report.add_failed_parcel(parcel_name=parcel.parcel_name, parcel_type=parcel.type_, error_info=e.info) - def push_app_priority_and_sla_policies(self) -> None: - qos_map_policies = [ - transformed_parcel - for transformed_parcel in self._ux2_config.profile_parcels - if type(transformed_parcel.parcel) in list_types(QosPolicyParcel) - ] - - for i, qos_policy in enumerate(qos_map_policies): - parcel = cast(QosPolicyParcel, qos_policy.parcel) - parcel = update_parcel_references(parcel, self.push_context.id_lookup) - - msg = f"Creating app priorit and sla policies: {parcel.parcel_name}" - self._progress(msg, i + 1, len(qos_map_policies)) - - profile_report = self._push_app_priority_and_sla_profile(parcel.parcel_name, parcel.parcel_description) - - if (profile_id := profile_report.profile_uuid) == UUID(int=0): - continue - - try: - qos_id = self._app_profile_api.create_parcel(profile_id, parcel).id - profile_report.add_created_parcel(parcel.parcel_name, qos_id) - self.push_context.id_lookup[qos_policy.header.origin] = qos_id - except ManagerHTTPError as e: - logger.error(f"Error occured during creating QoSParcel in app priority and sla profile: {e.info}") - profile_report.add_failed_parcel( - parcel_name=parcel.parcel_name, parcel_type=parcel.type_, error_info=e.info - ) - continue - - # ------- temporary - need to find source of that settings ------- - policy_settings_parcel = PolicySettingsParcel( - parcel_description="desc", - parcel_name=parcel.parcel_name + "set", - app_visibility=Global[bool](value=False), - flow_visibility=Global[bool](value=False), - app_visibility_ipv6=Global[bool](value=False), - flow_visibility_ipv6=Global[bool](value=False), - cflowd=Cflowd(value=True), - ) - - try: - qos_id = self._app_profile_api.create_parcel(profile_id, policy_settings_parcel).id - profile_report.add_created_parcel(policy_settings_parcel.parcel_name, qos_id) - # self.push_context.id_lookup[qos_policy.header.origin] = qos_id # update when source is found - except ManagerHTTPError as e: - logger.error(f"Error occured during creating QoSParcel in app priority and sla profile: {e.info}") - profile_report.add_failed_parcel( - parcel_name=policy_settings_parcel.parcel_name, - parcel_type=policy_settings_parcel.type_, - error_info=e.info, - ) - continue - # ------- temporary ------- - def push_embedded_security_policies(self) -> None: security_policies = [ transformed_parcel diff --git a/catalystwan/workflows/config_migration.py b/catalystwan/workflows/config_migration.py index d7d2c82a..d597f8a4 100644 --- a/catalystwan/workflows/config_migration.py +++ b/catalystwan/workflows/config_migration.py @@ -33,6 +33,7 @@ ) from catalystwan.utils.config_migration.converters.policy.policy_definitions import convert as convert_policy_definition from catalystwan.utils.config_migration.converters.policy.policy_lists import convert as convert_policy_list +from catalystwan.utils.config_migration.converters.policy.policy_settings import convert_localized_policy_settings from catalystwan.utils.config_migration.converters.policy.security_policy import convert_security_policy from catalystwan.utils.config_migration.creators.config_pusher import UX2ConfigPusher, UX2ConfigPushResult from catalystwan.utils.config_migration.reverters.config_reverter import UX2ConfigReverter @@ -426,25 +427,74 @@ def transform(ux1: UX1Config, add_suffix: bool = False) -> ConfigTransformResult _lookup = ux1.templates.create_device_template_by_policy_id_lookup() dt_by_policy_id = _lookup["policy"] for localized_policy in ux1.policies.localized_policies: - if not isinstance(localized_policy.policy_definition, str): - if dt_id := dt_by_policy_id.get(localized_policy.policy_id): - for item in localized_policy.policy_definition.assembly: - if item.type == "deviceaccesspolicy" or item.type == "deviceaccesspolicyv6": - ux2.add_subelement_in_config_group( - profile_types=["system"], device_template_id=dt_id, subelement=item.definition_id - ) - elif item.type == "acl" or item.type == "aclv6": - ux2.add_subelement_in_config_group( - profile_types=["transport", "service"], - device_template_id=dt_id, - subelement=item.definition_id, - ) - elif item.type == "vedgeRoute": - ux2.add_subelement_in_config_group( - profile_types=["transport", "service"], - device_template_id=dt_id, - subelement=item.definition_id, - ) + if isinstance(localized_policy.policy_definition, str): + continue + application_prio = TransformedFeatureProfile( + header=TransformHeader( + type="application-priority", + origin=localized_policy.policy_id, + origname=localized_policy.policy_name, + localized_policy_subelements=set(), + ), + feature_profile=FeatureProfileCreationPayload( + name=f"FROM_{localized_policy.policy_name}", + description=localized_policy.policy_description, + ), + ) + ux2.feature_profiles.append(application_prio) + settings_result = convert_localized_policy_settings( + localized_policy, localized_policy.policy_id, policy_context + ) + settings_parcel = settings_result.output + settings_status = settings_result.status + if settings_status == "unsupported": + transform_result.add_unsupported_item( + name=f"{localized_policy.policy_name} Settings", + uuid=policy_definition.definition_id, + type=policy_definition.type, + ) + elif settings_status == "failed": + transform_result.add_failed_conversion_parcel( + exception_message="\n".join(pd_result.info), + policy=localized_policy, + ) + elif settings_parcel is not None: + header = TransformHeader( + type=settings_parcel._get_parcel_type(), + origin=localized_policy.policy_id, + origname=localized_policy.policy_description, + status=settings_status, + info=settings_result.info, + ) + tp_settings = TransformedParcel(header=header, parcel=settings_parcel) + ux2.profile_parcels.append(tp_settings) + application_prio.header.add_localized_policy_subelement(tp_settings.header.origin) + + for item in localized_policy.policy_definition.assembly: + # ----- Device Template Independent Items ----- + if item.type == "qosMap": + application_prio.header.add_localized_policy_subelement(item.definition_id) + continue + # ----- Device Template Dependent Items ----- + dt_id = dt_by_policy_id.get(localized_policy.policy_id) + if dt_id is None: + continue + if item.type == "deviceaccesspolicy" or item.type == "deviceaccesspolicyv6": + ux2.add_subelement_in_config_group( + profile_types=["system"], device_template_id=dt_id, subelement=item.definition_id + ) + elif item.type == "acl" or item.type == "aclv6": + ux2.add_subelement_in_config_group( + profile_types=["transport", "service"], + device_template_id=dt_id, + subelement=item.definition_id, + ) + elif item.type == "vedgeRoute": + ux2.add_subelement_in_config_group( + profile_types=["transport", "service"], + device_template_id=dt_id, + subelement=item.definition_id, + ) # Security policies for security_policy in ux1.policies.security_policies: