AAR policy, include serviceArea and trafficCategory entry (#851)

catalystwan/models/common.py

@@ -3,7 +3,7 @@
 import re
 from dataclasses import InitVar, dataclass, field
 from ipaddress import IPv4Interface, IPv6Interface
-from typing import Any, Dict, Iterator, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union, get_args
+from typing import Any, Dict, Iterator, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union, cast, get_args
 from uuid import UUID
 
 from annotated_types import Ge, Le
@@ -94,8 +94,9 @@ def dump(
                 if current_field_path := replaced_keys.get(current_field_name):
                     path, name = current_field_path
                     dict_ = model_dict[path] if path is not None else model_dict
-                    dict_[new_field_name] = dict_[name]
-                    del dict_[name]
+                    if new_field_name is not None:
+                        dict_[new_field_name] = dict_[name]
+                        del dict_[name]
         return model_dict
 
 
@@ -152,7 +153,7 @@ def str_as_uuid_list(val: Union[str, Sequence[UUID]]) -> Sequence[UUID]:
 
 def str_as_positive_int_list(val: Union[str, Sequence[PositiveInt]]) -> Sequence[PositiveInt]:
     if isinstance(val, str):
-        return [PositiveInt(element) for element in val.split()]
+        return [int(element) for element in val.split()]
     return val
 
 
@@ -307,7 +308,7 @@ def int_range_serializer(value: IntRange) -> str:
 
 def str_as_interface_list(val: Union[str, Sequence[InterfaceStr]]) -> Sequence[InterfaceStr]:
     if isinstance(val, str):
-        return [InterfaceStr(element) for element in val.split()]
+        return [str(element) for element in val.split()]
     return val
 
 
@@ -1458,3 +1459,29 @@ def str_as_interface_list(val: Union[str, Sequence[InterfaceStr]]) -> Sequence[I
     "jitter-latency-loss",
     "jitter-loss-latency",
 ]
+
+ServiceAreaValue = Literal[
+    "common",
+    "exchange",
+    "sharepoint",
+    "skype",
+]
+
+TrafficCategory = Literal[
+    "all",
+    "optimize",
+    "optimize-allow",
+]
+
+
+def str_as_service_area_list(val: Union[str, Sequence[ServiceAreaValue]]) -> Sequence[ServiceAreaValue]:
+    if isinstance(val, str):
+        return [cast(ServiceAreaValue, item) for item in val.split() if item in get_args(ServiceAreaValue)]
+    return val
+
+
+SpaceSeparatedServiceAreaList = Annotated[
+    List[ServiceAreaValue],
+    PlainSerializer(lambda x: " ".join(map(str, x)), return_type=str, when_used="json-unless-none"),
+    BeforeValidator(str_as_service_area_list),
+]

catalystwan/models/configuration/feature_profile/sdwan/application_priority/traffic_policy.py

@@ -24,9 +24,11 @@
     Icmp6MsgType,
     IcmpMsgType,
     SequenceIpType,
+    ServiceAreaValue,
     ServiceChainNumber,
     ServiceType,
     TLOCColor,
+    TrafficCategory,
     TrafficTargetType,
     VersionedField,
 )
@@ -46,19 +48,6 @@ class TrafficPolicyTarget(BaseModel):
     vpn: Global[List[str]]
 
 
-ServiceAreaValue = Literal[
-    "common",
-    "exchange",
-    "sharepoint",
-    "skype",
-]
-
-TrafficCategory = Literal[
-    "all",
-    "optimize",
-    "optimize-allow",
-]
-
 TrafficClass = Literal[
     "bronze",
     "gold-broadcast-video",

catalystwan/models/policy/definition/app_route.py

@@ -5,7 +5,15 @@
 from pydantic import ConfigDict, Field
 from typing_extensions import Annotated
 
-from catalystwan.models.common import DestinationRegion, DNSEntryType, SequenceIpType, TLOCColor, TrafficTargetType
+from catalystwan.models.common import (
+    DestinationRegion,
+    DNSEntryType,
+    SequenceIpType,
+    ServiceAreaValue,
+    TLOCColor,
+    TrafficCategory,
+    TrafficTargetType,
+)
 from catalystwan.models.policy.policy_definition import (
     AppListEntry,
     BackupSlaPrefferedColorAction,
@@ -32,13 +40,15 @@
     PolicyDefinitionSequenceBase,
     ProtocolEntry,
     SaaSAppListEntry,
+    ServiceAreaEntry,
     SlaClassAction,
     SlaNotMetAction,
     SourceDataIPv6PrefixListEntry,
     SourceDataPrefixListEntry,
     SourceIPEntry,
     SourceIPv6Entry,
     SourcePortEntry,
+    TrafficCategoryEntry,
     TrafficToEntry,
 )
 
@@ -57,11 +67,13 @@
         PLPEntry,
         ProtocolEntry,
         SaaSAppListEntry,
+        ServiceAreaEntry,
         SourceDataIPv6PrefixListEntry,
         SourceDataPrefixListEntry,
         SourceIPEntry,
         SourceIPv6Entry,
         SourcePortEntry,
+        TrafficCategoryEntry,
         TrafficToEntry,
     ],
     Field(discriminator="field"),
@@ -127,6 +139,9 @@ def match_protocols(self, protocols: Set[int]) -> None:
     def match_saas_app_list(self, saas_app_list_id: UUID) -> None:
         self._insert_match(SaaSAppListEntry(ref=saas_app_list_id))
 
+    def match_service_areas(self, service_areas: Set[ServiceAreaValue]) -> None:
+        self._insert_match(ServiceAreaEntry(value=list(service_areas)))
+
     def match_source_data_prefix_list(self, data_prefix_list_id: UUID) -> None:
         self._insert_match(SourceDataPrefixListEntry(ref=[data_prefix_list_id]))
 
@@ -139,6 +154,9 @@ def match_source_ipv6(self, networks: List[IPv6Network]) -> None:
     def match_source_port(self, ports: Set[int] = set(), port_ranges: List[Tuple[int, int]] = []) -> None:
         self._insert_match(SourcePortEntry.from_port_set_and_ranges(ports, port_ranges))
 
+    def match_traffic_category(self, traffic_category: TrafficCategory) -> None:
+        self._insert_match(TrafficCategoryEntry(value=traffic_category))
+
     def match_traffic_to(self, traffic_to: TrafficTargetType) -> None:
         self._insert_match(TrafficToEntry(value=traffic_to))
 

catalystwan/models/policy/policy_definition.py

@@ -31,10 +31,12 @@
     SpaceSeparatedIPv4,
     SpaceSeparatedIPv6,
     SpaceSeparatedNonNegativeIntList,
+    SpaceSeparatedServiceAreaList,
     SpaceSeparatedTLOCColorStr,
     SpaceSeparatedUUIDList,
     TLOCActionType,
     TLOCColor,
+    TrafficCategory,
     TrafficTargetType,
     check_fields_exclusive,
     str_as_str_list,
@@ -949,6 +951,16 @@ class AffinityEntry(BaseModel):
 RedirectDNSActionEntry = Union[IPAddressEntry, DNSTypeEntry]
 
 
+class ServiceAreaEntry(BaseModel):
+    field: Literal["serviceArea"] = "serviceArea"
+    value: SpaceSeparatedServiceAreaList
+
+
+class TrafficCategoryEntry(BaseModel):
+    field: Literal["trafficCategory"] = "trafficCategory"
+    value: TrafficCategory
+
+
 class LogAction(BaseModel):
     type: Literal["log"] = "log"
     parameter: str = ""
@@ -1299,6 +1311,7 @@ class ActionSet(BaseModel):
         RoleEntry,
         RuleSetListEntry,
         SaaSAppListEntry,
+        ServiceAreaEntry,
         SiteEntry,
         SiteListEntry,
         SiteListEntry,
@@ -1317,6 +1330,7 @@ class ActionSet(BaseModel):
         TCPEntry,
         TLOCEntry,
         TLOCListEntry,
+        TrafficCategoryEntry,
         TrafficClassEntry,
         TrafficToEntry,
         VPNEntry,

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "catalystwan"
-version = "0.36.0dev1"
+version = "0.36.0dev2"
 description = "Cisco Catalyst WAN SDK for Python"
 authors = ["kagorski <[email protected]>"]
 readme = "README.md"