Merge pull request #768 from cisco-open/coverter/localapp

dev-uxmt: Add localapp converter.

catalystwan/integration_tests/feature_profile/sdwan/policy/base.py

@@ -0,0 +1,32 @@
+from typing import Type
+
+from catalystwan.api.feature_profile_api import PolicyObjectFeatureProfileAPI
+from catalystwan.endpoints.configuration_feature_profile import ConfigurationFeatureProfile
+from catalystwan.integration_tests.base import TestCaseBase
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel
+
+
+class PolicyTestCaseBase(TestCaseBase):
+    policy_api: PolicyObjectFeatureProfileAPI
+    parcel_type: Type[AnyPolicyObjectParcel]
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+        cls.policy_api = cls.session.api.sdwan_feature_profiles.policy_object
+        cls.profile_uuid = (
+            ConfigurationFeatureProfile(cls.session)
+            .get_sdwan_feature_profiles()
+            .filter(profile_type="policy-object")
+            .single_or_default()
+        ).profile_id
+
+    def setUp(self) -> None:
+        self.created_id = None
+        return super().setUp()
+
+    def tearDown(self) -> None:
+        if self.created_id:
+            self.policy_api.delete(self.profile_uuid, self.parcel_type, self.created_id)
+
+        return super().tearDown()

catalystwan/integration_tests/feature_profile/sdwan/policy/test_extended_community.py

@@ -0,0 +1,33 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+
+from catalystwan.integration_tests.base import create_name_with_run_id
+from catalystwan.integration_tests.feature_profile.sdwan.policy.base import PolicyTestCaseBase
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object import ExtendedCommunityParcel
+from catalystwan.typed_list import DataSequence
+
+
+class TestExtendedCommunityParcel(PolicyTestCaseBase):
+    parcel_type = ExtendedCommunityParcel
+
+    def test_get_all_parcels(self):
+        parcels = self.policy_api.get(self.profile_uuid, self.parcel_type)
+        assert type(parcels) is DataSequence
+
+    def test_create_extended_community_parcel(self):
+        # Arrange
+        parcel_name = create_name_with_run_id("ext")
+        ext = ExtendedCommunityParcel(parcel_name=parcel_name)
+        ext.add_route_target_community(100, 2000)
+        ext.add_route_target_community(300, 5000)
+        ext.add_site_of_origin_community("1.2.3.4", 1000)
+        ext.add_site_of_origin_community("10.20.30.40", 3000)
+        # Act
+        self.created_id = self.policy_api.create_parcel(self.profile_uuid, ext).id
+        parcel = self.policy_api.get(self.profile_uuid, self.parcel_type, parcel_id=self.created_id)
+        # Assert
+        assert parcel.payload.parcel_name == parcel_name
+        assert len(parcel.payload.entries) == 4
+        assert parcel.payload.entries[0].extended_community.value == "rt 100:2000"
+        assert parcel.payload.entries[1].extended_community.value == "rt 300:5000"
+        assert parcel.payload.entries[2].extended_community.value == "soo 1.2.3.4:1000"
+        assert parcel.payload.entries[3].extended_community.value == "soo 10.20.30.40:3000"

catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py

@@ -0,0 +1,32 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+from catalystwan.integration_tests.base import create_name_with_run_id
+from catalystwan.integration_tests.feature_profile.sdwan.policy.base import PolicyTestCaseBase
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.application_list import (
+    SecurityApplicationListParcel,
+)
+from catalystwan.typed_list import DataSequence
+
+
+class TestSecurityApplicationListParcel(PolicyTestCaseBase):
+    parcel_type = SecurityApplicationListParcel
+
+    def test_get_all_parcels(self):
+        parcels = self.policy_api.get(self.profile_uuid, self.parcel_type)
+        assert type(parcels) is DataSequence
+
+    def test_createsecurity_application_list_parcel(self):
+        # Arrange
+        parcel_name = create_name_with_run_id("sal")
+        sal = SecurityApplicationListParcel(parcel_name=parcel_name)
+        sal.add_application_family("web")
+        sal.add_application_family("file-server")
+        sal.add_application_family("audio-video")
+        # Act
+        self.created_id = self.policy_api.create_parcel(self.profile_uuid, sal).id
+        parcel = self.policy_api.get(self.profile_uuid, self.parcel_type, parcel_id=self.created_id)
+        # Assert
+        assert parcel.payload.parcel_name == parcel_name
+        assert len(parcel.payload.entries) == 3
+        assert parcel.payload.entries[0].app_list_family.value == "web"
+        assert parcel.payload.entries[1].app_list_family.value == "file-server"
+        assert parcel.payload.entries[2].app_list_family.value == "audio-video"

catalystwan/models/policy/list/local_app.py

@@ -24,6 +24,18 @@ class LocalAppList(PolicyListBase):
     type: Literal["localApp"] = "localApp"
     entries: List[LocalAppListEntry] = []
 
+    def add_app(self, app: str) -> None:
+        self._add_entry(LocalAppListEntry(app=app))
+
+    def add_app_family(self, app_family: str) -> None:
+        self._add_entry(LocalAppListEntry(app_family=app_family))
+
+    def list_all_app(self) -> List[str]:
+        return [e.app for e in self.entries if e.app is not None]
+
+    def list_all_app_family(self) -> List[str]:
+        return [e.app_family for e in self.entries if e.app_family is not None]
+
 
 class LocalAppListEditPayload(LocalAppList, PolicyListId):
     pass

catalystwan/tests/config_migration/policy_converters/test_app_list.py

@@ -0,0 +1,56 @@
+import unittest
+
+from catalystwan.models.configuration.config_migration import PolicyConvertContext
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object.policy.application_list import (
+    ApplicationListParcel,
+)
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.application_list import (
+    SecurityApplicationListParcel,
+)
+from catalystwan.models.policy.list.app import AppList
+from catalystwan.models.policy.list.local_app import LocalAppList
+from catalystwan.utils.config_migration.converters.policy.policy_lists import convert
+
+
+class TestAppListConverter(unittest.TestCase):
+    def setUp(self) -> None:
+        self.context = PolicyConvertContext()
+
+    def test_app_list_conversion(self):
+        policy = AppList(
+            name="app_list",
+            description="app list description",
+        )
+        policy.add_app_family("TestFamily")
+        policy.add_app_family("TestFamily2")
+        # Act
+        parcel = convert(policy, self.context).output
+        # Assert
+        assert isinstance(parcel, ApplicationListParcel)
+        assert parcel.parcel_name == "app_list"
+        assert parcel.parcel_description == "app list description"
+        assert len(parcel.entries) == 2
+        assert parcel.entries[0].app_list_family.value == "TestFamily"
+        assert parcel.entries[1].app_list_family.value == "TestFamily2"
+
+
+class TestLocalAppListConverter(unittest.TestCase):
+    def setUp(self) -> None:
+        self.context = PolicyConvertContext()
+
+    def test_local_app_list_conversion(self):
+        policy = LocalAppList(
+            name="local_app_list",
+            description="app list description",
+        )
+        policy.add_app_family("TestFamily")
+        policy.add_app_family("TestFamily2")
+        # Act
+        parcel = convert(policy, self.context).output
+        # Assert
+        assert isinstance(parcel, SecurityApplicationListParcel)
+        assert parcel.parcel_name == "local_app_list"
+        assert parcel.parcel_description == "app list description"
+        assert len(parcel.entries) == 2
+        assert parcel.entries[0].app_list_family.value == "TestFamily"
+        assert parcel.entries[1].app_list_family.value == "TestFamily2"

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

@@ -37,6 +37,9 @@
     URLBlockParcel,
 )
 from catalystwan.models.configuration.feature_profile.sdwan.policy_object.policy.sla_class import SLAClassCriteria
+from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.application_list import (
+    SecurityApplicationListParcel,
+)
 from catalystwan.models.policy import (
     AnyPolicyList,
     AppList,
@@ -66,6 +69,7 @@
     URLBlockList,
     ZoneList,
 )
+from catalystwan.models.policy.list.local_app import LocalAppList
 from catalystwan.models.policy.list.region import RegionList, RegionListInfo
 from catalystwan.models.policy.list.site import SiteList, SiteListInfo
 from catalystwan.models.policy.list.vpn import VPNList, VPNListInfo
@@ -77,6 +81,10 @@ def _get_parcel_name_desc(policy_list: AnyPolicyList) -> Dict[str, Any]:
     return dict(parcel_name=policy_list.name, parcel_description=policy_list.description)
 
 
+def _get_sorted_unique_list(in_list: List[str]) -> List[str]:
+    return sorted(list(set(in_list)))
+
+
 def app_probe(in_: AppProbeClassList, context) -> ConvertResult[AppProbeParcel]:
     out = AppProbeParcel(**_get_parcel_name_desc(in_))
     for entry in in_.entries:
@@ -86,9 +94,9 @@ def app_probe(in_: AppProbeClassList, context) -> ConvertResult[AppProbeParcel]:
 
 def app_list(in_: AppList, context) -> ConvertResult[ApplicationListParcel]:
     out = ApplicationListParcel(**_get_parcel_name_desc(in_))
-    for app in set(in_.list_all_app()):
+    for app in _get_sorted_unique_list(in_.list_all_app()):
         out.add_application(app)
-    for app_family in set(in_.list_all_app_family()):
+    for app_family in _get_sorted_unique_list(in_.list_all_app_family()):
         out.add_application_family(app_family)
     return ConvertResult[ApplicationListParcel](output=out, status="complete")
 
@@ -403,26 +411,38 @@ def zone(in_: ZoneList, context) -> ConvertResult[SecurityZoneListParcel]:
     return ConvertResult[SecurityZoneListParcel](output=out, status="complete")
 
 
+def local_app_list(in_: LocalAppList, context: PolicyConvertContext) -> ConvertResult[SecurityApplicationListParcel]:
+    out = SecurityApplicationListParcel(**_get_parcel_name_desc(in_))
+    for app in _get_sorted_unique_list(in_.list_all_app()):
+        out.add_application(app)
+    for app_family in _get_sorted_unique_list(in_.list_all_app_family()):
+        out.add_application_family(app_family)
+    return ConvertResult[SecurityApplicationListParcel](output=out, status="complete")
+
+
 OPL = TypeVar("OPL", AnyPolicyObjectParcel, None)
 Input = AnyPolicyList
 Output = ConvertResult[OPL]
 
 
 CONVERTERS: Mapping[Type[Input], Callable[..., Output]] = {
-    AppProbeClassList: app_probe,
-    AppList: app_list,
     ASPathList: as_path,
+    AppList: app_list,
+    AppProbeClassList: app_probe,
     ClassMapList: class_map,
     ColorList: color,
     CommunityList: community,
     DataIPv6PrefixList: data_prefix_ipv6,
     DataPrefixList: data_prefix,
     ExpandedCommunityList: expanded_community,
+    ExtendedCommunityList: extended_community,
     FQDNList: fqdn,
     GeoLocationList: geo_location,
     IPSSignatureList: ips_signature,
     IPv6PrefixList: prefix_ipv6,
+    LocalAppList: local_app_list,
     LocalDomainList: local_domain,
+    MirrorList: mirror,
     PolicerList: policer,
     PortList: port,
     PreferredColorGroupList: preferred_color_group,
@@ -436,8 +456,6 @@ def zone(in_: ZoneList, context) -> ConvertResult[SecurityZoneListParcel]:
     URLBlockList: url_block,
     VPNList: vpn,
     ZoneList: zone,
-    MirrorList: mirror,
-    ExtendedCommunityList: extended_community,
 }