dev-uxmt: convert and push hubspoke/mesh, fix bugs, bump version (#701)

* prep work

* move acl parcel to common

* move acl parcel to acl, fix resolver

* add ipv4acl model and integration tests

* fix unit tests

* add ipv4acl converter, add integration tests for service profile

* add original name to header before conversion

* add network-hierarchy endpoints

* collect network hierarchy during migration flow

* fix typo in serialization alias

* push/rollback for topology items, emit vpn, site, region assignemt during conversion

* hubspoke converter draft

* fix hubspoke converter 2

* Remove duplicated feature templates, add unit test

* Add interface to ft list

* Change logic to remove all unused templates, not only copied (#706)

* Change logic to remove all unused templates, not only copied

* Fix. take name from deepcopied model

* push/rollback hubspoke mesh parcels

* fix bugs, bump version

---------

Co-authored-by: Jakub Krajewski <[email protected]>
Co-authored-by: Jakub Krajewski <[email protected]>

catalystwan/api/feature_profile_api.py

@@ -15,6 +15,8 @@
 from catalystwan.endpoints.configuration.feature_profile.sdwan.topology import TopologyFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.transport import TransportFeatureProfile
 from catalystwan.exceptions import ManagerHTTPError
+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 import (
     AnyApplicationPriorityParcel,
     PolicySettingsParcel,
@@ -288,6 +290,14 @@ def get_parcel(
     ) -> Parcel[ManagementVpnParcel]:
         ...
 
+    @overload
+    def get_parcel(self, profile_id: UUID, parcel_type: Type[Ipv4AclParcel], parcel_id: UUID) -> Parcel[Ipv4AclParcel]:
+        ...
+
+    @overload
+    def get_parcel(self, profile_id: UUID, parcel_type: Type[Ipv6AclParcel], parcel_id: UUID) -> Parcel[Ipv6AclParcel]:
+        ...
+
     def get_parcel(
         self, profile_id: UUID, parcel_type: Type[Union[AnyTransportParcel, AnyRoutingParcel]], parcel_id: UUID
     ) -> Parcel:

catalystwan/endpoints/__init__.py

@@ -52,7 +52,6 @@
     Sequence,
     Set,
     Tuple,
-    Type,
     TypeVar,
     Union,
     runtime_checkable,
@@ -116,12 +115,18 @@ def model_union(cls, models: Sequence[type]) -> TypeSpecifier:
         return TypeSpecifier(present=True, payload_union_model_types=models)
 
     @classmethod
-    def resolve_nested(cls, annotation: Any, models_types: List[Type[BaseModel]]) -> List[Type[BaseModel]]:
+    def resolve_nested(cls, annotation: Any, models_types: List[type]) -> List[type]:
         try:
             return resolve_nested_base_model_unions(annotation)
         except TypeError:
             raise APIEndpointError(f"Expected: {PayloadType}")
 
+    @property
+    def payload_model_set(self) -> Optional[Set[type]]:
+        if self.payload_union_model_types is not None:
+            return set(self.payload_union_model_types)
+        return None
+
 
 @dataclass
 class APIEndpointRequestMeta:

catalystwan/endpoints/configuration/network_hierarchy.py

@@ -0,0 +1,13 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+
+# mypy: disable-error-code="empty-body"
+from catalystwan.endpoints import APIEndpoints, get, versions
+from catalystwan.models.configuration.network_hierarchy import NodeInfo
+from catalystwan.typed_list import DataSequence
+
+
+class NetworkHierarchy(APIEndpoints):
+    @get("/v1/network-hierarchy")
+    @versions(">=20.10")
+    def list_nodes(self) -> DataSequence[NodeInfo]:
+        ...

catalystwan/endpoints/configuration/topology_group.py

@@ -0,0 +1,25 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+
+# mypy: disable-error-code="empty-body"
+from uuid import UUID
+
+from catalystwan.endpoints import APIEndpoints, delete, get, post, versions
+from catalystwan.models.configuration.topology_group import TopologyGroup, TopologyGroupId
+from catalystwan.typed_list import DataSequence
+
+
+class TopologyGroupEndpoints(APIEndpoints):
+    @post("/v1/topology-group")
+    @versions(">=20.12")
+    def create_topology_group(self, payload: TopologyGroup) -> TopologyGroupId:
+        ...
+
+    @get("/v1/topology-group")
+    @versions(">=20.12")
+    def get_all(self) -> DataSequence[TopologyGroupId]:
+        ...
+
+    @delete("/v1/topology-group/{group_id}")
+    @versions(">=20.12")
+    def delete(self, group_id: UUID) -> None:
+        ...

catalystwan/endpoints/endpoints_container.py

@@ -16,6 +16,7 @@
 from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.topology import TopologyFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.transport import TransportFeatureProfile
+from catalystwan.endpoints.configuration.network_hierarchy import NetworkHierarchy
 from catalystwan.endpoints.configuration.policy.definition.access_control_list import ConfigurationPolicyAclDefinition
 from catalystwan.endpoints.configuration.policy.definition.access_control_list_ipv6 import (
     ConfigurationPolicyAclIPv6Definition,
@@ -76,6 +77,7 @@
 from catalystwan.endpoints.configuration.policy.vedge_template import ConfigurationVEdgeTemplatePolicy
 from catalystwan.endpoints.configuration.policy.vsmart_template import ConfigurationVSmartTemplatePolicy
 from catalystwan.endpoints.configuration.software_actions import ConfigurationSoftwareActions
+from catalystwan.endpoints.configuration.topology_group import TopologyGroupEndpoints
 from catalystwan.endpoints.configuration_dashboard_status import ConfigurationDashboardStatus
 from catalystwan.endpoints.configuration_device_actions import ConfigurationDeviceActions
 from catalystwan.endpoints.configuration_device_inventory import ConfigurationDeviceInventory
@@ -178,7 +180,8 @@ def __init__(self, session: ManagerSession):
 class ConfigurationContainer:
     def __init__(self, session: ManagerSession):
         self.policy = ConfigurationPolicyContainer(session)
-        self.feature_profile = ConfigurationFeatureProfileContainer(session=session)
+        self.feature_profile = ConfigurationFeatureProfileContainer(session)
+        self.topology_group = TopologyGroupEndpoints(session)
 
 
 class TroubleshootingToolsContainer:
@@ -208,6 +211,7 @@ def __init__(self, session: ManagerSession):
         self.monitoring_device_details = MonitoringDeviceDetails(session)
         self.monitoring_server_info = ServerInfo(session)
         self.monitoring_status = MonitoringStatus(session)
+        self.network_hierarchy = NetworkHierarchy(session)
         self.sdavc_cloud_connector = SDAVCCloudConnector(session)
         self.tenant_backup_restore = TenantBackupRestore(session)
         self.tenant_management = TenantManagement(session)

catalystwan/integration_tests/feature_profile/sdwan/test_service.py

@@ -1,5 +1,5 @@
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
-from ipaddress import IPv4Address, IPv6Interface
+from ipaddress import IPv4Address, IPv4Interface, IPv6Interface
 from typing import Literal
 from uuid import UUID
 
@@ -26,6 +26,8 @@
     MultilinkNimList,
     MultilinkTxExName,
 )
+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.routing.ospf import RoutingOspfParcel
 from catalystwan.models.configuration.feature_profile.sdwan.routing.ospfv3 import (
     Ospfv3InterfaceParametres,
@@ -34,7 +36,6 @@
     RoutingOspfv3IPv4Parcel,
     RoutingOspfv3IPv6Parcel,
 )
-from catalystwan.models.configuration.feature_profile.sdwan.service.acl import Ipv4AclParcel, Ipv6AclParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import (
     AddressPool,
     LanVpnDhcpServerParcel,
@@ -207,6 +208,42 @@ def test_when_default_values_route_policy_parcel_expect_successful_post(self):
         # Assert
         assert parcel_id
 
+    def test_when_default_values_acl_ipv4_expect_successful_post(self):
+        # Arrange
+        acl_ipv4_parcel = Ipv4AclParcel(
+            parcel_name="TestAclIpv4Parcel",
+            parcel_description="Test Acl Ipv4 Parcel",
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id
+        # Assert
+        assert parcel_id
+
+    def test_when_fully_specified_acl_ipv4_expect_successful_post(self):
+        # Arrange
+        acl_ipv4_parcel = Ipv4AclParcel(
+            parcel_name="TestAclIpv4Parcel-Full",
+            parcel_description="Test Acl Ipv4 Parcel",
+        )
+        # Arrange Sequence 1
+        seq1 = acl_ipv4_parcel.add_sequence("Sequence1", 10, "accept")
+        seq1.match_destination_data_prefix(IPv4Interface("10.0.0.0/16"))
+        seq1.match_dscp([50, 55])
+        seq1.match_icmp_msg(["dod-host-prohibited", "extended-echo", "dod-net-prohibited"])
+        seq1.match_packet_length((1000, 8000))
+        seq1.match_protocol([1])
+        seq1.match_source_data_prefix(IPv4Interface("11.0.0.0/16"))
+        # Arrange Sequence 2
+        seq2 = acl_ipv4_parcel.add_sequence("Sequence2", 20, "drop")
+        seq2.match_destination_data_prefix_variable("varDestPrefix2")
+        seq2.match_source_data_prefix_variable("varSrcPrefix2")
+        seq2.match_destination_ports([233])
+        seq2.match_source_ports([1, 3, (10, 100), (50, 200), 600])
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id
+        # Assert
+        assert parcel_id
+
     def test_when_default_values_acl_ipv6_expect_successful_post(self):
         # Arrange
         acl_ipv6_parcel = Ipv6AclParcel(
@@ -218,14 +255,25 @@ def test_when_default_values_acl_ipv6_expect_successful_post(self):
         # Assert
         assert parcel_id
 
-    def test_when_default_values_acl_ipv4_expect_successful_post(self):
+    def test_when_fully_specified_acl_ipv6_expect_successful_post(self):
         # Arrange
-        acl_ipv4_parcel = Ipv4AclParcel(
-            parcel_name="TestAclIpv4Parcel",
-            parcel_description="Test Acl Ipv4 Parcel",
+        acl_ipv6_parcel = Ipv6AclParcel(
+            parcel_name="TestAclIpv6Parcel-Full",
+            parcel_description="Test Acl Ipv6 Parcel",
         )
+        # Arrange Sequence 1
+        seq1 = acl_ipv6_parcel.add_sequence("Sequence1", 10, "accept")
+        seq1.match_destination_data_prefix(IPv6Interface("2001:db8:abcd:0012::/64"))
+        seq1.match_icmp_msg(["cp-solicitation", "ind-advertisement"])
+        seq1.match_packet_length((1000, 8000))
+        seq1.match_source_data_prefix(IPv6Interface("2001:db8:1111:0012::/64"))
+        seq1.match_traffic_class([3])
+        # Arrange Sequence 2
+        seq2 = acl_ipv6_parcel.add_sequence("Sequence2", 20, "drop")
+        seq2.match_destination_ports([233])
+        seq2.match_source_ports([1, 3, (10, 100), (50, 200), 600])
         # Act
-        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv6_parcel).id
         # Assert
         assert parcel_id
 

catalystwan/integration_tests/feature_profile/sdwan/test_transport.py

@@ -1,5 +1,5 @@
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
-from ipaddress import IPv4Address, IPv6Address, IPv6Interface
+from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
 from typing import List, Literal
 from uuid import UUID
 
@@ -51,6 +51,8 @@
     StaticNat,
     TunnelSourceType,
 )
+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.transport.management.ethernet import (
     Advanced as ManagementEthernetAdvanced,
 )
@@ -298,6 +300,79 @@ def test_when_fully_specifed_cellular_profile_expect_successful_post(self):
         # Assert
         assert parcel_id
 
+    def test_when_default_values_acl_ipv4_expect_successful_post(self):
+        # Arrange
+        acl_ipv4_parcel = Ipv4AclParcel(
+            parcel_name="TestAclIpv4Parcel-Defaults",
+            parcel_description="Test Acl Ipv4 Parcel",
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id
+        self.api.get_parcel(self.profile_uuid, Ipv4AclParcel, parcel_id)
+        # Assert
+        assert parcel_id
+
+    def test_when_fully_specified_acl_ipv4_expect_successful_post(self):
+        # Arrange
+        acl_ipv4_parcel = Ipv4AclParcel(
+            parcel_name="TestAclIpv4Parcel-Full",
+            parcel_description="Test Acl Ipv4 Parcel",
+        )
+        # Arrange Sequence 1
+        seq1 = acl_ipv4_parcel.add_sequence("Sequence1", 10, "accept")
+        seq1.match_destination_data_prefix(IPv4Interface("10.0.0.0/16"))
+        seq1.match_dscp([50, 55])
+        seq1.match_icmp_msg(["dod-host-prohibited", "extended-echo", "dod-net-prohibited"])
+        seq1.match_packet_length((1000, 8000))
+        seq1.match_protocol([1])
+        seq1.match_source_data_prefix(IPv4Interface("11.0.0.0/16"))
+        # Arrange Sequence 2
+        seq2 = acl_ipv4_parcel.add_sequence("Sequence2", 20, "drop")
+        seq2.match_destination_data_prefix_variable("varDestPrefix2")
+        seq2.match_source_data_prefix_variable("varSrcPrefix2")
+        seq2.match_destination_ports([233])
+        seq2.match_source_ports([1, 3, (10, 100), (50, 200), 600])
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv4_parcel).id
+        self.api.get_parcel(self.profile_uuid, Ipv4AclParcel, parcel_id)
+        # Assert
+        assert parcel_id
+
+    def test_when_default_values_acl_ipv6_expect_successful_post(self):
+        # Arrange
+        acl_ipv6_parcel = Ipv6AclParcel(
+            parcel_name="TestAclIpv6Parcel-Defaults",
+            parcel_description="Test Acl Ipv6 Parcel",
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv6_parcel).id
+        self.api.get_parcel(self.profile_uuid, Ipv6AclParcel, parcel_id)
+        # Assert
+        assert parcel_id
+
+    def test_when_fully_specified_acl_ipv6_expect_successful_post(self):
+        # Arrange
+        acl_ipv6_parcel = Ipv6AclParcel(
+            parcel_name="TestAclIpv6Parcel-Full",
+            parcel_description="Test Acl Ipv6 Parcel",
+        )
+        # Arrange Sequence 1
+        seq1 = acl_ipv6_parcel.add_sequence("Sequence1", 10, "accept")
+        seq1.match_destination_data_prefix(IPv6Interface("2001:db8:abcd:0012::/64"))
+        seq1.match_icmp_msg(["cp-solicitation", "ind-advertisement"])
+        seq1.match_packet_length((1000, 8000))
+        seq1.match_source_data_prefix(IPv6Interface("2001:db8:1111:0012::/64"))
+        seq1.match_traffic_class([3])
+        # Arrange Sequence 2
+        seq2 = acl_ipv6_parcel.add_sequence("Sequence2", 20, "drop")
+        seq2.match_destination_ports([233])
+        seq2.match_source_ports([1, 3, (10, 100), (50, 200), 600])
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, acl_ipv6_parcel).id
+        self.api.get_parcel(self.profile_uuid, Ipv6AclParcel, parcel_id)
+        # Assert
+        assert parcel_id
+
     @classmethod
     def tearDownClass(cls) -> None:
         cls.api.delete_profile(cls.profile_uuid)

catalystwan/models/common.py

@@ -178,14 +178,19 @@ def str_as_str_list(val: Union[str, Sequence[str]]) -> Sequence[str]:
 ]
 
 
-def int_range_str_validator(value: Union[str, IntRange], ascending: bool = True) -> IntRange:
-    """Validates input given as string containing integer pair separated by hyphen eg: '1-3' or single number '1'"""
+def int_range_str_validator(value: Union[str, int, IntRange], ascending: bool = True) -> IntRange:
+    """
+    Validates input given as string containing integer pair separated by hyphen
+    eg: '1-3' or single number '1'
+    """
     if isinstance(value, str):
         int_list = [int(i) for i in value.strip().split("-")]
-        assert 0 < len(int_list) <= 2, "Number range must contain one or two numbers"
+        assert 0 < len(int_list) <= 2, "Number range string must contain one or two numbers"
         first = int_list[0]
         second = None if len(int_list) == 1 else int_list[1]
         int_range = (first, second)
+    elif isinstance(value, int):
+        int_range = (value, None)
     else:
         int_range = value
     if ascending and int_range[1] is not None:
@@ -204,6 +209,7 @@ def int_range_serializer(value: IntRange) -> str:
     BeforeValidator(int_range_str_validator),
 ]
 
+BasicPolicyActionType = Literal["accept", "drop"]
 
 CarrierType = Literal[
     "default",