dev-uxmt: multiple minor fixes and improvements (#613)

* allow any payload for topology

* refactor security policy api

* fix policy items

* bump v0.33.4dev2

catalystwan/api/config_group_api.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, overload
 from uuid import UUID
 
 from catalystwan.typed_list import DataSequence
@@ -32,8 +32,8 @@
 
 class ConfigGroupAPI:
     def __init__(self, session: ManagerSession):
-        self.session = session
-        self.endpoint = ConfigurationGroup(session)
+        self._session = session
+        self._endpoints = ConfigurationGroup(session)
 
     def associate(self, cg_id: str, device_ids: list) -> None:
         """
@@ -46,7 +46,7 @@ def associate(self, cg_id: str, device_ids: list) -> None:
 
         payload = ConfigGroupAssociatePayload(devices=devices)
 
-        self.endpoint.associate(config_group_id=cg_id, payload=payload)
+        self._endpoints.associate(config_group_id=cg_id, payload=payload)
 
     def create(self, name: str, description: str, solution: Solution, profile_ids: list) -> ConfigGroupCreationResponse:
         """
@@ -60,7 +60,7 @@ def create(self, name: str, description: str, solution: Solution, profile_ids: l
             name=name, description=description, solution=solution, profiles=profiles
         )
 
-        return self.endpoint.create_config_group(cg_payload)
+        return self._endpoints.create_config_group(cg_payload)
 
     def create_variables(
         self, cg_id: str, device_ids: list, suggestions: bool = True
@@ -69,13 +69,13 @@ def create_variables(
         Creates device specific variable data in given config-group
         """
         payload = ConfigGroupVariablesCreatePayload(deviceIds=device_ids, suggestions=suggestions)
-        return self.endpoint.create_variables(config_group_id=cg_id, payload=payload)
+        return self._endpoints.create_variables(config_group_id=cg_id, payload=payload)
 
     def delete(self, cg_id: str) -> None:
         """
         Deletes existing config-group with given ID
         """
-        self.endpoint.delete_config_group(cg_id)
+        self._endpoints.delete_config_group(cg_id)
 
     def deploy(self, cg_id: str, device_ids: list) -> ConfigGroupDeployResponse:
         """
@@ -86,7 +86,7 @@ def deploy(self, cg_id: str, device_ids: list) -> ConfigGroupDeployResponse:
             devices.append(DeviceId(id=device_id))
 
         payload = ConfigGroupDeployPayload(devices=devices)
-        return self.endpoint.deploy(config_group_id=cg_id, payload=payload)
+        return self._endpoints.deploy(config_group_id=cg_id, payload=payload)
 
     def disassociate(self, cg_id: str, device_ids: list) -> ConfigGroupDisassociateResponse:
         """
@@ -97,7 +97,7 @@ def disassociate(self, cg_id: str, device_ids: list) -> ConfigGroupDisassociateR
             devices.append(DeviceId(id=device_id))
 
         payload = ConfigGroupAssociatePayload(devices=devices)
-        return self.endpoint.disassociate(config_group_id=cg_id, payload=payload)
+        return self._endpoints.disassociate(config_group_id=cg_id, payload=payload)
 
     def edit(
         self, cg_id: str, name: str, description: str, solution: Solution, profile_ids: list
@@ -111,21 +111,29 @@ def edit(
             profiles.append(ProfileId(id=profile_id))
         payload = ConfigGroupEditPayload(name=name, description=description, solution=solution, profiles=profiles)
 
-        return self.endpoint.edit_config_group(config_group_id=cg_id, payload=payload)
+        return self._endpoints.edit_config_group(config_group_id=cg_id, payload=payload)
 
-    def get(self, group_id: Optional[UUID] = None) -> Union[DataSequence[ConfigGroup], ConfigGroup, None]:
+    @overload
+    def get(self) -> DataSequence[ConfigGroup]:
+        ...
+
+    @overload
+    def get(self, group_id: UUID) -> ConfigGroup:
+        ...
+
+    def get(self, group_id: Optional[UUID] = None) -> Any:
         """
         Gets list of existing config-groups or single config-group with given ID
-         If given ID is not correct return None
+        If given ID is not correct return None
         """
         if group_id is None:
-            return self.endpoint.get()
-        return self.endpoint.get().filter(id=group_id).single_or_default()
+            return self._endpoints.get()
+        return self._endpoints.get().filter(id=group_id).single_or_default()
 
     def update_variables(self, cg_id: str, solution: Solution, device_variables: list) -> None:
         """
         Updates device specific variable data in given config-group
         """
         payload = ConfigGroupVariablesEditPayload(solution=solution, devices=device_variables)
 
-        self.endpoint.update_variables(config_group_id=cg_id, payload=payload)
+        self._endpoints.update_variables(config_group_id=cg_id, payload=payload)

catalystwan/api/policy_api.py

@@ -164,7 +164,7 @@
 from catalystwan.models.policy.policy_list import PolicyListBase
 from catalystwan.models.policy.security import (
     AnySecurityPolicy,
-    AnySecurityPolicyInfo,
+    AnySecurityPolicyInfoList,
     SecurityPolicy,
     SecurityPolicyEditResponse,
     UnifiedSecurityPolicy,
@@ -328,7 +328,7 @@ def delete(self, id: UUID) -> None:
         self._endpoints.delete_security_template(id)
 
     @overload
-    def get(self) -> List[AnySecurityPolicyInfo]:
+    def get(self) -> AnySecurityPolicyInfoList:
         ...
 
     @overload
@@ -338,7 +338,7 @@ def get(self, id: UUID) -> AnySecurityPolicy:
     def get(self, id: Optional[UUID] = None) -> Any:
         if id is not None:
             return self._endpoints.get_security_template(id).root
-        return [info.root for info in self._endpoints.generate_security_template_list()]
+        return self._endpoints.generate_security_template_list()
 
 
 class PolicyListsAPI:

catalystwan/endpoints/configuration/policy/security_template.py

@@ -6,11 +6,10 @@
 from catalystwan.endpoints import JSON, APIEndpoints, delete, get, post, put
 from catalystwan.models.policy.security import (
     AnySecurityPolicy,
+    AnySecurityPolicyInfoList,
     SecurityPolicyEditResponse,
-    SecurityPolicyInfoRoot,
     SecurityPolicyRoot,
 )
-from catalystwan.typed_list import DataSequence
 
 
 class ConfigurationSecurityTemplatePolicy(APIEndpoints):
@@ -36,7 +35,7 @@ def generate_security_policy_summary(self):
         ...
 
     @get("/template/policy/security", "data")
-    def generate_security_template_list(self) -> DataSequence[SecurityPolicyInfoRoot]:
+    def generate_security_template_list(self) -> AnySecurityPolicyInfoList:
         ...
 
     def get_device_list_by_id(self):

catalystwan/endpoints/configuration_group.py

@@ -2,7 +2,7 @@
 
 # mypy: disable-error-code="empty-body"
 from datetime import datetime
-from typing import List, Optional
+from typing import Any, List, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, ConfigDict, Field
@@ -58,7 +58,7 @@ class ConfigGroup(BaseModel):
         serialization_alias="numberOfDevicesUpToDate", validation_alias="numberOfDevicesUpToDate"
     )
     origin: Optional[str] = None
-    topology: Optional[str] = None
+    topology: Any = None
     full_config_cli: Optional[bool] = Field(
         default=None, serialization_alias="fullConfigCli", validation_alias="fullConfigCli"
     )

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

@@ -15,7 +15,7 @@ class AsPathEntry(BaseModel):
 class AsPathParcel(_ParcelBase):
     type_: Literal["as-path"] = Field(default="as-path", exclude=True)
     as_path_list_num: Global[int] = Field(validation_alias=AliasPath("data", "asPathListNum"))
-    entries: List[AsPathEntry] = Field(validation_alias=AliasPath("data", "entries"))
+    entries: List[AsPathEntry] = Field(default=[], validation_alias=AliasPath("data", "entries"))
 
     def add_as_path(self, as_path: str):
         self.entries.append(AsPathEntry(as_path=as_global(as_path)))

catalystwan/models/policy/definition/amp.py

@@ -57,7 +57,7 @@ class AdvancedMalwareProtectionDefinition(BaseModel):
 
 class AdvancedMalwareProtectionPolicy(PolicyDefinitionBase):
     type: Literal["advancedMalwareProtection"] = "advancedMalwareProtection"
-    mode: AMPPolicyType
+    mode: AMPPolicyType = "security"
     definition: AdvancedMalwareProtectionDefinition
 
 

catalystwan/models/policy/definition/zone_based_firewall.py

@@ -21,6 +21,7 @@
     DestinationIPEntry,
     DestinationPortEntry,
     DestinationPortListEntry,
+    DestinationScalableGroupTagListEntry,
     LogAction,
     Match,
     PolicyActionType,
@@ -40,6 +41,7 @@
     SourceIPEntry,
     SourcePortEntry,
     SourcePortListEntry,
+    SourceScalableGroupTagListEntry,
 )
 
 ZoneBasedFWPolicySequenceEntry = Annotated[
@@ -53,6 +55,7 @@
         DestinationIPEntry,
         DestinationPortEntry,
         DestinationPortListEntry,
+        DestinationScalableGroupTagListEntry,
         ProtocolEntry,
         ProtocolNameEntry,
         ProtocolNameListEntry,
@@ -65,6 +68,7 @@
         SourceIPEntry,
         SourcePortEntry,
         SourcePortListEntry,
+        SourceScalableGroupTagListEntry,
     ],
     Field(discriminator="field"),
 ]

catalystwan/models/policy/policy_definition.py

@@ -567,11 +567,21 @@ class SourcePortListEntry(BaseModel):
     ref: UUID
 
 
+class SourceScalableGroupTagListEntry(BaseModel):
+    field: Literal["sourceScalableGroupTagList"] = "sourceScalableGroupTagList"
+    ref: UUID
+
+
 class DestinationPortListEntry(BaseModel):
     field: Literal["destinationPortList"] = "destinationPortList"
     ref: UUID
 
 
+class DestinationScalableGroupTagListEntry(BaseModel):
+    field: Literal["destinationScalableGroupTagList"] = "destinationScalableGroupTagList"
+    ref: UUID
+
+
 class RuleSetListEntry(BaseModel):
     field: Literal["ruleSetList"] = "ruleSetList"
     ref: str
@@ -858,6 +868,7 @@ class ActionSet(BaseModel):
         DestinationPortEntry,
         DestinationPortListEntry,
         DestinationRegionEntry,
+        DestinationScalableGroupTagListEntry,
         DNSAppListEntry,
         DNSEntry,
         DomainIDEntry,
@@ -894,6 +905,7 @@ class ActionSet(BaseModel):
         SourceIPv6Entry,
         SourcePortEntry,
         SourcePortListEntry,
+        SourceScalableGroupTagListEntry,
         TCPEntry,
         TLOCEntry,
         TLOCListEntry,

catalystwan/models/policy/security.py

@@ -19,6 +19,7 @@
     URLFilteringAssemblyItem,
     ZoneBasedFWAssemblyItem,
 )
+from catalystwan.typed_list import DataSequence
 
 SecurityPolicyAssemblyItem = Annotated[
     Union[
@@ -121,7 +122,7 @@ class UnifiedSecurityPolicyDefinition(PolicyDefinition):
 
 
 class SecurityPolicy(PolicyCreationPayload):
-    policy_mode: Union[Literal["security"], None] = Field(
+    policy_mode: Literal[None, "security"] = Field(
         default="security", serialization_alias="policyMode", validation_alias="policyMode"
     )
     policy_type: str = Field(default="feature", serialization_alias="policyType", validation_alias="policyType")
@@ -223,5 +224,17 @@ class UnifiedSecurityPolicyInfo(UnifiedSecurityPolicy, PolicyInfo):
 AnySecurityPolicyInfo = Union[SecurityPolicyInfo, UnifiedSecurityPolicyInfo]
 
 
-class SecurityPolicyInfoRoot(RootModel):
-    root: AnySecurityPolicyInfo
+class AnySecurityPolicyInfoList(RootModel):
+    root: List[AnySecurityPolicyInfo]
+
+    @property
+    def all(self) -> List[AnySecurityPolicyInfo]:
+        return self.root
+
+    @property
+    def security(self) -> DataSequence[SecurityPolicyInfo]:
+        return DataSequence(SecurityPolicyInfo, [p for p in self.root if p.policy_mode == "security"])
+
+    @property
+    def unified(self) -> DataSequence[UnifiedSecurityPolicyInfo]:
+        return DataSequence(UnifiedSecurityPolicyInfo, [p for p in self.root if p.policy_mode == "unified"])