topology parcels draft

update deprecated github actions (#544)

* update deprecated github actions

* fix type error

add missing file

fix import order

catalystwan/api/feature_profile_api.py

@@ -10,11 +10,13 @@
 from catalystwan.endpoints.configuration.feature_profile.sdwan.other import OtherFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.service import ServiceFeatureProfile
 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.models.configuration.feature_profile.sdwan.other import AnyOtherParcel
 from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.url import URLParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.multicast import MulticastParcel
+from catalystwan.models.configuration.feature_profile.sdwan.topology import AnyTopologyParcel
 from catalystwan.models.configuration.feature_profile.sdwan.transport import AnyTransportParcel
 from catalystwan.typed_list import DataSequence
 
@@ -87,6 +89,7 @@ def __init__(self, session: ManagerSession):
         self.system = SystemFeatureProfileAPI(session=session)
         self.other = OtherFeatureProfileAPI(session=session)
         self.service = ServiceFeatureProfileAPI(session=session)
+        self.topology = TopologyFeatureProfileAPI(session=session)
         self.transport = TransportFeatureProfileAPI(session=session)
 
 
@@ -1037,3 +1040,49 @@ def delete(self, profile_id: UUID, parcel_type: Type[AnyPolicyObjectParcel], lis
         return self.endpoint.delete(
             profile_id=profile_id, policy_object_list_type=policy_object_list_type, list_object_id=list_object_id
         )
+
+
+class TopologyFeatureProfileAPI:
+    """
+    SDWAN Feature Profile Topology APIs
+    """
+
+    def __init__(self, session: ManagerSession):
+        self.session = session
+        self.endpoint = TopologyFeatureProfile(session)
+
+    def get_profiles(
+        self, limit: Optional[int] = None, offset: Optional[int] = None
+    ) -> DataSequence[FeatureProfileInfo]:
+        """
+        Get all Service Feature Profiles
+        """
+        payload = GetFeatureProfilesPayload(limit=limit, offset=offset)
+        return self.endpoint.get_topology_feature_profiles(payload)
+
+    def create_profile(self, name: str, description: str) -> FeatureProfileCreationResponse:
+        """
+        Create Service Feature Profile
+        """
+        payload = FeatureProfileCreationPayload(name=name, description=description)
+        return self.endpoint.create_topology_feature_profile(payload)
+
+    def delete_profile(self, profile_id: UUID) -> None:
+        """
+        Delete Service Feature Profile
+        """
+        self.endpoint.delete_topology_feature_profile(profile_id)
+
+    def create_parcel(self, profile_id: UUID, parcel: AnyTopologyParcel) -> ParcelCreationResponse:
+        """
+        Create Topology Parcel for selected profile_id based on payload type
+        """
+        return self.endpoint.create_any_parcel(profile_id, parcel._get_parcel_type(), parcel)
+
+    def delete_parcel(self, profile_id: UUID, parcel_type: Type[AnyTopologyParcel], parcel_id: UUID) -> None:
+        """
+        Delete Policy Object for selected profile_id based on payload type
+        """
+        return self.endpoint.delete_any_parcel(
+            profile_id=profile_id, parcel_type=parcel_type._get_parcel_type(), parcel_id=parcel_id
+        )

catalystwan/endpoints/configuration/feature_profile/sdwan/topology.py

@@ -0,0 +1,88 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+
+# mypy: disable-error-code="empty-body"
+
+from uuid import UUID
+
+from catalystwan.endpoints import JSON, APIEndpoints, delete, get, post, put, versions
+from catalystwan.models.configuration.feature_profile.common import (
+    FeatureProfileCreationPayload,
+    FeatureProfileCreationResponse,
+    FeatureProfileDetail,
+    FeatureProfileEditPayload,
+    FeatureProfileInfo,
+    GetFeatureProfilesPayload,
+    SchemaTypeQuery,
+)
+from catalystwan.models.configuration.feature_profile.parcel import ParcelCreationResponse
+from catalystwan.models.configuration.feature_profile.sdwan.topology import AnyTopologyParcel
+from catalystwan.typed_list import DataSequence
+
+
+class TopologyFeatureProfile(APIEndpoints):
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @post("/v1/feature-profile/sdwan/topology")
+    def create_topology_feature_profile(self, payload: FeatureProfileCreationPayload) -> FeatureProfileCreationResponse:
+        ...
+
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @get("/v1/feature-profile/sdwan/topology")
+    def get_topology_feature_profiles(self, params: GetFeatureProfilesPayload) -> DataSequence[FeatureProfileInfo]:
+        ...
+
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @get("/v1/feature-profile/sdwan/topology/{profile_id}")
+    def get_topology_feature_profile(self, profile_id: str, params: GetFeatureProfilesPayload) -> FeatureProfileDetail:
+        ...
+
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @put("/v1/feature-profile/sdwan/topology/{profile_id}")
+    def edit_topology_feature_profile(
+        self, profile_id: str, payload: FeatureProfileEditPayload
+    ) -> FeatureProfileCreationResponse:
+        ...
+
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @delete("/v1/feature-profile/sdwan/topology/{profile_id}")
+    def delete_topology_feature_profile(self, profile_id: str) -> None:
+        ...
+
+    #
+    # Create/Delete Any Topology Parcel
+    #
+
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @post("/v1/feature-profile/sdwan/topology/{profile_id}/{parcel_type}")
+    def create_any_parcel(
+        self, profile_id: UUID, parcel_type: str, payload: AnyTopologyParcel
+    ) -> ParcelCreationResponse:
+        ...
+
+    @versions(supported_versions=(">=20.9"), raises=False)
+    @delete("/v1/feature-profile/sdwan/topology/{profile_id}/{parcel_type}/{parcel_id}")
+    def delete_any_parcel(self, profile_id: UUID, parcel_type: str, parcel_id: UUID) -> None:
+        ...
+
+    #
+    # Mesh Parcel
+    #
+    @versions(supported_versions=(">=20.9"), raises=False)
+    @get("/v1/feature-profile/sdwan/topology/mesh/schema", resp_json_key="request")
+    def get_mesh_parcel_schema(self, params: SchemaTypeQuery) -> JSON:
+        ...
+
+    #
+    # Hub and Spoke Parcel
+    #
+    @versions(supported_versions=(">=20.9"), raises=False)
+    @get("/v1/feature-profile/sdwan/topology/hubspoke/schema", resp_json_key="request")
+    def get_hubspoke_parcel_schema(self, params: SchemaTypeQuery) -> JSON:
+        ...
+
+    #
+    # Custom Control Parcel
+    #
+    @versions(supported_versions=(">=20.9"), raises=False)
+    @get("/v1/feature-profile/sdwan/topology/custom-control/schema", resp_json_key="request")
+    def get_custom_control_parcel_schema(self, params: SchemaTypeQuery) -> JSON:
+        ...

catalystwan/endpoints/endpoints_container.py

@@ -12,6 +12,7 @@
 from catalystwan.endpoints.configuration.device.software_update import ConfigurationDeviceSoftwareUpdate
 from catalystwan.endpoints.configuration.disaster_recovery import ConfigurationDisasterRecovery
 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.policy.definition.access_control_list import ConfigurationPolicyAclDefinition
 from catalystwan.endpoints.configuration.policy.definition.access_control_list_ipv6 import (
@@ -163,6 +164,7 @@ class ConfigurationSDWANFeatureProfileContainer:
     def __init__(self, session: ManagerSession):
         self.transport = TransportFeatureProfile(client=session)
         self.system = SystemFeatureProfile(client=session)
+        self.topology = TopologyFeatureProfile(client=session)
 
 
 class ConfigurationFeatureProfileContainer:

catalystwan/integration_tests/feature_profile/sdwan/topology/test_topology_api.py

@@ -0,0 +1,50 @@
+import os
+import unittest
+from typing import cast
+from uuid import UUID
+
+from catalystwan.api.feature_profile_api import TopologyFeatureProfileAPI
+from catalystwan.models.configuration.feature_profile.sdwan.topology.hubspoke import HubSpokeParcel
+from catalystwan.models.configuration.feature_profile.sdwan.topology.mesh import MeshParcel
+from catalystwan.session import ManagerSession, create_manager_session
+
+
+class TestTopologyFeatureProfileModels(unittest.TestCase):
+    session: ManagerSession
+    api: TopologyFeatureProfileAPI
+    profile_id: UUID
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.session = create_manager_session(
+            url=cast(str, os.environ.get("TEST_VMANAGE_URL")),
+            port=cast(int, int(os.environ.get("TEST_VMANAGE_PORT"))),  # type: ignore
+            username=cast(str, os.environ.get("TEST_VMANAGE_USERNAME")),
+            password=cast(str, os.environ.get("TEST_VMANAGE_PASSWORD")),
+        )
+        cls.api = cls.session.api.sdwan_feature_profiles.topology
+        cls.profile_id = cls.api.create_profile("TestProfile", "Description").id
+
+    # TODO: need service parcel vpn api implemented to create referenced VPN-1 as precondition
+    def test_mesh(self):
+        mesh = MeshParcel(parcel_name="MeshParcel-1")
+        mesh.add_target_vpn("VPN-1")
+        mesh.add_site("SITE-1")
+        mesh_id = self.api.create_parcel(self.profile_id, mesh)
+        self.api.delete_parcel(self.profile_id, MeshParcel, mesh_id)
+
+    # TODO: need service parcel vpn api implemented to create referenced VPN-1 as precondition
+    def test_hubspoke(self):
+        hubspoke = HubSpokeParcel(parcel_name="HubSpokeParcel-1")
+        spoke = hubspoke.add_spoke(name="Spoke-1", spoke_sites=["SITE-1"])
+        spoke.add_spoke_site("SITE-2")
+        spoke.add_hub_site(["SITE-3"], preference=100891)
+        hubspoke.add_target_vpn("VPN-1")
+        hubspoke.add_selected_hub("HUB-1")
+        hubspoke_id = self.api.create_parcel(self.profile_id, hubspoke)
+        self.api.delete_parcel(self.profile_id, HubSpokeParcel, hubspoke_id)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.api.delete_profile(cls.profile_id)
+        cls.session.close()

catalystwan/models/configuration/feature_profile/sdwan/topology/__init__.py

@@ -0,0 +1,24 @@
+from typing import List, Union
+
+from pydantic import Field
+from typing_extensions import Annotated
+
+from catalystwan.models.configuration.feature_profile.sdwan.topology.hubspoke import HubSpokeParcel
+from catalystwan.models.configuration.feature_profile.sdwan.topology.mesh import MeshParcel
+
+AnyTopologyParcel = Annotated[
+    Union[
+        MeshParcel,
+        HubSpokeParcel,
+    ],
+    Field(discriminator="type_"),
+]
+
+__all__ = [
+    "AnyTopologyParcel",
+    "HubSpokeParcel" "MeshParcel",
+]
+
+
+def __dir__() -> "List[str]":
+    return list(__all__)

catalystwan/models/configuration/feature_profile/sdwan/topology/hubspoke.py

@@ -0,0 +1,59 @@
+from typing import List, Literal, Optional
+
+from pydantic import AliasPath, BaseModel, ConfigDict, Field
+
+from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global
+
+
+class Target(BaseModel):
+    vpn: Global[List[str]] = as_global([])
+
+
+class HubSite(BaseModel):
+    sites: Global[List[str]]
+    preference: Global[int]
+
+
+class Spoke(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    name: Global[str]
+    spoke_sites: Global[List[str]] = Field(
+        default=as_global([]), validation_alias="spokeSites", serialization_alias="spokeSites"
+    )
+    hub_sites: Optional[List[HubSite]] = Field(
+        default=None, validation_alias="hubSites", serialization_alias="hubSites"
+    )
+
+    @staticmethod
+    def create(name: str, spoke_sites: List[str]) -> "Spoke":
+        return Spoke(name=as_global(name), spoke_sites=Global[List[str]](value=spoke_sites))
+
+    def add_hub_site(self, sites: List[str], preference: int) -> HubSite:
+        hub_site = HubSite(sites=Global[List[str]](value=sites), preference=as_global(preference))
+        if self.hub_sites is None:
+            self.hub_sites = [hub_site]
+        else:
+            self.hub_sites.append(hub_site)
+        return hub_site
+
+    def add_spoke_site(self, site: str):
+        self.spoke_sites.value.append(site)
+
+
+class HubSpokeParcel(_ParcelBase):
+    model_config = ConfigDict(populate_by_name=True)
+    type_: Literal["hubspoke"] = Field(default="hubspoke", exclude=True)
+    target: Target = Field(default=Target(), validation_alias=AliasPath("data", "target"))
+    selected_hubs: Global[List[str]] = Field(default=as_global([]), validation_alias=AliasPath("data", "selectedHubs"))
+    spokes: List[Spoke] = Field(default=[], validation_alias=AliasPath("data", "spokes"))
+
+    def add_spoke(self, name: str, spoke_sites: List[str]) -> Spoke:
+        spoke = Spoke.create(name=name, spoke_sites=spoke_sites)
+        self.spokes.append(spoke)
+        return spoke
+
+    def add_target_vpn(self, vpn: str) -> None:
+        self.target.vpn.value.append(vpn)
+
+    def add_selected_hub(self, hub: str) -> None:
+        self.selected_hubs.value.append(hub)

catalystwan/models/configuration/feature_profile/sdwan/topology/mesh.py

@@ -0,0 +1,22 @@
+from typing import List, Literal
+
+from pydantic import AliasPath, BaseModel, ConfigDict, Field
+
+from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global
+
+
+class Target(BaseModel):
+    vpn: Global[List[str]] = as_global([])
+
+
+class MeshParcel(_ParcelBase):
+    model_config = ConfigDict(populate_by_name=True)
+    type_: Literal["mesh"] = Field(default="mesh", exclude=True)
+    target: Target = Field(default=Target(), validation_alias=AliasPath("data", "target"), description="Target Vpn")
+    sites: Global[List[str]] = Field(default=as_global([]), validation_alias=AliasPath("data", "sites"))
+
+    def add_target_vpn(self, vpn: str) -> None:
+        self.target.vpn.value.append(vpn)
+
+    def add_site(self, site: str) -> None:
+        self.sites.value.append(site)