introduce trunkgroup policy list (#28)

* introduce trunkgroup policy list

* Update trunkgroup.py

---------

Co-authored-by: radkrawczyk <[email protected]>

catalystwan/api/policy_api.py

@@ -64,6 +64,7 @@
 from catalystwan.endpoints.configuration.policy.list.site import ConfigurationPolicySiteList, SiteListInfo
 from catalystwan.endpoints.configuration.policy.list.sla import ConfigurationPolicySLAClassList, SLAClassListInfo
 from catalystwan.endpoints.configuration.policy.list.tloc import ConfigurationPolicyTLOCList, TLOCListInfo
+from catalystwan.endpoints.configuration.policy.list.trunkgroup import ConfigurationPolicyTrunkGroupList
 from catalystwan.endpoints.configuration.policy.list.url_allow_list import (
     ConfigurationPolicyURLAllowList,
     URLAllowListInfo,
@@ -141,6 +142,7 @@
 from catalystwan.models.policy.list.ips_signature import IPSSignatureListInfo
 from catalystwan.models.policy.list.ipv6_prefix import IPv6PrefixListInfo
 from catalystwan.models.policy.list.local_domain import LocalDomainListInfo
+from catalystwan.models.policy.list.trunkgroup import TrunkGroupList, TrunkGroupListInfo
 from catalystwan.models.policy.localized import (
     LocalizedPolicy,
     LocalizedPolicyDeviceInfo,
@@ -192,6 +194,7 @@
     SiteList: ConfigurationPolicySiteList,
     SLAClassList: ConfigurationPolicySLAClassList,
     TLOCList: ConfigurationPolicyTLOCList,
+    TrunkGroupList: ConfigurationPolicyTrunkGroupList,
     URLBlockList: ConfigurationPolicyURLBlockList,
     URLAllowList: ConfigurationPolicyURLAllowList,
     VPNList: ConfigurationPolicyVPNList,
@@ -451,6 +454,10 @@ def get(self, type: Type[SLAClassList]) -> DataSequence[SLAClassListInfo]:
     def get(self, type: Type[TLOCList]) -> DataSequence[TLOCListInfo]:
         ...
 
+    @overload
+    def get(self, type: Type[TrunkGroupList]) -> DataSequence[TrunkGroupListInfo]:
+        ...
+
     @overload
     def get(self, type: Type[URLBlockList]) -> DataSequence[URLBlockListInfo]:
         ...
@@ -569,6 +576,10 @@ def get(self, type: Type[SLAClassList], id: UUID) -> SLAClassListInfo:
     def get(self, type: Type[TLOCList], id: UUID) -> TLOCListInfo:
         ...
 
+    @overload
+    def get(self, type: Type[TrunkGroupList], id: UUID) -> TrunkGroupListInfo:
+        ...
+
     @overload
     def get(self, type: Type[URLBlockList], id: UUID) -> URLBlockListInfo:
         ...

catalystwan/endpoints/configuration/policy/list/trunkgroup.py

@@ -0,0 +1,48 @@
+# 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 PolicyListEndpoints
+from catalystwan.models.policy.list.trunkgroup import TrunkGroupList, TrunkGroupListEditPayload, TrunkGroupListInfo
+from catalystwan.models.policy.policy_list import InfoTag, PolicyListId, PolicyListPreview
+from catalystwan.typed_list import DataSequence
+
+
+class ConfigurationPolicyTrunkGroupList(APIEndpoints, PolicyListEndpoints):
+    @post("/template/policy/list/trunkgroup")
+    def create_policy_list(self, payload: TrunkGroupList) -> PolicyListId:
+        ...
+
+    @delete("/template/policy/list/trunkgroup/{id}")
+    def delete_policy_list(self, id: UUID) -> None:
+        ...
+
+    @delete("/template/policy/list/trunkgroup")
+    def delete_policy_lists_with_info_tag(self, params: InfoTag) -> None:
+        ...
+
+    @put("/template/policy/list/trunkgroup/{id}")
+    def edit_policy_list(self, id: UUID, payload: TrunkGroupListEditPayload) -> None:
+        ...
+
+    @get("/template/policy/list/trunkgroup/{id}")
+    def get_lists_by_id(self, id: UUID) -> TrunkGroupListInfo:
+        ...
+
+    @get("/template/policy/list/trunkgroup", "data")
+    def get_policy_lists(self) -> DataSequence[TrunkGroupListInfo]:
+        ...
+
+    @get("/template/policy/list/trunkgroup/filtered", "data")
+    def get_policy_lists_with_info_tag(self, params: InfoTag) -> DataSequence[TrunkGroupListInfo]:
+        ...
+
+    @post("/template/policy/list/trunkgroup/preview")
+    def preview_policy_list(self, payload: TrunkGroupList) -> PolicyListPreview:
+        ...
+
+    @get("/template/policy/list/trunkgroup/preview/{id}")
+    def preview_policy_list_by_id(self, id: UUID) -> PolicyListPreview:
+        ...

catalystwan/endpoints/endpoints_container.py

@@ -64,6 +64,7 @@
 from catalystwan.endpoints.configuration.policy.list.site import ConfigurationPolicySiteList
 from catalystwan.endpoints.configuration.policy.list.sla import ConfigurationPolicySLAClassList
 from catalystwan.endpoints.configuration.policy.list.tloc import ConfigurationPolicyTLOCList
+from catalystwan.endpoints.configuration.policy.list.trunkgroup import ConfigurationPolicyTrunkGroupList
 from catalystwan.endpoints.configuration.policy.list.url_allow_list import ConfigurationPolicyURLAllowList
 from catalystwan.endpoints.configuration.policy.list.url_block_list import ConfigurationPolicyURLBlockList
 from catalystwan.endpoints.configuration.policy.list.vpn import ConfigurationPolicyVPNList
@@ -123,6 +124,7 @@ def __init__(self, session: ManagerSession):
         self.site = ConfigurationPolicySiteList(session)
         self.sla = ConfigurationPolicySLAClassList(session)
         self.tloc = ConfigurationPolicyTLOCList(session)
+        self.trunkgroup = ConfigurationPolicyTrunkGroupList(session)
         self.url_block_list = ConfigurationPolicyURLBlockList(session)
         self.url_allow_list = ConfigurationPolicyURLAllowList(session)
         self.vpn = ConfigurationPolicyVPNList(session)

catalystwan/models/policy/__init__.py

@@ -35,6 +35,7 @@
 from catalystwan.models.policy.list.site import SiteList, SiteListInfo
 from catalystwan.models.policy.list.sla import SLAClassList, SLAClassListInfo
 from catalystwan.models.policy.list.tloc import TLOCList, TLOCListInfo
+from catalystwan.models.policy.list.trunkgroup import TrunkGroupList, TrunkGroupListInfo
 from catalystwan.models.policy.list.url import URLAllowList, URLAllowListInfo, URLBlockList, URLBlockListInfo
 from catalystwan.models.policy.list.vpn import VPNList, VPNListInfo
 from catalystwan.models.policy.list.zone import ZoneList, ZoneListInfo
@@ -115,6 +116,7 @@
         SiteList,
         SLAClassList,
         TLOCList,
+        TrunkGroupList,
         URLBlockList,
         URLAllowList,
         VPNList,
@@ -150,6 +152,7 @@
         SiteListInfo,
         SLAClassListInfo,
         TLOCListInfo,
+        TrunkGroupListInfo,
         URLAllowListInfo,
         URLBlockListInfo,
         VPNListInfo,
@@ -231,6 +234,7 @@
     "SLAClassList",
     "TLOCActionType",
     "TLOCList",
+    "TrunkGroupList",
     "TrafficDataDirection",
     "TrafficDataPolicy",
     "UnifiedSecurityPolicy",

catalystwan/models/policy/list/trunkgroup.py

@@ -0,0 +1,57 @@
+# Copyright 2022 Cisco Systems, Inc. and its affiliates
+
+from typing import List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from catalystwan.models.policy.policy_list import PolicyListBase, PolicyListId, PolicyListInfo
+
+HuntSchemeMethods = Literal["least-idle", "least-used", "longest-idle", "round-robin", "sequential", "random"]
+HuntSchemeMethodPrecedences = Literal["even", "odd", "both"]
+
+
+class TrunkGroupListEntry(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    hunt_scheme_method: HuntSchemeMethods = Field(
+        validation_alias="huntSchemeMethod", serialization_alias="huntSchemeMethod"
+    )
+    hunt_scheme_precedence: Optional[HuntSchemeMethodPrecedences] = Field(
+        default=None, validation_alias="huntSchemePrecedence", serialization_alias="huntSchemePrecedence"
+    )
+    max_calls_in: Optional[int] = Field(
+        default=None, validation_alias="maxCallsIn", serialization_alias="maxCallsIn", ge=0, le=1000
+    )
+    max_calls_out: Optional[int] = Field(
+        default=None, validation_alias="maxCallsOut", serialization_alias="maxCallsOut", ge=0, le=1000
+    )
+    max_retries: int = Field(validation_alias="maxRetries", serialization_alias="maxRetries", ge=1, le=5)
+
+
+class TrunkGroupList(PolicyListBase):
+    type: Literal["trunkGroup"] = "trunkGroup"
+    entries: List[TrunkGroupListEntry] = []
+
+    def add_entry(
+        self,
+        hunt_scheme_method: HuntSchemeMethods,
+        hunt_scheme_precedence: Optional[HuntSchemeMethodPrecedences],
+        max_retries: int,
+        max_calls_in: Optional[int] = None,
+        max_calls_out: Optional[int] = None,
+    ) -> None:
+        entry = TrunkGroupListEntry(
+            hunt_scheme_method=hunt_scheme_method,
+            hunt_scheme_precedence=hunt_scheme_precedence,
+            max_retries=max_retries,
+            max_calls_in=max_calls_in,
+            max_calls_out=max_calls_out,
+        )
+        self.entries.append(entry)
+
+
+class TrunkGroupListEditPayload(TrunkGroupList, PolicyListId):
+    pass
+
+
+class TrunkGroupListInfo(TrunkGroupList, PolicyListInfo):
+    pass