diff --git a/catalystwan/api/feature_profile_api.py b/catalystwan/api/feature_profile_api.py index aa08e294..683113c5 100644 --- a/catalystwan/api/feature_profile_api.py +++ b/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 + ) diff --git a/catalystwan/endpoints/configuration/feature_profile/sdwan/topology.py b/catalystwan/endpoints/configuration/feature_profile/sdwan/topology.py new file mode 100644 index 00000000..90268603 --- /dev/null +++ b/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: + ... diff --git a/catalystwan/endpoints/endpoints_container.py b/catalystwan/endpoints/endpoints_container.py index 2b634e70..60dbbcbc 100644 --- a/catalystwan/endpoints/endpoints_container.py +++ b/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: diff --git a/catalystwan/integration_tests/feature_profile/sdwan/topology/test_topology_api.py b/catalystwan/integration_tests/feature_profile/sdwan/topology/test_topology_api.py new file mode 100644 index 00000000..59a6837e --- /dev/null +++ b/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() diff --git a/catalystwan/models/configuration/feature_profile/sdwan/topology/__init__.py b/catalystwan/models/configuration/feature_profile/sdwan/topology/__init__.py new file mode 100644 index 00000000..8da89fce --- /dev/null +++ b/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__) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/topology/hubspoke.py b/catalystwan/models/configuration/feature_profile/sdwan/topology/hubspoke.py new file mode 100644 index 00000000..14a8e5ce --- /dev/null +++ b/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) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/topology/mesh.py b/catalystwan/models/configuration/feature_profile/sdwan/topology/mesh.py new file mode 100644 index 00000000..d151aae4 --- /dev/null +++ b/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)