Skip to content
Snippets Groups Projects
Commit 01bbcbfc authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Common - Type Checkers: 

- Completed assertions for slices
- Added assertions for Constraint actions
- Added missing values in enums
- Added assertions for non-custom constraints
- Added field "name" in multiple objects
- Added assertions for device components
- Added assertions for link attributes
- Added assertions for connection settings
- Minor cosmetic improvements
parent 87e4ef0d
No related branches found
No related tags found
2 merge requests!235Release TeraFlowSDN 3.0,!195Resolve "(CTTC) NBI unitary tests not running and/or not working"
Hide whitespace changes
Inline Side-by-side
src/common/type_checkers/Assertions.py
+ 228
− 31
View file @ 01bbcbfc
...@@ -12,7 +12,10 @@ ...@@ -12,7 +12,10 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from typing import Dict import logging
from typing import Callable, Dict
LOGGER = logging.getLogger(__name__)
# ----- Enumerations --------------------------------------------------------------------------------------------------- # ----- Enumerations ---------------------------------------------------------------------------------------------------
def validate_config_action_enum(message): def validate_config_action_enum(message):
...@@ -23,6 +26,14 @@ def validate_config_action_enum(message): ...@@ -23,6 +26,14 @@ def validate_config_action_enum(message):
'CONFIGACTION_DELETE', 'CONFIGACTION_DELETE',
] ]
def validate_constraint_action_enum(message):
assert isinstance(message, str)
assert message in [
'CONSTRAINTACTION_UNDEFINED',
'CONSTRAINTACTION_SET',
'CONSTRAINTACTION_DELETE',
]
def validate_device_driver_enum(message): def validate_device_driver_enum(message):
assert isinstance(message, str) assert isinstance(message, str)
assert message in [ assert message in [
...@@ -35,6 +46,7 @@ def validate_device_driver_enum(message): ...@@ -35,6 +46,7 @@ def validate_device_driver_enum(message):
'DEVICEDRIVER_XR', 'DEVICEDRIVER_XR',
'DEVICEDRIVER_IETF_L2VPN', 'DEVICEDRIVER_IETF_L2VPN',
'DEVICEDRIVER_GNMI_OPENCONFIG', 'DEVICEDRIVER_GNMI_OPENCONFIG',
'DEVICEDRIVER_FLEXSCALE',
] ]
def validate_device_operational_status_enum(message): def validate_device_operational_status_enum(message):
...@@ -64,6 +76,8 @@ def validate_service_type_enum(message): ...@@ -64,6 +76,8 @@ def validate_service_type_enum(message):
'SERVICETYPE_L3NM', 'SERVICETYPE_L3NM',
'SERVICETYPE_L2NM', 'SERVICETYPE_L2NM',
'SERVICETYPE_TAPI_CONNECTIVITY_SERVICE', 'SERVICETYPE_TAPI_CONNECTIVITY_SERVICE',
'SERVICETYPE_TE',
'SERVICETYPE_E2E',
] ]
def validate_service_state_enum(message): def validate_service_state_enum(message):
...@@ -77,6 +91,17 @@ def validate_service_state_enum(message): ...@@ -77,6 +91,17 @@ def validate_service_state_enum(message):
'SERVICESTATUS_SLA_VIOLATED', 'SERVICESTATUS_SLA_VIOLATED',
] ]
def validate_slice_status_enum(message):
assert isinstance(message, str)
assert message in [
'SLICESTATUS_UNDEFINED',
'SLICESTATUS_PLANNED',
'SLICESTATUS_INIT',
'SLICESTATUS_ACTIVE',
'SLICESTATUS_DEINIT',
'SLICESTATUS_SLA_VIOLATED',
]
# ----- Common --------------------------------------------------------------------------------------------------------- # ----- Common ---------------------------------------------------------------------------------------------------------
def validate_uuid(message, allow_empty=False): def validate_uuid(message, allow_empty=False):
...@@ -114,28 +139,61 @@ def validate_config_rules(message): ...@@ -114,28 +139,61 @@ def validate_config_rules(message):
assert 'config_rules' in message assert 'config_rules' in message
for config_rule in message['config_rules']: validate_config_rule(config_rule) for config_rule in message['config_rules']: validate_config_rule(config_rule)
CONSTRAINT_TYPES = { def validate_constraint_custom(message):
'custom', assert isinstance(message, dict)
'schedule', assert len(message.keys()) == 2
'endpoint_location', assert 'constraint_type' in message
'sla_capacity', assert isinstance(message['constraint_type'], str)
'sla_latency', assert 'constraint_value' in message
'sla_availability', assert isinstance(message['constraint_value'], str)
'sla_isolation',
def validate_constraint_sla_capacity(message):
assert isinstance(message, dict)
assert len(message.keys()) == 1
assert 'capacity_gbps' in message
assert isinstance(message['capacity_gbps'], (int, float))
def validate_constraint_sla_latency(message):
assert isinstance(message, dict)
assert len(message.keys()) == 1
assert 'e2e_latency_ms' in message
assert isinstance(message['e2e_latency_ms'], (int, float))
def validate_constraint_sla_availability(message):
assert isinstance(message, dict)
assert len(message.keys()) == 3
assert 'num_disjoint_paths' in message
assert isinstance(message['num_disjoint_paths'], int)
assert message['num_disjoint_paths'] >= 0
assert 'all_active' in message
assert isinstance(message['all_active'], bool)
assert 'availability' in message
assert isinstance(message['availability'], (int, float))
assert message['availability'] >= 0 and message['availability'] <= 100
CONSTRAINT_TYPE_TO_VALIDATOR = {
'custom' : validate_constraint_custom,
#'schedule' : validate_constraint_schedule,
#'endpoint_location' : validate_constraint_endpoint_location,
#'endpoint_priority' : validate_constraint_endpoint_priority,
'sla_capacity' : validate_constraint_sla_capacity,
'sla_latency' : validate_constraint_sla_latency,
'sla_availability' : validate_constraint_sla_availability,
#'sla_isolation' : validate_constraint_sla_isolation,
#'exclusions' : validate_constraint_exclusions,
} }
def validate_constraint(message): def validate_constraint(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 1 assert len(message.keys()) == 2
other_keys = list(message.keys()) assert 'action' in message
constraint_type = other_keys[0] validate_constraint_action_enum(message['action'])
assert constraint_type in CONSTRAINT_TYPES other_keys = set(list(message.keys()))
assert constraint_type == 'custom', 'Constraint Type Validator for {:s} not implemented'.format(constraint_type) other_keys.discard('action')
custom : Dict = message['custom'] constraint_type = other_keys.pop()
assert len(custom.keys()) == 2 validator : Callable = CONSTRAINT_TYPE_TO_VALIDATOR.get(constraint_type)
assert 'constraint_type' in custom assert validator is not None, 'Constraint Type Validator for {:s} not implemented'.format(constraint_type)
assert isinstance(custom['constraint_type'], str) validator(message[constraint_type])
assert 'constraint_value' in custom
assert isinstance(custom['constraint_value'], str)
# ----- Identifiers ---------------------------------------------------------------------------------------------------- # ----- Identifiers ----------------------------------------------------------------------------------------------------
...@@ -218,6 +276,13 @@ def validate_service_ids(message, context_uuid=None): ...@@ -218,6 +276,13 @@ def validate_service_ids(message, context_uuid=None):
assert isinstance(message['service_ids'], list) assert isinstance(message['service_ids'], list)
for service_id in message['service_ids']: validate_service_id(service_id, context_uuid=context_uuid) for service_id in message['service_ids']: validate_service_id(service_id, context_uuid=context_uuid)
def validate_slice_ids(message, context_uuid=None):
assert isinstance(message, dict)
assert len(message.keys()) == 1
assert 'slice_ids' in message
assert isinstance(message['slice_ids'], list)
for slice_id in message['slice_ids']: validate_slice_id(slice_id, context_uuid=context_uuid)
def validate_topology_ids(message, context_uuid=None): def validate_topology_ids(message, context_uuid=None):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 1 assert len(message.keys()) == 1
...@@ -252,17 +317,17 @@ def validate_connection_ids(message): ...@@ -252,17 +317,17 @@ def validate_connection_ids(message):
def validate_context(message): def validate_context(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 5 assert len(message.keys()) == 5
assert 'name' in message
assert isinstance(message['name'], str)
assert 'context_id' in message assert 'context_id' in message
validate_context_id(message['context_id']) validate_context_id(message['context_id'])
context_uuid = message['context_id']['context_uuid']['uuid'] context_uuid = message['context_id']['context_uuid']['uuid']
assert 'service_ids' in message assert 'name' in message
assert isinstance(message['service_ids'], list) assert isinstance(message['name'], str)
for service_id in message['service_ids']: validate_service_id(service_id, context_uuid=context_uuid)
assert 'topology_ids' in message assert 'topology_ids' in message
assert isinstance(message['topology_ids'], list) assert isinstance(message['topology_ids'], list)
for topology_id in message['topology_ids']: validate_topology_id(topology_id, context_uuid=context_uuid) for topology_id in message['topology_ids']: validate_topology_id(topology_id, context_uuid=context_uuid)
assert 'service_ids' in message
assert isinstance(message['service_ids'], list)
for service_id in message['service_ids']: validate_service_id(service_id, context_uuid=context_uuid)
assert 'slice_ids' in message assert 'slice_ids' in message
assert isinstance(message['slice_ids'], list) assert isinstance(message['slice_ids'], list)
for slice_id in message['slice_ids']: validate_slice_id(slice_id, context_uuid=context_uuid) for slice_id in message['slice_ids']: validate_slice_id(slice_id, context_uuid=context_uuid)
...@@ -273,11 +338,19 @@ def validate_service_state(message): ...@@ -273,11 +338,19 @@ def validate_service_state(message):
assert 'service_status' in message assert 'service_status' in message
validate_service_state_enum(message['service_status']) validate_service_state_enum(message['service_status'])
def validate_slice_status(message):
assert isinstance(message, dict)
assert len(message.keys()) == 1
assert 'slice_status' in message
validate_slice_status_enum(message['slice_status'])
def validate_service(message): def validate_service(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 6 assert len(message.keys()) == 7
assert 'service_id' in message assert 'service_id' in message
validate_service_id(message['service_id']) validate_service_id(message['service_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'service_type' in message assert 'service_type' in message
validate_service_type_enum(message['service_type']) validate_service_type_enum(message['service_type'])
assert 'service_endpoint_ids' in message assert 'service_endpoint_ids' in message
...@@ -291,11 +364,44 @@ def validate_service(message): ...@@ -291,11 +364,44 @@ def validate_service(message):
assert 'service_config' in message assert 'service_config' in message
validate_config_rules(message['service_config']) validate_config_rules(message['service_config'])
def validate_slice(message):
assert isinstance(message, dict)
assert len(message.keys()) in {8, 9}
assert 'slice_id' in message
validate_slice_id(message['slice_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'slice_endpoint_ids' in message
assert isinstance(message['slice_endpoint_ids'], list)
for endpoint_id in message['slice_endpoint_ids']: validate_endpoint_id(endpoint_id)
assert 'slice_constraints' in message
assert isinstance(message['slice_constraints'], list)
for constraint in message['slice_constraints']: validate_constraint(constraint)
assert 'slice_service_ids' in message
assert isinstance(message['slice_service_ids'], list)
for service_id in message['slice_service_ids']: validate_service_id(service_id)
assert 'slice_subslice_ids' in message
assert isinstance(message['slice_subslice_ids'], list)
for slice_id in message['slice_subslice_ids']: validate_slice_id(slice_id)
assert 'slice_status' in message
validate_slice_status(message['slice_status'])
assert 'slice_config' in message
validate_config_rules(message['slice_config'])
if len(message.keys()) == 9:
assert 'slice_owner' in message
assert isinstance(message['slice_owner'], dict)
assert 'owner_uuid' in message['slice_owner']
validate_uuid(message['slice_owner']['owner_uuid'])
assert 'owner_string' in message['slice_owner']
assert isinstance(message['slice_owner']['owner_string'], str)
def validate_topology(message, num_devices=None, num_links=None): def validate_topology(message, num_devices=None, num_links=None):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 3 assert len(message.keys()) == 4
assert 'topology_id' in message assert 'topology_id' in message
validate_topology_id(message['topology_id']) validate_topology_id(message['topology_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'device_ids' in message assert 'device_ids' in message
assert isinstance(message['device_ids'], list) assert isinstance(message['device_ids'], list)
if num_devices is not None: assert len(message['device_ids']) == num_devices if num_devices is not None: assert len(message['device_ids']) == num_devices
...@@ -307,20 +413,49 @@ def validate_topology(message, num_devices=None, num_links=None): ...@@ -307,20 +413,49 @@ def validate_topology(message, num_devices=None, num_links=None):
def validate_endpoint(message): def validate_endpoint(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 3 assert len(message.keys()) == 4
assert 'endpoint_id' in message assert 'endpoint_id' in message
validate_endpoint_id(message['endpoint_id']) validate_endpoint_id(message['endpoint_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'endpoint_type' in message assert 'endpoint_type' in message
assert isinstance(message['endpoint_type'], str) assert isinstance(message['endpoint_type'], str)
assert 'kpi_sample_types' in message assert 'kpi_sample_types' in message
assert isinstance(message['kpi_sample_types'], list) assert isinstance(message['kpi_sample_types'], list)
for kpi_sample_type in message['kpi_sample_types']: validate_kpi_sample_types_enum(kpi_sample_type) for kpi_sample_type in message['kpi_sample_types']: validate_kpi_sample_types_enum(kpi_sample_type)
def validate_component(component):
assert isinstance(component, dict)
assert len(component.keys()) == 5
assert 'component_uuid' in component
validate_uuid(component['component_uuid'])
assert 'name' in component
assert isinstance(component['name'], str)
assert 'type' in component
assert isinstance(component['type'], str)
assert 'attributes' in component
assert isinstance(component['attributes'], dict)
for k,v in component['attributes'].items():
assert isinstance(k, str)
assert isinstance(v, str)
assert 'parent' in component
assert isinstance(component['parent'], str)
def validate_link_attributes(link_attributes):
assert isinstance(link_attributes, dict)
assert len(link_attributes.keys()) == 2
assert 'total_capacity_gbps' in link_attributes
assert isinstance(link_attributes['total_capacity_gbps'], (int, float))
assert 'used_capacity_gbps' in link_attributes
assert isinstance(link_attributes['used_capacity_gbps'], (int, float))
def validate_device(message): def validate_device(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 6 assert len(message.keys()) in {8, 9}
assert 'device_id' in message assert 'device_id' in message
validate_device_id(message['device_id']) validate_device_id(message['device_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'device_type' in message assert 'device_type' in message
assert isinstance(message['device_type'], str) assert isinstance(message['device_type'], str)
assert 'device_config' in message assert 'device_config' in message
...@@ -333,19 +468,30 @@ def validate_device(message): ...@@ -333,19 +468,30 @@ def validate_device(message):
assert 'device_endpoints' in message assert 'device_endpoints' in message
assert isinstance(message['device_endpoints'], list) assert isinstance(message['device_endpoints'], list)
for endpoint in message['device_endpoints']: validate_endpoint(endpoint) for endpoint in message['device_endpoints']: validate_endpoint(endpoint)
assert 'components' in message
assert isinstance(message['components'], list)
for component in message['components']: validate_component(component)
if len(message.keys()) == 9:
assert 'controller_id' in message
if len(message['controller_id']) > 0:
validate_device_id(message['controller_id'])
def validate_link(message): def validate_link(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 2 assert len(message.keys()) == 4
assert 'link_id' in message assert 'link_id' in message
validate_link_id(message['link_id']) validate_link_id(message['link_id'])
assert 'name' in message
assert isinstance(message['name'], str)
assert 'link_endpoint_ids' in message assert 'link_endpoint_ids' in message
assert isinstance(message['link_endpoint_ids'], list) assert isinstance(message['link_endpoint_ids'], list)
for endpoint_id in message['link_endpoint_ids']: validate_endpoint_id(endpoint_id) for endpoint_id in message['link_endpoint_ids']: validate_endpoint_id(endpoint_id)
assert 'attributes' in message
validate_link_attributes(message['attributes'])
def validate_connection(message): def validate_connection(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 4 assert len(message.keys()) in {4, 5}
assert 'connection_id' in message assert 'connection_id' in message
validate_connection_id(message['connection_id']) validate_connection_id(message['connection_id'])
assert 'service_id' in message assert 'service_id' in message
...@@ -356,6 +502,50 @@ def validate_connection(message): ...@@ -356,6 +502,50 @@ def validate_connection(message):
assert 'sub_service_ids' in message assert 'sub_service_ids' in message
assert isinstance(message['sub_service_ids'], list) assert isinstance(message['sub_service_ids'], list)
for sub_service_id in message['sub_service_ids']: validate_service_id(sub_service_id) for sub_service_id in message['sub_service_ids']: validate_service_id(sub_service_id)
if len(message.keys()) == 5:
assert 'settings' in message
assert isinstance(message['settings'], dict)
# TODO: improve validation of data types, especially for uint values, IP/MAC addresses, TCP/UDP ports, etc.
if 'l0' in message['settings']:
assert isinstance(message['settings']['l0'], dict)
if 'lsp_symbolic_name' in message['settings']['l0']:
assert isinstance(message['settings']['l0']['lsp_symbolic_name'], str)
if 'l2' in message['settings']:
assert isinstance(message['settings']['l2'], dict)
if 'src_mac_address' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['src_mac_address'], str)
if 'dst_mac_address' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['dst_mac_address'], str)
if 'ether_type' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['ether_type'], int)
if 'vlan_id' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['vlan_id'], int)
if 'mpls_label' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['mpls_label'], int)
if 'mpls_traffic_class' in message['settings']['l2']:
assert isinstance(message['settings']['l2']['mpls_traffic_class'], int)
if 'l3' in message['settings']:
assert isinstance(message['settings']['l3'], dict)
if 'src_ip_address' in message['settings']['l3']:
assert isinstance(message['settings']['l3']['src_ip_address'], str)
if 'dst_ip_address' in message['settings']['l3']:
assert isinstance(message['settings']['l3']['dst_ip_address'], str)
if 'dscp' in message['settings']['l3']:
assert isinstance(message['settings']['l3']['dscp'], int)
if 'protocol' in message['settings']['l3']:
assert isinstance(message['settings']['l3']['protocol'], int)
if 'ttl' in message['settings']['l3']:
assert isinstance(message['settings']['l3']['ttl'], int)
if 'l4' in message['settings']:
assert isinstance(message['settings']['l4'], dict)
if 'src_port' in message['settings']['l4']:
assert isinstance(message['settings']['l4']['src_port'], int)
if 'dst_port' in message['settings']['l4']:
assert isinstance(message['settings']['l4']['dst_port'], int)
if 'tcp_flags' in message['settings']['l4']:
assert isinstance(message['settings']['l4']['tcp_flags'], int)
if 'ttl' in message['settings']['l4']:
assert isinstance(message['settings']['l4']['ttl'], int)
# ----- Lists of Objects ----------------------------------------------------------------------------------------------- # ----- Lists of Objects -----------------------------------------------------------------------------------------------
...@@ -374,6 +564,13 @@ def validate_services(message): ...@@ -374,6 +564,13 @@ def validate_services(message):
assert isinstance(message['services'], list) assert isinstance(message['services'], list)
for service in message['services']: validate_service(service) for service in message['services']: validate_service(service)
def validate_slices(message):
assert isinstance(message, dict)
assert len(message.keys()) == 1
assert 'slices' in message
assert isinstance(message['slices'], list)
for slice_ in message['slices']: validate_slice(slice_)
def validate_topologies(message): def validate_topologies(message):
assert isinstance(message, dict) assert isinstance(message, dict)
assert len(message.keys()) == 1 assert len(message.keys()) == 1
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment