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

Commit

Permalink
Merge branch 'main' into fr/enhance_feature_template_docs
Browse files Browse the repository at this point in the history
  • Loading branch information
@cicharka
cicharka committed May 22, 2024
2 parents f159059 + 5454f14 commit 8cccb61
Show file tree
Hide file tree
Showing 6 changed files with 230 additions and 27 deletions.
6 changes: 1 addition & 5 deletions catalystwan/api/templates/feature_template.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,16 +122,12 @@ def get(cls, session: ManagerSession, name: str) -> FeatureTemplate:

feature_template_model = choose_model(type_value=template_info.template_type)

device_specific_variables: Dict[str, DeviceVariable] = {}
values_from_template_definition = find_template_values(
template_definition_as_dict, device_specific_variables=device_specific_variables
)
values_from_template_definition = find_template_values(template_definition_as_dict)
flattened_values = flatten_dict(values_from_template_definition)

return feature_template_model(
template_name=template_info.name,
template_description=template_info.description,
device_models=[model for model in template_info.device_type],
device_specific_variables=device_specific_variables,
**flattened_values,
)
6 changes: 3 additions & 3 deletions catalystwan/exceptions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,9 @@


class ManagerErrorInfo(BaseModel):
message: Union[str, None]
details: Union[str, None]
code: Union[str, None]
message: Union[str, None] = None
details: Union[str, None] = None
code: Union[str, None] = None


class CatalystwanException(Exception):
Expand Down
8 changes: 6 additions & 2 deletions catalystwan/response.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -120,8 +120,9 @@ def parse_cookies_to_dict(cookies: str) -> Dict[str, str]:


class JsonPayload:
def __init__(self, json: Any = None):
def __init__(self, json: Any = None, empty: bool = False):
self.json = json
self.empty = empty
self.data = None
self.error = None
self.headers = None
Expand All @@ -141,7 +142,7 @@ def __init__(self, response: Response):
try:
self.payload = JsonPayload(response.json())
except JSONDecodeError:
self.payload = JsonPayload()
self.payload = JsonPayload(empty=True)

def _detect_expired_jsessionid(self) -> bool:
"""Determines if server sent expired JSESSIONID"""
Expand Down Expand Up @@ -184,6 +185,9 @@ def dataseq(self, cls: Type[T], sourcekey: Optional[str] = "data", validate: boo
DataSequence[T] of given type T which is subclassing from Dataclass/BaseModel,
in case JSON payload was containing a single Object - sequence with one element is returned
"""
if self.payload.empty:
return DataSequence(cls, [])

if sourcekey is None:
data = self.payload.json
else:
Expand Down
192 changes: 192 additions & 0 deletions catalystwan/tests/templates/test_find_template_values.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,192 @@
from catalystwan.api.templates.device_variable import DeviceVariable
from catalystwan.utils.feature_template.find_template_values import find_template_values


def test_find_template_values():
input_values = {
"vpn-id": {"vipObjectType": "object", "vipType": "constant", "vipValue": 0},
"name": {
"vipObjectType": "object",
"vipType": "ignore",
"vipVariableName": "vpn_name",
},
"ecmp-hash-key": {
"layer4": {
"vipObjectType": "object",
"vipType": "ignore",
"vipValue": "false",
"vipVariableName": "vpn_layer4",
}
},
"nat64-global": {"prefix": {"stateful": {}}},
"nat64": {
"v4": {
"pool": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["name"],
}
}
},
"nat": {
"natpool": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["name"],
},
"port-forward": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["source-port", "translate-port"],
},
"static": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["source-ip", "translate-ip"],
},
},
"route-import": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["protocol"],
},
"route-export": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["protocol"],
},
"dns": {
"vipType": "constant",
"vipValue": [
{
"role": {
"vipType": "constant",
"vipValue": "primary",
"vipObjectType": "object",
},
"dns-addr": {
"vipType": "variableName",
"vipValue": "",
"vipObjectType": "object",
"vipVariableName": "vpn_dns_primary",
},
"priority-order": ["dns-addr", "role"],
},
{
"role": {
"vipType": "constant",
"vipValue": "secondary",
"vipObjectType": "object",
},
"dns-addr": {
"vipType": "variableName",
"vipValue": "",
"vipObjectType": "object",
"vipVariableName": "vpn_dns_secondary",
},
"priority-order": ["dns-addr", "role"],
},
],
"vipObjectType": "tree",
"vipPrimaryKey": ["dns-addr"],
},
"host": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["hostname"],
},
"service": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["svc-type"],
},
"ip": {
"route": {
"vipType": "constant",
"vipValue": [
{
"prefix": {
"vipObjectType": "object",
"vipType": "constant",
"vipValue": "0.0.0.0/0",
"vipVariableName": "vpn_ipv4_ip_prefix",
},
"next-hop": {
"vipType": "constant",
"vipValue": [
{
"address": {
"vipObjectType": "object",
"vipType": "variableName",
"vipValue": "",
"vipVariableName": "vpn_next_hop_ip_address_0",
},
"distance": {
"vipObjectType": "object",
"vipType": "ignore",
"vipValue": 1,
"vipVariableName": "vpn_next_hop_ip_distance_0",
},
"priority-order": ["address", "distance"],
}
],
"vipObjectType": "tree",
"vipPrimaryKey": ["address"],
},
"priority-order": ["prefix", "next-hop", "next-hop-with-track"],
}
],
"vipObjectType": "tree",
"vipPrimaryKey": ["prefix"],
},
"gre-route": {},
"ipsec-route": {},
"service-route": {},
},
"ipv6": {},
"omp": {
"advertise": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["protocol"],
},
"ipv6-advertise": {
"vipType": "ignore",
"vipValue": [],
"vipObjectType": "tree",
"vipPrimaryKey": ["protocol"],
},
},
}
expected_values = {
"vpn-id": 0,
"dns": [
{"role": "primary", "dns-addr": DeviceVariable(name="vpn_dns_primary")},
{"role": "secondary", "dns-addr": DeviceVariable(name="vpn_dns_secondary")},
],
"ip": {
"route": [
{
"prefix": "0.0.0.0/0",
"next-hop": [
{
"address": DeviceVariable(name="vpn_next_hop_ip_address_0"),
}
],
}
],
},
}
# Act
result = find_template_values(input_values)
# Assert
assert expected_values == result
10 changes: 10 additions & 0 deletions catalystwan/tests/test_response.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from attr import define, field # type: ignore
from parameterized import parameterized # type: ignore
from pydantic import BaseModel, Field, ValidationError
from requests.exceptions import JSONDecodeError

from catalystwan.dataclasses import DataclassBase
from catalystwan.response import ManagerErrorInfo, ManagerResponse
Expand Down Expand Up @@ -171,3 +172,12 @@ def test_dataseq_optional_validate(self):
assert data.important == VALIDATE_DATASEQ_TEST_DATA[i]["important"]
with self.assertRaises(ValidationError):
vmng_response.dataseq(DataForValidateTest, sourcekey=None, validate=True)

def test_dataseq_with_misisng_data(self):
self.response_mock.json.side_effect = JSONDecodeError("test", "test", 1)
vmng_response = ManagerResponse(self.response_mock)
# Act
dataseq = vmng_response.dataseq(DataForValidateTest, sourcekey="data", validate=True)
# Assert
assert isinstance(dataseq, DataSequence)
assert len(dataseq) == 0
35 changes: 18 additions & 17 deletions catalystwan/utils/feature_template/find_template_values.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ def find_template_values(
target_key: str = "vipType",
target_key_value_to_ignore: str = "ignore",
target_key_for_template_value: str = "vipValue",
device_specific_variables: Optional[Dict[str, DeviceVariable]] = None,
path: Optional[List[str]] = None,
) -> Dict[str, Union[str, list, dict]]:
"""Based on provided template definition generates a dictionary with template fields and values
Expand All @@ -38,22 +37,28 @@ def find_template_values(
if template_definition[target_key] == target_key_value_to_ignore:
return templated_values

value = template_definition[target_key]
value = template_definition[target_key] # vipType
template_value = template_definition.get(target_key_for_template_value)

field_key = path[-1]
# TODO: Handle nested DeviceVariable
if value == "variableName":
if device_specific_variables is not None:
device_specific_variables[field_key] = DeviceVariable(name=template_definition["vipVariableName"])
return template_definition
# For example this is the current dictionary:
# field_key is "dns-addr"
# {
# "vipType": "variableName",
# "vipValue": "",
# "vipObjectType": "object",
# "vipVariableName": "vpn_dns_primary",
# }
# vipType is "variableName" so we need to return
# {"dns-addr": DeviceVariable(name="vpn_dns_primary")}
templated_values[field_key] = DeviceVariable(name=template_definition["vipVariableName"])
return templated_values

if template_value is None:
return template_definition

if template_definition["vipType"] == "variable":
if device_specific_variables is not None and template_value:
device_specific_variables[field_key] = DeviceVariable(name=template_value)
elif template_definition["vipObjectType"] == "list":
if template_definition["vipObjectType"] == "list":
current_nesting = get_nested_dict(templated_values, path[:-1])
current_nesting[field_key] = []
for item in template_value:
Expand All @@ -62,24 +67,20 @@ def find_template_values(
current_nesting = get_nested_dict(templated_values, path[:-1])
current_nesting[field_key] = template_value
elif isinstance(template_value, dict):
find_template_values(
value, templated_values, device_specific_variables=device_specific_variables, path=path
)
find_template_values(value, templated_values, path=path)
elif isinstance(template_value, list):
current_nesting = get_nested_dict(templated_values, path[:-1])
current_nesting[field_key] = []
for item in template_value:
item_value = find_template_values(item, {}, device_specific_variables=device_specific_variables)
item_value = find_template_values(item, {})
if item_value:
current_nesting[field_key].append(item_value)
return templated_values

# iterate the dict to extract values and assign them to their fields
for key, value in template_definition.items():
if isinstance(value, dict) and value != target_key_value_to_ignore:
find_template_values(
value, templated_values, device_specific_variables=device_specific_variables, path=path + [key]
)
find_template_values(value, templated_values, path=path + [key])
return templated_values


Expand Down

0 comments on commit 8cccb61

Please sign in to comment.