[dev]: update route policy (#170)

* draft

* feat: add community default_factory

* feat: add IntStrAsNum without PlainSerializer

---------

Co-authored-by: Wojciech Czuma <czuma.wojciech@gmail.com>

Author: sbasan

catalystwan/models/common.py

@@ -187,6 +187,11 @@ def str_as_str_list(val: Union[str, Sequence[str]]) -> Sequence[str]:
     BeforeValidator(lambda x: int(x)),
 ]
 
+IntStrAsNum = Annotated[
+    int,
+    BeforeValidator(lambda x: int(x)),
+]
+
 IntRange = Tuple[int, Optional[int]]
 
 DualToneCadenceInterval = Annotated[int, Field(ge=50, le=10_000)]
@@ -203,6 +208,17 @@ def str_as_cadence_pairs(val: Union[str, Sequence[DualToneCadenceStartStop]]) ->
     return val
 
 
+AsNum = Annotated[IntStrAsNum, Field(ge=1, le=4294967295)]
+AsRegExp = Annotated[str, Field(pattern=r"^([0-9]+|[0-9]*\.[0-9]+)$")]
+AsPrepend = Union[AsNum, AsRegExp]
+
+
+def str_as_as_prepend_list(val: Any) -> Any:
+    if isinstance(val, str):
+        return [item.strip() for item in val.split() if item.strip()]
+    return val
+
+
 def tuple_to_str(val: Tuple[Any, ...]) -> str:
     return " ".join(map(str, val))
 
@@ -249,6 +265,20 @@ def tuple_to_str(val: Tuple[Any, ...]) -> str:
     Field(min_length=1),
 ]
 
+SpaceSeparatedAsNumList = Annotated[
+    List[AsNum],
+    PlainSerializer(lambda x: " ".join(map(str, x)), return_type=str, when_used="json-unless-none"),
+    BeforeValidator(str_as_positive_int_list),
+    Field(min_length=1),
+]
+
+SpaceSeparatedAsPrependList = Annotated[
+    List[AsPrepend],
+    PlainSerializer(lambda x: " ".join(map(str, x)), return_type=str, when_used="json-unless-none"),
+    BeforeValidator(str_as_as_prepend_list),
+    Field(min_length=1),
+]
+
 Ipv4PrefixLen = Annotated[
     int,
     Field(ge=0, le=32),

catalystwan/models/configuration/feature_profile/sdwan/service/route_policy.py

@@ -16,7 +16,7 @@
     as_global,
     as_variable,
 )
-from catalystwan.models.common import AcceptRejectActionType
+from catalystwan.models.common import AcceptRejectActionType, AsPrepend
 from catalystwan.models.configuration.feature_profile.common import RefIdItem
 
 Community = Literal["internet", "local-AS", "no-advertise", "no-export"]
@@ -130,11 +130,11 @@ class SetAsPath(BaseModel):
     model_config = ConfigDict(
         populate_by_name=True,
     )
-    prepend: Optional[List[Global[int]]] = None
+    prepend: Optional[Global[List[AsPrepend]]] = None
 
     @classmethod
-    def from_list(cls, prepend: List[int]) -> SetAsPath:
-        return cls(prepend=[as_global(i) for i in prepend])
+    def from_list(cls, prepend: List[Union[str, int]]) -> SetAsPath:
+        return cls(prepend=Global[List[AsPrepend]](value=prepend))
 
 
 class SetCommunity(BaseModel):
@@ -146,7 +146,7 @@ class SetCommunity(BaseModel):
         populate_by_name=True,
     )
     additive: Union[Global[bool], Default[bool]] = as_default(False)
-    community: Optional[Union[Global[List[str]], Variable]] = None
+    community: Union[Global[List[str]], Variable] = Global[List[str]](value=[])
 
 
 class Accept(BaseModel):
@@ -167,8 +167,11 @@ class Accept(BaseModel):
     as_path: Optional[SetAsPath] = Field(
         default=None, serialization_alias="setAsPath", validation_alias="setAsPath", description="Set AS Path"
     )
-    community: Optional[SetCommunity] = Field(
-        default=None, serialization_alias="setCommunity", validation_alias="setCommunity", description="Set Community"
+    community: SetCommunity = Field(
+        default_factory=SetCommunity,
+        serialization_alias="setCommunity",
+        validation_alias="setCommunity",
+        description="Set Community",
     )
     local_preference: Optional[Global[int]] = Field(
         default=None,
@@ -330,7 +333,7 @@ def match_ipv6_next_hop(self, ipv6_next_hop: UUID) -> None:
     def associate_reject_action(self) -> None:
         self.actions = [(RejectActions(reject=as_default(True)))]
 
-    def associate_as_path_action(self, prepend: List[int]) -> None:
+    def associate_as_path_action(self, prepend: List[Union[str, int]]) -> None:
         self._accept_action.as_path = SetAsPath.from_list(prepend)
 
     def associate_communities_action(self, additive: bool, communities: List[str]) -> None:

catalystwan/models/policy/definition/route_policy.py

@@ -108,7 +108,7 @@ def associate_aggregator_action(self, aggregator_value: int, ip_address: Union[I
             AggregatorActionEntry(value=AggregatorActionEntryValue(aggregator=aggregator_value, ip_address=ip_address))
         )
 
-    def associate_as_path_action(self, prepend_action: List[int], exclude_action: List[int]):
+    def associate_as_path_action(self, prepend_action: List[Union[str, int]], exclude_action: List[int]):
         self._insert_action_in_set(
             AsPathActionEntry(value=AsPathActionEntryValue(prepend=prepend_action, exclude=exclude_action))
         )

catalystwan/models/policy/policy_definition.py

@@ -27,6 +27,8 @@
     SequenceIpType,
     ServiceChainNumber,
     ServiceType,
+    SpaceSeparatedAsNumList,
+    SpaceSeparatedAsPrependList,
     SpaceSeparatedInterfaceStr,
     SpaceSeparatedIPv4,
     SpaceSeparatedIPv6,
@@ -406,8 +408,8 @@ class AsPathListMatchEntry(BaseModel):
 
 
 class AsPathActionEntryValue(BaseModel):
-    prepend: Optional[SpaceSeparatedNonNegativeIntList] = None
-    exclude: Optional[SpaceSeparatedNonNegativeIntList] = None
+    prepend: Optional[SpaceSeparatedAsPrependList] = None
+    exclude: Optional[SpaceSeparatedAsNumList] = None
 
 
 class AsPathActionEntry(BaseModel):