diff --git a/catalystwan/api/templates/feature_template.py b/catalystwan/api/templates/feature_template.py index 02a9141b..3fcf1312 100644 --- a/catalystwan/api/templates/feature_template.py +++ b/catalystwan/api/templates/feature_template.py @@ -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, ) diff --git a/catalystwan/exceptions.py b/catalystwan/exceptions.py index 7ebdc468..04dc351b 100644 --- a/catalystwan/exceptions.py +++ b/catalystwan/exceptions.py @@ -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): diff --git a/catalystwan/response.py b/catalystwan/response.py index 1524be66..442185ae 100644 --- a/catalystwan/response.py +++ b/catalystwan/response.py @@ -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 @@ -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""" @@ -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: diff --git a/catalystwan/tests/templates/test_find_template_values.py b/catalystwan/tests/templates/test_find_template_values.py new file mode 100644 index 00000000..50702369 --- /dev/null +++ b/catalystwan/tests/templates/test_find_template_values.py @@ -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 diff --git a/catalystwan/tests/test_response.py b/catalystwan/tests/test_response.py index 54af1ce1..09f0b059 100644 --- a/catalystwan/tests/test_response.py +++ b/catalystwan/tests/test_response.py @@ -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 @@ -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 diff --git a/catalystwan/utils/feature_template/find_template_values.py b/catalystwan/utils/feature_template/find_template_values.py index 6969b9c1..10d74ef7 100644 --- a/catalystwan/utils/feature_template/find_template_values.py +++ b/catalystwan/utils/feature_template/find_template_values.py @@ -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 @@ -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: @@ -62,14 +67,12 @@ 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 @@ -77,9 +80,7 @@ def find_template_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