add policy-list to policy-object-parcel converters (#516)

* collect security policies, fix collecting localized, centralized, lists so they include both id and definitions

* converter draft

* convert classmap, color

* convert standard community

* implement missing items

catalystwan/api/feature_profile_api.py

@@ -41,7 +41,7 @@
     IPv6DataPrefixParcel,
     IPv6PrefixListParcel,
     LocalDomainParcel,
-    PolicierParcel,
+    PolicerParcel,
     PreferredColorGroupParcel,
     PrefixListParcel,
     ProtocolListParcel,
@@ -655,7 +655,7 @@ def get(self, profile_id: UUID, parcel_type: Type[LocalDomainParcel]) -> DataSeq
         ...
 
     @overload
-    def get(self, profile_id: UUID, parcel_type: Type[PolicierParcel]) -> DataSequence[Parcel[Any]]:
+    def get(self, profile_id: UUID, parcel_type: Type[PolicerParcel]) -> DataSequence[Parcel[Any]]:
         ...
 
     @overload
@@ -767,7 +767,7 @@ def get(self, profile_id: UUID, parcel_type: Type[LocalDomainParcel], parcel_id:
         ...
 
     @overload
-    def get(self, profile_id: UUID, parcel_type: Type[PolicierParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
+    def get(self, profile_id: UUID, parcel_type: Type[PolicerParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
         ...
 
     @overload
@@ -915,7 +915,7 @@ def delete(self, profile_id: UUID, parcel_type: Type[LocalDomainParcel], list_ob
         ...
 
     @overload
-    def delete(self, profile_id: UUID, parcel_type: Type[PolicierParcel], list_object_id: UUID) -> None:
+    def delete(self, profile_id: UUID, parcel_type: Type[PolicerParcel], list_object_id: UUID) -> None:
         ...
 
     @overload

catalystwan/models/common.py

@@ -1,6 +1,6 @@
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
 
-from typing import Dict, List, Literal, Sequence, Set, Tuple, Union
+from typing import Dict, List, Literal, Optional, Sequence, Set, Tuple, Union
 from uuid import UUID
 
 from pydantic import PlainSerializer
@@ -59,6 +59,35 @@ def check_any_of_exclusive_field_sets(values: Dict, field_sets: List[Tuple[Set[s
     BeforeValidator(lambda x: int(x)),
 ]
 
+IntRange = Tuple[int, Optional[int]]
+
+
+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'"""
+    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"
+        first = int_list[0]
+        second = None if len(int_list) == 1 else int_list[1]
+        int_range = (first, second)
+    else:
+        int_range = value
+    if ascending and int_range[1] is not None:
+        assert int_range[0] < int_range[1], "Numbers in range must be in ascending order"
+    return int_range
+
+
+def int_range_serializer(value: IntRange) -> str:
+    """Serializes integer pair as string separated by hyphen eg: '1-3' or single number '1'"""
+    return "-".join((str(i) for i in value if i is not None))
+
+
+IntRangeStr = Annotated[
+    IntRange,
+    PlainSerializer(int_range_serializer, return_type=str, when_used="json-unless-none"),
+    BeforeValidator(int_range_str_validator),
+]
+
 
 def str_as_uuid_list(val: Union[str, Sequence[UUID]]) -> Sequence[UUID]:
     if isinstance(val, str):

catalystwan/models/configuration/config_migration.py

@@ -12,13 +12,10 @@
 from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel
 from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel
 from catalystwan.models.configuration.topology_group import TopologyGroup
-from catalystwan.models.policy import (
-    AnyPolicyDefinition,
-    AnyPolicyList,
-    CentralizedPolicy,
-    LocalizedPolicy,
-    SecurityPolicy,
-)
+from catalystwan.models.policy import AnyPolicyDefinition, AnyPolicyList
+from catalystwan.models.policy.centralized import CentralizedPolicyInfo
+from catalystwan.models.policy.localized import LocalizedPolicyInfo
+from catalystwan.models.policy.security import AnySecurityPolicyInfo
 
 AnyParcel = Annotated[
     Union[
@@ -31,13 +28,13 @@
 
 class UX1Policies(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
-    centralized_policies: List[CentralizedPolicy] = Field(
+    centralized_policies: List[CentralizedPolicyInfo] = Field(
         default=[], serialization_alias="centralizedPolicies", validation_alias="centralizedPolicies"
     )
-    localized_policies: List[LocalizedPolicy] = Field(
+    localized_policies: List[LocalizedPolicyInfo] = Field(
         default=[], serialization_alias="localizedPolicies", validation_alias="localizedPolicies"
     )
-    security_policies: List[SecurityPolicy] = Field(
+    security_policies: List[AnySecurityPolicyInfo] = Field(
         default=[], serialization_alias="securityPolicies", validation_alias="securityPolicies"
     )
     policy_definitions: List[AnyPolicyDefinition] = Field(

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

@@ -13,7 +13,7 @@
 from .policy.fowarding_class import FowardingClassParcel, FowardingClassQueueEntry
 from .policy.ipv6_data_prefix import IPv6DataPrefixEntry, IPv6DataPrefixParcel
 from .policy.ipv6_prefix_list import IPv6PrefixListEntry, IPv6PrefixListParcel
-from .policy.policier import PolicierEntry, PolicierParcel
+from .policy.policer import PolicerEntry, PolicerParcel
 from .policy.prefered_group_color import Preference, PreferredColorGroupEntry, PreferredColorGroupParcel
 from .policy.prefix_list import PrefixListEntry, PrefixListParcel
 from .policy.sla_class import FallbackBestTunnel, SLAAppProbeClass, SLAClassCriteria, SLAClassListEntry, SLAClassParcel
@@ -44,7 +44,7 @@
 
 AnyPolicyObjectParcel = Annotated[
     Union[
-        # AnyURLParcel,
+        AnyURLParcel,
         ApplicationListParcel,
         AppProbeParcel,
         ColorParcel,
@@ -57,7 +57,7 @@
         IPv6DataPrefixParcel,
         IPv6PrefixListParcel,
         LocalDomainParcel,
-        PolicierParcel,
+        PolicerParcel,
         PreferredColorGroupParcel,
         PrefixListParcel,
         ProtocolListParcel,
@@ -101,8 +101,8 @@
     "IPv6PrefixListParcel",
     "LocalDomainListEntry",
     "LocalDomainParcel",
-    "PolicierEntry",
-    "PolicierParcel",
+    "PolicerEntry",
+    "PolicerParcel",
     "Preference",
     "PreferredColorGroupEntry",
     "PreferredColorGroupParcel",

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/expanded_community_list.py

@@ -10,17 +10,17 @@
 class ExpandedCommunityParcel(_ParcelBase):
     type_: Literal["expanded-community"] = Field(default="expanded-community", exclude=True)
     model_config = ConfigDict(populate_by_name=True)
-    expandedCommunityList: Global[list] = Field(
+    expanded_community_list: Global[list] = Field(
         default=as_global([]),
         serialization_alias="expandedCommunityList",
         validation_alias=AliasPath("data", "expandedCommunityList"),
     )
 
     def add_community(self, expanded_community: str):
-        self.expandedCommunityList.value.append(expanded_community)
+        self.expanded_community_list.value.append(expanded_community)
 
-    @field_validator("expandedCommunityList")
+    @field_validator("expanded_community_list")
     @classmethod
-    def check_rate(cls, expanded_community_list: Global):
+    def check_list_str(cls, expanded_community_list: Global):
         assert all([isinstance(ec, str) for ec in expanded_community_list.value])
         return expanded_community_list

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_data_prefix.py

@@ -1,6 +1,6 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
-from ipaddress import IPv6Address, IPv6Network
+from ipaddress import IPv6Address, IPv6Interface
 from typing import List, Literal
 
 from pydantic import AliasPath, BaseModel, ConfigDict, Field
@@ -18,10 +18,10 @@ class IPv6DataPrefixParcel(_ParcelBase):
     type_: Literal["data-ipv6-prefix"] = Field(default="data-ipv6-prefix", exclude=True)
     entries: List[IPv6DataPrefixEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
-    def add_prefix(self, ipv6_network: IPv6Network):
+    def add_prefix(self, ipv6_network: IPv6Interface):
         self.entries.append(
             IPv6DataPrefixEntry(
-                ipv6_address=as_global(ipv6_network.network_address),
-                ipv6_prefix_length=as_global(ipv6_network.prefixlen),
+                ipv6_address=as_global(ipv6_network.network.network_address),
+                ipv6_prefix_length=as_global(ipv6_network.network.prefixlen),
             )
         )

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_prefix_list.py

@@ -1,7 +1,7 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
-from ipaddress import IPv6Address, IPv6Network
-from typing import List, Literal
+from ipaddress import IPv6Address, IPv6Interface
+from typing import List, Literal, Optional
 
 from pydantic import AliasPath, BaseModel, ConfigDict, Field
 
@@ -11,17 +11,27 @@
 class IPv6PrefixListEntry(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
     ipv6_address: Global[IPv6Address] = Field(serialization_alias="ipv6Address", validation_alias="ipv6Address")
-    ipv6_prefix_length: Global[int] = Field(serialization_alias="ipv6PrefixLength", validation_alias="ipv6PrefixLength")
+    ipv6_prefix_length: Global[int] = Field(
+        serialization_alias="ipv6PrefixLength", validation_alias="ipv6PrefixLength", ge=0, le=128
+    )
+    le_range_prefix_length: Optional[Global[int]] = Field(
+        serialization_alias="leRangePrefixLength", validation_alias="leRangePrefixLength"
+    )
+    ge_range_prefix_length: Optional[Global[int]] = Field(
+        serialization_alias="geRangePrefixLength", validation_alias="geRangePrefixLength"
+    )
 
 
 class IPv6PrefixListParcel(_ParcelBase):
     type_: Literal["ipv6-prefix"] = Field(default="ipv6-prefix", exclude=True)
     entries: List[IPv6PrefixListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
-    def add_prefix(self, ipv6_network: IPv6Network):
+    def add_prefix(self, ipv6_network: IPv6Interface, ge: Optional[int] = None, le: Optional[int] = None):
         self.entries.append(
             IPv6PrefixListEntry(
-                ipv6_address=as_global(ipv6_network.network_address),
-                ipv6_prefix_length=as_global(ipv6_network.prefixlen),
+                ipv6_address=as_global(ipv6_network.network.network_address),
+                ipv6_prefix_length=as_global(ipv6_network.network.prefixlen),
+                le_range_prefix_length=as_global(le) if le is not None else None,
+                ge_range_prefix_length=as_global(ge) if ge is not None else None,
             )
         )

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policier.py → catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policer.py

@@ -12,7 +12,7 @@
 ]
 
 
-class PolicierEntry(BaseModel):
+class PolicerEntry(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
     burst: Global[int]
     exceed: Global[PolicerExceedAction]
@@ -31,13 +31,13 @@ def check_rate(cls, rate_str: Global):
         return rate_str
 
 
-class PolicierParcel(_ParcelBase):
+class PolicerParcel(_ParcelBase):
     type_: Literal["policer"] = Field(default="policer", exclude=True)
-    entries: List[PolicierEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
+    entries: List[PolicerEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
     def add_entry(self, burst: int, exceed: PolicerExceedAction, rate: int):
         self.entries.append(
-            PolicierEntry(
+            PolicerEntry(
                 burst=as_global(burst),
                 exceed=as_global(exceed, PolicerExceedAction),
                 rate=as_global(rate),

catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/standard_community.py

@@ -10,7 +10,7 @@
 
 class StandardCommunityEntry(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
-    standard_community: Global[WellKnownBGPCommunities] = Field(
+    standard_community: Global[str] = Field(
         serialization_alias="standardCommunity", validation_alias="standardCommunity"
     )
 
@@ -19,7 +19,11 @@ class StandardCommunityParcel(_ParcelBase):
     type_: Literal["standard-community"] = Field(default="standard-community", exclude=True)
     entries: List[StandardCommunityEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
-    def add_community(self, standard_community: WellKnownBGPCommunities):
-        self.entries.append(
-            StandardCommunityEntry(standard_community=as_global(standard_community, WellKnownBGPCommunities))
-        )
+    def _add_community(self, standard_community: str):
+        self.entries.append(StandardCommunityEntry(standard_community=as_global(standard_community)))
+
+    def add_well_known_community(self, standard_community: WellKnownBGPCommunities):
+        self._add_community(standard_community)
+
+    def add_community(self, as_number: int, community_number: int) -> None:
+        self._add_community(f"{as_number}:{community_number}")

catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/security_port.py

@@ -32,5 +32,11 @@ class SecurityPortParcel(_ParcelBase):
     type_: Literal["security-port"] = Field(default="security-port", exclude=True)
     entries: List[SecurityPortListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
-    def add_port(self, port: str):
+    def _add_port(self, port: str):
         self.entries.append(SecurityPortListEntry(port=as_global(port)))
+
+    def add_port(self, port: int):
+        self._add_port(str(port))
+
+    def add_port_range(self, start_port: int, end_port: int):
+        self._add_port(f"{start_port}-{end_port}")

catalystwan/models/policy/centralized.py

@@ -228,7 +228,7 @@ def try_parse_policy_definition_string(cls, values):
         # while POST /template/policy/vsmart requires a regular object
         # it makes sense to reuse that model for both requests and present parsed data to the user
         # TODO: this is workaround, probably it is better to provide separate models for "cli" and "feature"
-        if policy_definition := values.get("policyDefinition") and values.get("policyType") != "cli":
+        if (policy_definition := values.get("policyDefinition")) and values.get("policyType") != "cli":
             if isinstance(policy_definition, str):
                 values["policyDefinition"] = CentralizedPolicyDefinition.model_validate_json(policy_definition)
         else: