dev: vpnQosMap definition support

catalystwan/api/policy_api.py

@@ -54,6 +54,7 @@
 from catalystwan.endpoints.configuration.policy.definition.vpn_membership import (
     ConfigurationPolicyVPNMembershipGroupDefinition,
 )
+from catalystwan.endpoints.configuration.policy.definition.vpn_qos_map import ConfigurationPolicyVPNQoSMapDefinition
 from catalystwan.endpoints.configuration.policy.definition.zone_based_firewall import (
     ConfigurationPolicyZoneBasedFirewallDefinition,
 )
@@ -204,6 +205,7 @@
 from catalystwan.models.policy.definition.traffic_data import TrafficDataPolicy, TrafficDataPolicyGetResponse
 from catalystwan.models.policy.definition.url_filtering import UrlFilteringPolicy, UrlFilteringPolicyGetResponse
 from catalystwan.models.policy.definition.vpn_membership import VPNMembershipPolicy, VPNMembershipPolicyGetResponse
+from catalystwan.models.policy.definition.vpn_qos_map import VPNQoSMapPolicy
 from catalystwan.models.policy.definition.zone_based_firewall import ZoneBasedFWPolicy, ZoneBasedFWPolicyGetResponse
 from catalystwan.models.policy.list.app_probe import AppProbeClassListInfo
 from catalystwan.models.policy.list.class_map import ClassMapListInfo
@@ -332,6 +334,7 @@
     TrafficDataPolicy: ConfigurationPolicyDataDefinition,
     UrlFilteringPolicy: ConfigurationPolicyUrlFilteringDefinition,
     VPNMembershipPolicy: ConfigurationPolicyVPNMembershipGroupDefinition,
+    VPNQoSMapPolicy: ConfigurationPolicyVPNQoSMapDefinition,
     ZoneBasedFWPolicy: ConfigurationPolicyZoneBasedFirewallDefinition,
 }
 
@@ -932,6 +935,10 @@ def get(self, type: Type[ControlPolicy]) -> DataSequence[PolicyDefinitionInfo]:
     def get(self, type: Type[VPNMembershipPolicy]) -> DataSequence[PolicyDefinitionInfo]:
         ...
 
+    @overload
+    def get(self, type: Type[VPNQoSMapPolicy]) -> DataSequence[PolicyDefinitionInfo]:
+        ...
+
     @overload
     def get(self, type: Type[HubAndSpokePolicy]) -> DataSequence[PolicyDefinitionInfo]:
         ...
@@ -1053,6 +1060,10 @@ def get(self, type: Type[ControlPolicy], id: UUID) -> ControlPolicyGetResponse:
     def get(self, type: Type[VPNMembershipPolicy], id: UUID) -> VPNMembershipPolicyGetResponse:
         ...
 
+    @overload
+    def get(self, type: Type[VPNQoSMapPolicy], id: UUID) -> VPNMembershipPolicyGetResponse:
+        ...
+
     @overload
     def get(self, type: Type[HubAndSpokePolicy], id: UUID) -> HubAndSpokePolicyGetResponse:
         ...

catalystwan/endpoints/configuration/policy/definition/vpn_qos_map.py

@@ -0,0 +1,50 @@
+# 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, put
+from catalystwan.endpoints.configuration.policy.abstractions import PolicyDefinitionEndpoints
+from catalystwan.models.policy.definition.vpn_qos_map import (
+    VPNQoSMapPolicy,
+    VPNQoSMapPolicyEditPayload,
+    VPNQoSMapPolicyGetResponse,
+)
+from catalystwan.models.policy.policy_definition import (
+    PolicyDefinitionEditResponse,
+    PolicyDefinitionId,
+    PolicyDefinitionInfo,
+    PolicyDefinitionPreview,
+)
+from catalystwan.typed_list import DataSequence
+
+
+class ConfigurationPolicyVPNQoSMapDefinition(APIEndpoints, PolicyDefinitionEndpoints):
+    @post("/template/policy/definition/vpnqosmap")
+    def create_policy_definition(self, payload: VPNQoSMapPolicy) -> PolicyDefinitionId:
+        ...
+
+    @delete("/template/policy/definition/vpnqosmap/{id}")
+    def delete_policy_definition(self, id: UUID) -> None:
+        ...
+
+    @put("/template/policy/definition/vpnqosmap/{id}")
+    def edit_policy_definition(self, id: UUID, payload: VPNQoSMapPolicyEditPayload) -> PolicyDefinitionEditResponse:
+        ...
+
+    @get("/template/policy/definition/vpnqosmap", "data")
+    def get_definitions(self) -> DataSequence[PolicyDefinitionInfo]:
+        ...
+
+    @get("/template/policy/definition/vpnqosmap/{id}")
+    def get_policy_definition(self, id: UUID) -> VPNQoSMapPolicyGetResponse:
+        ...
+
+    @post("/template/policy/definition/vpnqosmap/preview")
+    def preview_policy_definition(self, payload: VPNQoSMapPolicy) -> PolicyDefinitionPreview:
+        ...
+
+    @get("/template/policy/definition/vpnqosmap/preview/{id}")
+    def preview_policy_definition_by_id(self, id: UUID) -> PolicyDefinitionPreview:
+        ...

catalystwan/endpoints/endpoints_container.py

@@ -52,6 +52,7 @@
 from catalystwan.endpoints.configuration.policy.definition.vpn_membership import (
     ConfigurationPolicyVPNMembershipGroupDefinition,
 )
+from catalystwan.endpoints.configuration.policy.definition.vpn_qos_map import ConfigurationPolicyVPNQoSMapDefinition
 from catalystwan.endpoints.configuration.policy.definition.zone_based_firewall import (
     ConfigurationPolicyZoneBasedFirewallDefinition,
 )
@@ -195,6 +196,7 @@ def __init__(self, session: ManagerSession):
         self.security_group = ConfigurationPolicySecurityGroupDefinition(session)
         self.srst_phone_profile = ConfigurationPolicySrstPhoneProfileDefinition(session)
         self.vpn_membership = ConfigurationPolicyVPNMembershipGroupDefinition(session)
+        self.vpn_qos_map = ConfigurationPolicyVPNQoSMapDefinition(session)
         self.zone_based_firewall = ConfigurationPolicyZoneBasedFirewallDefinition(session)
 
 

catalystwan/models/policy/__init__.py

@@ -16,6 +16,7 @@
     SrstPhoneProfilePolicy,
     SrstPhoneProfilePolicyGetResponse,
 )
+from catalystwan.models.policy.definition.vpn_qos_map import VPNQoSMapPolicy, VPNQoSMapPolicyGetResponse
 from catalystwan.models.policy.list.app import AppList, AppListInfo
 from catalystwan.models.policy.list.app_probe import AppProbeClassList, AppProbeClassListInfo
 from catalystwan.models.policy.list.as_path import ASPathList, ASPathListInfo
@@ -136,6 +137,7 @@
         TrafficDataPolicy,
         UrlFilteringPolicy,
         VPNMembershipPolicy,
+        VPNQoSMapPolicy,
         ZoneBasedFWPolicy,
     ],
     Field(discriminator="type"),
@@ -266,6 +268,7 @@
         TrafficDataPolicyGetResponse,
         UrlFilteringPolicyGetResponse,
         VPNMembershipPolicyGetResponse,
+        VPNQoSMapPolicyGetResponse,
         ZoneBasedFWPolicyGetResponse,
     ],
     Field(discriminator="type"),
@@ -346,6 +349,7 @@
     "UrlFilteringPolicy",
     "VPNList",
     "VPNMembershipPolicy",
+    "VPNQoSMapPolicy",
     "ZoneBasedFWPolicy",
     "ZoneList",
 )

catalystwan/models/policy/definition/vpn_qos_map.py

@@ -0,0 +1,46 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+
+from typing import List, Literal, Optional
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from catalystwan.models.common import IntStr
+from catalystwan.models.policy.policy_definition import (
+    PolicyDefinitionBase,
+    PolicyDefinitionGetResponse,
+    PolicyDefinitionId,
+)
+
+
+class VPNQoSScheduler(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    child_map_ref: UUID = Field(validation_alias="childMapRef", serialization_alias="childMapRef")
+    vpn_list_ref: Optional[UUID] = Field(default=None, validation_alias="vpnListRef", serialization_alias="vpnListRef")
+    bandwidth_rate: Optional[IntStr] = Field(
+        default=None, validation_alias="bandwidthRate", serialization_alias="bandwidthRate", ge=8, le=100_000_000
+    )
+    shaping_rate: Optional[IntStr] = Field(
+        default=None, validation_alias="shapingRate", serialization_alias="shapingRate", ge=8, le=100_000_000
+    )
+
+
+class VPNQoSMapPolicyDefinition(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    vpn_qos_schedulers: List[VPNQoSScheduler] = Field(
+        validation_alias="vpnQosSchedulers", serialization_alias="vpnQosSchedulers"
+    )
+
+
+class VPNQoSMapPolicy(PolicyDefinitionBase):
+    model_config = ConfigDict(populate_by_name=True)
+    type: Literal["vpnQosMap", "vpnQoSMap"] = "vpnQosMap"
+    definition: VPNQoSMapPolicyDefinition
+
+
+class VPNQoSMapPolicyEditPayload(VPNQoSMapPolicy, PolicyDefinitionId):
+    pass
+
+
+class VPNQoSMapPolicyGetResponse(VPNQoSMapPolicy, PolicyDefinitionGetResponse):
+    pass

catalystwan/models/policy/localized.py

@@ -10,6 +10,7 @@
 LocalizedPolicySupportedItemType = Literal[
     "qosMap",
     "rewriteRule",
+    "vpnQosMap",
     "vpnQoSMap",
     "acl",
     "aclv6",
@@ -101,7 +102,7 @@ def add_rewrite_rule(self, definition_id: UUID) -> None:
         self._add_item("rewriteRule", definition_id)
 
     def add_vpn_qos_map(self, definition_id: UUID) -> None:
-        self._add_item("vpnQoSMap", definition_id)
+        self._add_item("vpnQosMap", definition_id)
 
     def add_access_control_list(self, definition_id: UUID) -> None:
         self._add_item("acl", definition_id)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "catalystwan"
-version = "0.40.0dev4"
+version = "0.40.0dev5"
 description = "Cisco Catalyst WAN SDK for Python"
 authors = ["kagorski <kagorski@cisco.com>"]
 readme = "README.md"