From 0a25de917d9d12a9197956c081d07f039f5f26fc Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Wed, 17 Jul 2024 12:25:06 +0200 Subject: [PATCH 1/3] Add localapp converter. Add unit tests --- catalystwan/models/policy/list/local_app.py | 12 ++++ .../policy_converters/test_app_list.py | 56 +++++++++++++++++++ .../converters/policy/policy_lists.py | 30 ++++++++-- 3 files changed, 92 insertions(+), 6 deletions(-) create mode 100644 catalystwan/tests/config_migration/policy_converters/test_app_list.py diff --git a/catalystwan/models/policy/list/local_app.py b/catalystwan/models/policy/list/local_app.py index d0fe37ed..d319fd03 100644 --- a/catalystwan/models/policy/list/local_app.py +++ b/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 diff --git a/catalystwan/tests/config_migration/policy_converters/test_app_list.py b/catalystwan/tests/config_migration/policy_converters/test_app_list.py new file mode 100644 index 00000000..cd4e0ac4 --- /dev/null +++ b/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" diff --git a/catalystwan/utils/config_migration/converters/policy/policy_lists.py b/catalystwan/utils/config_migration/converters/policy/policy_lists.py index d94f50a9..6afca8ee 100644 --- a/catalystwan/utils/config_migration/converters/policy/policy_lists.py +++ b/catalystwan/utils/config_migration/converters/policy/policy_lists.py @@ -36,6 +36,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, @@ -65,6 +68,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 @@ -76,6 +80,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: @@ -85,9 +93,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") @@ -380,26 +388,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, @@ -413,8 +433,6 @@ def zone(in_: ZoneList, context) -> ConvertResult[SecurityZoneListParcel]: URLBlockList: url_block, VPNList: vpn, ZoneList: zone, - MirrorList: mirror, - ExtendedCommunityList: extended_community, } From 25d3bc50a3669458dd92b2149809f514acaeeb3f Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Wed, 17 Jul 2024 13:01:05 +0200 Subject: [PATCH 2/3] Add integration tests --- .../feature_profile/sdwan/policy/__init__.py | 0 .../feature_profile/sdwan/policy/base.py | 32 +++++++++++ .../sdwan/policy/test_extended_community.py | 33 +++++++++++ .../sdwan/policy/test_security_app_list.py | 32 +++++++++++ .../sdwan/test_extended_community.py | 56 ------------------- 5 files changed, 97 insertions(+), 56 deletions(-) create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/policy/__init__.py create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/policy/base.py create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/policy/test_extended_community.py create mode 100644 catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py delete mode 100644 catalystwan/integration_tests/feature_profile/sdwan/test_extended_community.py diff --git a/catalystwan/integration_tests/feature_profile/sdwan/policy/__init__.py b/catalystwan/integration_tests/feature_profile/sdwan/policy/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/catalystwan/integration_tests/feature_profile/sdwan/policy/base.py b/catalystwan/integration_tests/feature_profile/sdwan/policy/base.py new file mode 100644 index 00000000..a408cc14 --- /dev/null +++ b/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() diff --git a/catalystwan/integration_tests/feature_profile/sdwan/policy/test_extended_community.py b/catalystwan/integration_tests/feature_profile/sdwan/policy/test_extended_community.py new file mode 100644 index 00000000..36788c2c --- /dev/null +++ b/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" diff --git a/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py b/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py new file mode 100644 index 00000000..93397a0a --- /dev/null +++ b/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 TestSecurityApplicationListParcelParcel(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" diff --git a/catalystwan/integration_tests/feature_profile/sdwan/test_extended_community.py b/catalystwan/integration_tests/feature_profile/sdwan/test_extended_community.py deleted file mode 100644 index 411fb212..00000000 --- a/catalystwan/integration_tests/feature_profile/sdwan/test_extended_community.py +++ /dev/null @@ -1,56 +0,0 @@ -# Copyright 2024 Cisco Systems, Inc. and its affiliates -from catalystwan.api.feature_profile_api import PolicyObjectFeatureProfileAPI -from catalystwan.endpoints.configuration_feature_profile import ConfigurationFeatureProfile -from catalystwan.integration_tests.base import TestCaseBase, create_name_with_run_id -from catalystwan.models.configuration.feature_profile.sdwan.policy_object import ExtendedCommunityParcel -from catalystwan.typed_list import DataSequence - -PROFILE_NAME = "Default_Policy_Object_Profile" - - -class TestExtendedCommunityParcel(TestCaseBase): - policy_api: PolicyObjectFeatureProfileAPI - - @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, ExtendedCommunityParcel, self.created_id) - - return super().tearDown() - - def test_get_all_extended_community_parcels(self): - parcels = self.policy_api.get(self.profile_uuid, ExtendedCommunityParcel) - assert type(parcels) is DataSequence - - def test_create_extended_community_parcel(self): - # Arrange - parcel_name = create_name_with_run_id("ExampleTestName") - 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, ExtendedCommunityParcel, 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" From 2b753a2a5833990b19c84df0b9b10a952ce2f9fa Mon Sep 17 00:00:00 2001 From: Jakub Krajewski Date: Wed, 17 Jul 2024 13:03:31 +0200 Subject: [PATCH 3/3] Fix class name --- .../feature_profile/sdwan/policy/test_security_app_list.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py b/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py index 93397a0a..92fba90b 100644 --- a/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py +++ b/catalystwan/integration_tests/feature_profile/sdwan/policy/test_security_app_list.py @@ -7,7 +7,7 @@ from catalystwan.typed_list import DataSequence -class TestSecurityApplicationListParcelParcel(PolicyTestCaseBase): +class TestSecurityApplicationListParcel(PolicyTestCaseBase): parcel_type = SecurityApplicationListParcel def test_get_all_parcels(self):