Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Appqoe Model and converter #530

Merged
merged 1 commit into from
Mar 19, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
# Copyright 2024 Cisco Systems, Inc. and its affiliates

from ipaddress import IPv4Address
from typing import List, Literal, Optional, Union

from pydantic import BaseModel, ConfigDict, Field
from pydantic import AliasPath, BaseModel, ConfigDict, Field

from catalystwan.api.configuration_groups.parcel import Default, Global, Variable
from catalystwan.api.configuration_groups.parcel import Default, Global, Variable, _ParcelBase, as_default, as_global

VirtualApplicationType = Literal["dreopt"]

Expand All @@ -26,6 +27,40 @@

AppnavControllerGroupName = Literal["ACG-APPQOE"]
ServiceNodeGroupName = Literal["SNG-APPQOE"]
ServiceNodeGroupsNames = Literal[
"SNG-APPQOE",
"SNG-APPQOE1",
"SNG-APPQOE2",
"SNG-APPQOE3",
"SNG-APPQOE4",
"SNG-APPQOE5",
"SNG-APPQOE6",
"SNG-APPQOE7",
"SNG-APPQOE8",
"SNG-APPQOE9",
"SNG-APPQOE10",
"SNG-APPQOE11",
"SNG-APPQOE12",
"SNG-APPQOE13",
"SNG-APPQOE14",
"SNG-APPQOE15",
"SNG-APPQOE16",
"SNG-APPQOE17",
"SNG-APPQOE18",
"SNG-APPQOE19",
"SNG-APPQOE20",
"SNG-APPQOE21",
"SNG-APPQOE22",
"SNG-APPQOE23",
"SNG-APPQOE24",
"SNG-APPQOE25",
"SNG-APPQOE26",
"SNG-APPQOE27",
"SNG-APPQOE28",
"SNG-APPQOE29",
"SNG-APPQOE30",
"SNG-APPQOE31",
]
ForwarderAndServiceNodeAddress = Literal["192.168.2.2"] # TODO: 1.Is it really constant? 2.Use ipaddress.IPv4Address?
ForwarderAndServiceNodeControllerAddress = Literal[
"192.168.2.1"
Expand Down Expand Up @@ -66,8 +101,8 @@ class Appqoe(BaseModel):
serialization_alias="serviceNodeGroup",
validation_alias="serviceNodeGroup",
)
service_node_groups: List[Global[ServiceNodeGroupName]] = Field(
default=[Global[ServiceNodeGroupName](value="SNG-APPQOE")],
service_node_groups: List[Global[ServiceNodeGroupsNames]] = Field(
default=[Global[ServiceNodeGroupsNames](value="SNG-APPQOE")],
serialization_alias="serviceNodeGroups",
validation_alias="serviceNodeGroups",
)
Expand All @@ -84,14 +119,16 @@ class ServiceContext(BaseModel):


class ServiceNodeInformation(BaseModel):
address: Global[str]
address: Global[IPv4Address]


class ForwarderController(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)

address: Union[Global[str], Variable]
vpn: Global[int] = Global[int](value=1)
address: Union[Global[str], Global[IPv4Address], Variable]
vpn: Global[int] = Field(
default=Global[int](value=1), description="This is field is a depended on the Service VPN value."
)


class ForwarderAppnavControllerGroup(BaseModel):
Expand Down Expand Up @@ -210,26 +247,21 @@ class ServiceNodeRole(BaseModel):
)


class AppqoeData(BaseModel):
class AppqoeParcel(_ParcelBase):
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)

dreopt: Optional[Union[Global[bool], Default[bool]]] = Default[bool](value=False)
dreopt: Optional[Union[Global[bool], Default[bool]]] = Field(
default=as_default(False), validation_alias=AliasPath("data", "dreopt")
)
virtual_application: Optional[List[VirtualApplication]] = Field(
serialization_alias="virtualApplication", validation_alias="virtualApplication"
default=None, validation_alias=AliasPath("data", "virtualApplication")
)
appqoe_device_role: Global[str] = Field(
default=Global(value="forwarder"), serialization_alias="appqoeDeviceRole", validation_alias="appqoeDeviceRole"
default=as_global("forwarder"), validation_alias=AliasPath("data", "appqoeDeviceRole")
)

forwarder: Optional[ForwarderRole]
forwarder: Optional[ForwarderRole] = Field(default=None, validation_alias=AliasPath("data", "forwarder"))
forwarder_and_service_node: Optional[ForwarderAndServiceNodeRole] = Field(
serialization_alias="forwarderAndServiceNode", validation_alias="forwarderAndServiceNode"
default=None, validation_alias=AliasPath("data", "forwarderAndServiceNode")
)
service_node: Optional[ServiceNodeRole] = Field(serialization_alias="serviceNode", validation_alias="serviceNode")


class AppqoeCreationPayload(BaseModel):
name: str
description: Optional[str] = None
data: AppqoeData
metadata: Optional[dict] = None
service_node: Optional[ServiceNodeRole] = Field(default=None, validation_alias=AliasPath("data", "serviceNode"))
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
from copy import deepcopy

from catalystwan.api.configuration_groups.parcel import Global, as_default, as_global
from catalystwan.models.configuration.feature_profile.sdwan.service.appqoe import (
AppnavControllerGroupName,
AppqoeParcel,
ServiceNodeGroupName,
ServiceNodeGroupsNames,
)


class AppqoeTemplateConverter:
supported_template_types = ("appqoe",)

@staticmethod
def create_parcel(name: str, description: str, template_values: dict) -> AppqoeParcel:
"""
Create an AppqoeParcel object based on the provided name, description, and template values.

Args:
name (str): The name of the parcel.
description (str): The description of the parcel.
template_values (dict): The template values used to create the parcel.

Returns:
AppqoeParcel: The created AppqoeParcel object.
"""
values = deepcopy(template_values)
for appqoe_item in values.get("service_context", {}).get("appqoe", []):
if item_name := appqoe_item.get("name"):
appqoe_item["name"] = as_default(value=item_name.value)
if appnav_controller_group := appqoe_item.get("appnav_controller_group"):
appqoe_item["appnav_controller_group"] = as_global(
appnav_controller_group.value, AppnavControllerGroupName
)
if service_node_group := appqoe_item.get("service_node_group"):
appqoe_item["service_node_group"] = as_global(service_node_group.value, ServiceNodeGroupName)
if service_node_groups := appqoe_item.get("service_node_groups"):
appqoe_item["service_node_groups"] = [
Global[ServiceNodeGroupsNames](value=value) for value in service_node_groups.value
]
for group in values.get("service_node_group", []):
if group_name := group.get("group_name"):
group["group_name"] = as_default(group_name.value, ServiceNodeGroupName)
internal = group.get("internal")
if internal is not None:
group["internal"] = as_default(internal.value)
for appnav in values.get("appnav_controller_group", []):
if group_name := appnav.get("group_name"):
appnav["group_name"] = as_default(group_name.value, AppnavControllerGroupName)
for controller in appnav.get("appnav_controllers", []):
if _vpn := controller.get("vpn"): # noqa: F841
# VPN field is depended on existence of the Service VPN value
# also from UI this list contains only 1 item and should not be a list.
# AppqoeParcel.forwarder.appnav_controller_group.appnav_controllers[0].vpn
# must be populated in the parcel creation process.
pass
parcel_values = {
"parcel_name": name,
"parcel_description": description,
"forwarder": {**values}, # There is not any other option from UX1 than forwarder
}
return AppqoeParcel(**parcel_values) # type: ignore
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from catalystwan.utils.feature_template.find_template_values import find_template_values

from .aaa import AAATemplateConverter
from .appqoe import AppqoeTemplateConverter
from .banner import BannerTemplateConverter
from .base import FeatureTemplateConverter
from .basic import SystemToBasicTemplateConverter
Expand Down Expand Up @@ -41,6 +42,7 @@
UcseTemplateConverter,
DhcpTemplateConverter,
SNMPTemplateConverter,
AppqoeTemplateConverter,
]


Expand Down
Loading