dev: Converter: custom control (#767)

* Add converter and unit tests. todo: fix typechecking, add model for actions

* fix match entries in converter

* fix: circular import in utils, update: unittests

* fix: unittests

* begin work with actions conversion

* implement associate actions methods

* draft

* fix: rollback update missing for app-prio profile

* fix issues

* fix issues after testing

---------

Co-authored-by: Szymon Basan <[email protected]>

catalystwan/endpoints/configuration/topology_group.py

@@ -3,15 +3,31 @@
 # mypy: disable-error-code="empty-body"
 from uuid import UUID
 
-from catalystwan.endpoints import APIEndpoints, delete, get, post, versions
-from catalystwan.models.configuration.topology_group import TopologyGroup, TopologyGroupId
+from catalystwan.endpoints import APIEndpoints, delete, get, post, put, versions
+from catalystwan.models.configuration.topology_group import (
+    ActivateRequest,
+    DeployResponse,
+    Preview,
+    TopologyGroup,
+    TopologyGroupId,
+)
 from catalystwan.typed_list import DataSequence
 
 
 class TopologyGroupEndpoints(APIEndpoints):
     @post("/v1/topology-group")
     @versions(">=20.12")
-    def create_topology_group(self, payload: TopologyGroup) -> TopologyGroupId:
+    def create(self, payload: TopologyGroup) -> TopologyGroupId:
+        ...
+
+    @put("/v1/topology-group/{group_id}")
+    @versions(">=20.12")
+    def edit(self, group_id: UUID, payload: TopologyGroup) -> TopologyGroupId:
+        ...
+
+    @get("/v1/topology-group/{group_id}")
+    @versions(">=20.12")
+    def get_by_id(self, group_id: UUID) -> TopologyGroup:
         ...
 
     @get("/v1/topology-group")
@@ -23,3 +39,13 @@ def get_all(self) -> DataSequence[TopologyGroupId]:
     @versions(">=20.12")
     def delete(self, group_id: UUID) -> None:
         ...
+
+    @post("/v1/topology-group/{group_id}/device/{device_id}/preview")
+    @versions(">=20.12")
+    def preview(self, group_id: UUID, device_id: UUID, payload: ActivateRequest) -> Preview:
+        ...
+
+    @post("/v1/topology-group/{group_id}/device/deploy")
+    @versions(">=20.12")
+    def deploy(self, group_id: UUID, payload: ActivateRequest) -> DeployResponse:
+        ...

catalystwan/integration_tests/feature_profile/sdwan/topology/test_topology.py

@@ -67,7 +67,7 @@ def test_hubspoke(self):
     def test_custom_control(self):
         cc = CustomControlParcel(parcel_name="CustomControlParcel-1")
         cc.set_default_action("accept")
-        cc.assign_target([self.lanvpn_parcel_name])
+        cc.assign_target_sites([self.lanvpn_parcel_name])
         s = cc.add_sequence("my_sequence", 1, "route", "ipv4", "reject")
         s.match_carrier("carrier4")
         s.match_domain_id(555)

catalystwan/models/configuration/config_migration.py

@@ -40,7 +40,7 @@
 )
 from catalystwan.models.settings import ThreatGridApi
 from catalystwan.models.templates import FeatureTemplateInformation, TemplateInformation
-from catalystwan.version import parse_api_version
+from catalystwan.version import NullVersion, parse_api_version
 
 T = TypeVar("T", bound=AnyParcel)
 TO = TypeVar("TO")
@@ -235,6 +235,9 @@ def insert_parcel_type_from_headers(cls, values: Dict[str, Any]):
     def list_transformed_parcels_with_origin(self, origin: Set[UUID]) -> List[TransformedParcel]:
         return [p for p in self.profile_parcels if p.header.origin in origin]
 
+    def remove_transformed_parcels_with_origin(self, origin: Set[UUID]):
+        self.profile_parcels = [p for p in self.profile_parcels if p.header.origin not in origin]
+
     def add_subelement_in_config_group(
         self, profile_types: List[ProfileType], device_template_id: UUID, subelement: UUID
     ) -> bool:
@@ -542,6 +545,7 @@ class QoSMapResidues:
 @dataclass
 class PolicyConvertContext:
     # conversion input
+    platform_version: Version = NullVersion()
     region_map: Dict[str, int] = field(default_factory=dict)
     site_map: Dict[str, int] = field(default_factory=dict)
     lan_vpn_map: Dict[str, Union[int, str]] = field(default_factory=dict)
@@ -582,8 +586,9 @@ def generate_as_path_list_num_from_name(self, name: str) -> int:
     def from_configs(
         network_hierarchy: List[NodeInfo],
         transformed_parcels: List[TransformedParcel],
+        platform_version: Version,
     ) -> "PolicyConvertContext":
-        context = PolicyConvertContext()
+        context = PolicyConvertContext(platform_version=platform_version)
         for node in network_hierarchy:
             if node.data.hierarchy_id is not None:
                 if node.data.label == "SITE" and node.data.hierarchy_id.site_id is not None:

catalystwan/models/configuration/feature_profile/sdwan/topology/custom_control.py

@@ -1,5 +1,5 @@
 from ipaddress import IPv4Address
-from typing import List, Literal, Optional
+from typing import List, Literal, Optional, overload
 from uuid import UUID
 
 from pydantic import AliasPath, BaseModel, ConfigDict, Field, model_validator
@@ -69,6 +69,13 @@ class Tloc(BaseModel):
     encap: Optional[Global[EncapType]] = Field(default=None)
     ip: Optional[Global[str]] = Field(default=None)
 
+    @staticmethod
+    def from_params(color: TLOCColor, encap: EncapType, ip: IPv4Address) -> "Tloc":
+        _color = as_global(color, TLOCColor)
+        _encap = as_global(encap, EncapType)
+        _ip = as_global(str(ip))
+        return Tloc(color=_color, encap=_encap, ip=_ip)
+
 
 class Entry(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
@@ -162,7 +169,7 @@ class Actions(BaseModel):
 
 class Sequence(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
-    actions: Optional[List[Actions]] = Field(default=None)
+    actions: Optional[List[Actions]] = Field(default_factory=list)
     base_action: Optional[Global[BaseAction]] = Field(
         default=None, validation_alias="baseAction", serialization_alias="baseAction"
     )
@@ -199,6 +206,19 @@ def _get_regions_entry(self) -> Entry:
         entries.append(entry)
         return entry
 
+    @property
+    def _action(self) -> Actions:
+        if not self.actions:
+            self.actions = [Actions()]
+        return self.actions[0]
+
+    @property
+    def _action_set(self) -> ActionSet:
+        action = self._action
+        if action.set is None:
+            action.set = [ActionSet()]
+        return action.set[0]
+
     def match_carrier(self, carrier: CarrierType):
         entry = Entry(carrier=as_global(carrier, CarrierType))
         self._match(entry)
@@ -284,6 +304,56 @@ def match_vpns(self, vpns: List[str]):
         entry = Entry(vpn=as_global(vpns))
         self._match(entry)
 
+    def associate_affinitty_action(self, affinity: int) -> None:
+        self._action_set.affinity = as_global(affinity)
+
+    def associate_community_additive_action(self, additive: bool) -> None:
+        self._action_set.community_additive = as_global(additive)
+
+    def associate_community_action(self, community: str) -> None:
+        self._action_set.community = as_global(community)
+
+    def associate_omp_tag_action(self, omp_tag: int) -> None:
+        self._action_set.omp_tag = as_global(omp_tag)
+
+    def associate_preference_action(self, preference: int) -> None:
+        self._action_set.preference = as_global(preference)
+
+    @overload
+    def associate_service_action(self, service_type: ServiceType, vpn: Optional[int], *, tloc_list_id: UUID) -> None:
+        ...
+
+    @overload
+    def associate_service_action(
+        self, service_type: ServiceType, vpn: Optional[int], *, ip: IPv4Address, color: TLOCColor, encap: EncapType
+    ) -> None:
+        ...
+
+    def associate_service_action(
+        self, service_type=ServiceType, vpn=Optional[int], *, tloc_list_id=None, ip=None, color=None, encap=None
+    ) -> None:
+        _vpn = as_global(vpn) if vpn is not None else None
+        _service_type = as_global(service_type, ServiceType)
+        if tloc_list_id is not None:
+            service = Service(
+                tloc_list=RefIdItem(ref_id=as_global(str(tloc_list_id))),
+                type=_service_type,
+                vpn=_vpn,
+            )
+        else:
+            _tloc = Tloc.from_params(color=color, encap=encap, ip=ip)
+            service = Service(tloc=_tloc, type=_service_type, vpn=_vpn)
+        self._action_set.service = service
+
+    def associate_tloc(self, color: TLOCColor, encap: EncapType, ip: IPv4Address) -> None:
+        self._action_set.tloc = Tloc.from_params(color=color, encap=encap, ip=ip)
+
+    def associate_tloc_action(self, tloc_action_type: TLOCActionType) -> None:
+        self._action_set.tloc_action = as_global(tloc_action_type, TLOCActionType)
+
+    def associate_tloc_list(self, tloc_list_id: UUID) -> None:
+        self._action_set.tloc_list = RefIdItem(ref_id=as_global(str(tloc_list_id)))
+
 
 class CustomControlParcel(_ParcelBase):
     model_config = ConfigDict(populate_by_name=True)
@@ -305,16 +375,17 @@ class CustomControlParcel(_ParcelBase):
     def set_default_action(self, action: BaseAction = "reject"):
         self.default_action = as_global(action, BaseAction)
 
-    def assign_target(
+    def assign_target_sites(
         self,
-        vpns: List[str],
-        inbound_sites: Optional[List[str]] = None,
-        outbound_sites: Optional[List[str]] = None,
+        inbound_sites: List[str],
+        outbound_sites: List[str],
+        _dummy_vpns: List[str] = [";dummy-vpn"],
     ) -> Target:
         self.target = Target(
+            vpn=Global[List[str]](value=_dummy_vpns),
+            level=as_global("SITE", Level),
             inbound_sites=as_global(inbound_sites) if inbound_sites else None,
             outbound_sites=as_global(outbound_sites) if outbound_sites else None,
-            vpn=Global[List[str]](value=vpns),
         )
         return self.target
 

catalystwan/models/configuration/topology_group.py

@@ -31,5 +31,21 @@ def add_profiles(self, ids: List[UUID]):
         self.profiles.extend([Profile(id=i) for i in ids])
 
 
+class ActivateRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    deactivate_topology: bool = Field(validation_alias="deactivateTopology", serialization_alias="deactivateTopology")
+
+
+class DeployResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    parent_task_id: str = Field(validation_alias="parentTaskId", serialization_alias="parentTaskId")
+
+
+class Preview(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    existing_config: str = Field(validation_alias="existingConfig", serialization_alias="existingConfig")
+    new_config: str = Field(validation_alias="newConfig", serialization_alias="newConfig")
+
+
 class TopologyGroupId(BaseModel):
     id: UUID

catalystwan/models/policy/centralized.py

@@ -71,7 +71,7 @@ def assign_to_inbound_sites(self, site_lists: List[UUID]) -> None:
         self.entries.append(entry)
 
     def assign_to_outbound_sites(self, site_lists: List[UUID]) -> None:
-        entry = ControlApplicationEntry(direction="in", site_lists=site_lists)
+        entry = ControlApplicationEntry(direction="out", site_lists=site_lists)
         self.entries.append(entry)
 
     @overload
@@ -200,6 +200,12 @@ class CentralizedPolicyDefinition(PolicyDefinition):
     assembly: List[AnyAssemblyItem] = []
     model_config = ConfigDict(populate_by_name=True)
 
+    def find_assembly_item_by_definition_id(self, definition_id: UUID) -> Optional[AnyAssemblyItem]:
+        for item in self.assembly:
+            if item.definition_id == definition_id:
+                return item
+        return None
+
 
 class CentralizedPolicy(PolicyCreationPayload):
     model_config = ConfigDict(populate_by_name=True)

catalystwan/models/policy/definition/control.py

@@ -1,14 +1,15 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
 from ipaddress import IPv4Address
-from typing import Any, List, Literal, Optional, Union, overload
+from typing import List, Literal, Optional, Union, overload
 from uuid import UUID
 
 from pydantic import ConfigDict, Field
 from typing_extensions import Annotated
 
-from catalystwan.models.common import AcceptRejectActionType, EncapType, TLOCColor
+from catalystwan.models.common import AcceptRejectActionType, EncapType, SequenceIpType, TLOCColor
 from catalystwan.models.policy.policy_definition import (
+    ActionSet,
     AffinityEntry,
     CarrierEntry,
     CarrierType,
@@ -49,6 +50,7 @@
     TLOCEntry,
     TLOCEntryValue,
     TLOCListEntry,
+    VPNEntry,
     VPNListEntry,
     accept_action,
 )
@@ -73,6 +75,7 @@
         TLOCEntry,
         TLOCListEntry,
         VPNListEntry,
+        VPNEntry,
     ],
     Field(discriminator="field"),
 ]
@@ -99,16 +102,22 @@
     Field(discriminator="field"),
 ]
 
-ControlPolicyRouteSequenceActions = Any  # TODO
-ControlPolicyTLOCSequenceActions = Any  # TODO
+ControlPolicyRouteSequenceActions = Annotated[
+    Union[
+        ActionSet,
+        ExportToAction,
+    ],
+    Field(discriminator="type"),
+]
+ControlPolicyTLOCSequenceActions = ActionSet
 
 
 class ControlPolicyHeader(PolicyDefinitionBase):
     type: Literal["control"] = "control"
 
 
 class ControlPolicyRouteSequenceMatch(Match):
-    entries: List[AnyControlPolicyRouteSequenceMatchEntry] = []
+    entries: List[AnyControlPolicyRouteSequenceMatchEntry] = Field(default_factory=list)
 
 
 class ControlPolicyTLOCSequenceMatch(Match):
@@ -218,7 +227,7 @@ def associate_service_action(
         else:
             tloc_entry = None
             tloc_list_entry = TLOCListEntry(ref=tloc_list_id)
-        service_value = ServiceEntryValue(type=service_type, vpn=str(vpn), tloc=tloc_entry, tloc_list=tloc_list_entry)
+        service_value = ServiceEntryValue(type=service_type, vpn=vpn, tloc=tloc_entry, tloc_list=tloc_list_entry)
         self._insert_action_in_set(ServiceEntry(value=service_value))
 
     @accept_action
@@ -322,23 +331,23 @@ class ControlPolicy(ControlPolicyHeader, DefinitionWithSequencesCommonBase):
     model_config = ConfigDict(populate_by_name=True)
 
     def add_route_sequence(
-        self, name: str = "Route", base_action: AcceptRejectActionType = "reject"
+        self, name: str = "Route", base_action: AcceptRejectActionType = "reject", ip_type: SequenceIpType = "ipv4"
     ) -> ControlPolicyRouteSequence:
         seq = ControlPolicyRouteSequence(
             sequence_name=name,
             base_action=base_action,
-            sequence_ip_type="ipv4",
+            sequence_ip_type=ip_type,
         )
         self.add(seq)
         return seq
 
     def add_tloc_sequence(
-        self, name: str = "TLOC", base_action: AcceptRejectActionType = "reject"
+        self, name: str = "TLOC", base_action: AcceptRejectActionType = "reject", ip_type: SequenceIpType = "ipv4"
     ) -> ControlPolicyTLOCSequence:
         seq = ControlPolicyTLOCSequence(
             sequence_name=name,
             base_action=base_action,
-            sequence_ip_type="ipv4",
+            sequence_ip_type=ip_type,
         )
         self.add(seq)
         return seq

catalystwan/models/policy/list/as_path.py

@@ -17,6 +17,10 @@ class ASPathList(PolicyListBase):
     type: Literal["asPath"] = "asPath"
     entries: List[ASPathListEntry] = []
 
+    def add_as_path(self, as_path: str):
+        as_path_entry = ASPathListEntry(as_path=as_path)
+        self._add_entry(entry=as_path_entry)
+
 
 class ASPathListEditPayload(ASPathList, PolicyListId):
     pass