Loading src/common/tests/LoadScenario.py +7 −5 Original line number Diff line number Diff line Loading @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, logging import logging from common.tools.descriptor.Loader import DescriptorLoader, compose_notifications from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient Loading @@ -29,15 +29,15 @@ LOGGERS = { def load_scenario_from_descriptor( descriptor_file : str, context_client : ContextClient, device_client : DeviceClient, service_client : ServiceClient, slice_client : SliceClient ) -> None: ) -> DescriptorLoader: with open(descriptor_file, 'r', encoding='UTF-8') as f: descriptors = json.loads(f.read()) descriptors = f.read() descriptor_loader = DescriptorLoader( descriptors, context_client=context_client, device_client=device_client, service_client=service_client, slice_client=slice_client) descriptor_loader.process_descriptors(descriptors) results = descriptor_loader.get_results() results = descriptor_loader.process() num_errors = 0 for message,level in compose_notifications(results): Loading @@ -46,3 +46,5 @@ def load_scenario_from_descriptor( if num_errors > 0: MSG = 'Failed to load descriptors in file {:s}' raise Exception(MSG.format(str(descriptor_file))) return descriptor_loader No newline at end of file src/common/tools/descriptor/Loader.py +102 −36 Original line number Diff line number Diff line Loading @@ -16,25 +16,23 @@ # Usage example (WebUI): # descriptors = json.loads(descriptors_data_from_client) # # descriptor_loader = DescriptorLoader() # descriptor_loader.process_descriptors(descriptors) # results = descriptor_loader.get_results() # descriptor_loader = DescriptorLoader(descriptors) # results = descriptor_loader.process() # for message,level in compose_notifications(results): # flash(message, level) # Usage example (pytest): # with open('path/to/descriptor.json', 'r', encoding='UTF-8') as f: # descriptors = json.loads(f.read()) # # descriptor_loader = DescriptorLoader() # descriptor_loader.process_descriptors(descriptors) # results = descriptor_loader.get_results() # descriptor_loader = DescriptorLoader( # descriptors, context_client=..., device_client=..., service_client=..., slice_client=...) # results = descriptor_loader.process() # loggers = {'success': LOGGER.info, 'danger': LOGGER.error, 'error': LOGGER.error} # for message,level in compose_notifications(results): # loggers.get(level)(message) from typing import Dict, List, Optional, Tuple import json from typing import Dict, List, Optional, Tuple, Union from common.proto.context_pb2 import Connection, Context, Device, Link, Service, Slice, Topology from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient Loading Loading @@ -81,41 +79,107 @@ def compose_notifications(results : TypeResults) -> TypeNotificationList: class DescriptorLoader: def __init__( self, context_client : Optional[ContextClient] = None, device_client : Optional[DeviceClient] = None, self, descriptors : Union[str, Dict], context_client : Optional[ContextClient] = None, device_client : Optional[DeviceClient] = None, service_client : Optional[ServiceClient] = None, slice_client : Optional[SliceClient] = None ) -> None: self.__ctx_cli = ContextClient() if context_client is None else context_client self.__dev_cli = DeviceClient() if device_client is None else device_client self.__svc_cli = ServiceClient() if service_client is None else service_client self.__slc_cli = SliceClient() if slice_client is None else slice_client self.__results : TypeResults = list() self.__connections = None self.__contexts = None self.__descriptors = json.loads(descriptors) if isinstance(descriptors, str) else descriptors self.__dummy_mode = self.__descriptors.get('dummy_mode' , False) self.__contexts = self.__descriptors.get('contexts' , []) self.__topologies = self.__descriptors.get('topologies' , []) self.__devices = self.__descriptors.get('devices' , []) self.__links = self.__descriptors.get('links' , []) self.__services = self.__descriptors.get('services' , []) self.__slices = self.__descriptors.get('slices' , []) self.__connections = self.__descriptors.get('connections', []) self.__contexts_add = None self.__devices = None self.__topologies_add = None self.__devices_add = None self.__devices_config = None self.__dummy_mode = None self.__links = None self.__services = None self.__services_add = None self.__slices = None self.__slices_add = None self.__topologies = None self.__topologies_add = None def get_results(self) -> TypeResults: return self.__results self.__ctx_cli = ContextClient() if context_client is None else context_client self.__dev_cli = DeviceClient() if device_client is None else device_client self.__svc_cli = ServiceClient() if service_client is None else service_client self.__slc_cli = SliceClient() if slice_client is None else slice_client def process_descriptors(self, descriptors : Dict) -> None: self.__dummy_mode = descriptors.get('dummy_mode' , False) self.__contexts = descriptors.get('contexts' , []) self.__topologies = descriptors.get('topologies' , []) self.__devices = descriptors.get('devices' , []) self.__links = descriptors.get('links' , []) self.__services = descriptors.get('services' , []) self.__slices = descriptors.get('slices' , []) self.__connections = descriptors.get('connections', []) self.__results : TypeResults = list() @property def contexts(self) -> List[Dict]: return self.__contexts @property def num_contexts(self) -> int: return len(self.__contexts) @property def topologies(self) -> Dict[str, List[Dict]]: _topologies = {} for topology in self.__topologies: context_uuid = topology.topology_id.context_id.context_uuid.uuid _topologies.setdefault(context_uuid, []).append(topology) return _topologies @property def num_topologies(self) -> Dict[str, int]: _num_topologies = {} for topology in self.__topologies: context_uuid = topology.topology_id.context_id.context_uuid.uuid _num_topologies[context_uuid] = _num_topologies.get(context_uuid, 0) + 1 return _num_topologies @property def devices(self) -> List[Dict]: return self.__devices @property def num_devices(self) -> int: return len(self.__devices) @property def links(self) -> List[Dict]: return self.__links @property def num_links(self) -> int: return len(self.__links) @property def services(self) -> Dict[str, List[Dict]]: _services = {} for service in self.__services: context_uuid = service.service_id.context_id.context_uuid.uuid _services.setdefault(context_uuid, []).append(service) return _services @property def num_services(self) -> Dict[str, int]: _num_services = {} for service in self.__services: context_uuid = service.service_id.context_id.context_uuid.uuid _num_services[context_uuid] = _num_services.get(context_uuid, 0) + 1 return _num_services @property def slices(self) -> Dict[str, List[Dict]]: _slices = {} for slice_ in self.__slices: context_uuid = slice_.slice_id.context_id.context_uuid.uuid _slices.setdefault(context_uuid, []).append(slice_) return _slices @property def num_slices(self) -> Dict[str, int]: _num_slices = {} for slice_ in self.__slices: context_uuid = slice_.slice_id.context_id.context_uuid.uuid _num_slices[context_uuid] = _num_slices.get(context_uuid, 0) + 1 return _num_slices @property def connections(self) -> List[Dict]: return self.__connections @property def num_connections(self) -> int: return len(self.__connections) def process(self) -> TypeResults: # Format CustomConfigRules in Devices, Services and Slices provided in JSON format self.__devices = [format_device_custom_config_rules (device ) for device in self.__devices ] self.__services = [format_service_custom_config_rules(service) for service in self.__services] Loading @@ -131,6 +195,8 @@ class DescriptorLoader: else: self._normal_mode() return self.__results def _dummy_mode(self) -> None: # Dummy Mode: used to pre-load databases (WebUI debugging purposes) with no smart or automated tasks. self.__ctx_cli.connect() Loading Loading
src/common/tests/LoadScenario.py +7 −5 Original line number Diff line number Diff line Loading @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, logging import logging from common.tools.descriptor.Loader import DescriptorLoader, compose_notifications from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient Loading @@ -29,15 +29,15 @@ LOGGERS = { def load_scenario_from_descriptor( descriptor_file : str, context_client : ContextClient, device_client : DeviceClient, service_client : ServiceClient, slice_client : SliceClient ) -> None: ) -> DescriptorLoader: with open(descriptor_file, 'r', encoding='UTF-8') as f: descriptors = json.loads(f.read()) descriptors = f.read() descriptor_loader = DescriptorLoader( descriptors, context_client=context_client, device_client=device_client, service_client=service_client, slice_client=slice_client) descriptor_loader.process_descriptors(descriptors) results = descriptor_loader.get_results() results = descriptor_loader.process() num_errors = 0 for message,level in compose_notifications(results): Loading @@ -46,3 +46,5 @@ def load_scenario_from_descriptor( if num_errors > 0: MSG = 'Failed to load descriptors in file {:s}' raise Exception(MSG.format(str(descriptor_file))) return descriptor_loader No newline at end of file
src/common/tools/descriptor/Loader.py +102 −36 Original line number Diff line number Diff line Loading @@ -16,25 +16,23 @@ # Usage example (WebUI): # descriptors = json.loads(descriptors_data_from_client) # # descriptor_loader = DescriptorLoader() # descriptor_loader.process_descriptors(descriptors) # results = descriptor_loader.get_results() # descriptor_loader = DescriptorLoader(descriptors) # results = descriptor_loader.process() # for message,level in compose_notifications(results): # flash(message, level) # Usage example (pytest): # with open('path/to/descriptor.json', 'r', encoding='UTF-8') as f: # descriptors = json.loads(f.read()) # # descriptor_loader = DescriptorLoader() # descriptor_loader.process_descriptors(descriptors) # results = descriptor_loader.get_results() # descriptor_loader = DescriptorLoader( # descriptors, context_client=..., device_client=..., service_client=..., slice_client=...) # results = descriptor_loader.process() # loggers = {'success': LOGGER.info, 'danger': LOGGER.error, 'error': LOGGER.error} # for message,level in compose_notifications(results): # loggers.get(level)(message) from typing import Dict, List, Optional, Tuple import json from typing import Dict, List, Optional, Tuple, Union from common.proto.context_pb2 import Connection, Context, Device, Link, Service, Slice, Topology from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient Loading Loading @@ -81,41 +79,107 @@ def compose_notifications(results : TypeResults) -> TypeNotificationList: class DescriptorLoader: def __init__( self, context_client : Optional[ContextClient] = None, device_client : Optional[DeviceClient] = None, self, descriptors : Union[str, Dict], context_client : Optional[ContextClient] = None, device_client : Optional[DeviceClient] = None, service_client : Optional[ServiceClient] = None, slice_client : Optional[SliceClient] = None ) -> None: self.__ctx_cli = ContextClient() if context_client is None else context_client self.__dev_cli = DeviceClient() if device_client is None else device_client self.__svc_cli = ServiceClient() if service_client is None else service_client self.__slc_cli = SliceClient() if slice_client is None else slice_client self.__results : TypeResults = list() self.__connections = None self.__contexts = None self.__descriptors = json.loads(descriptors) if isinstance(descriptors, str) else descriptors self.__dummy_mode = self.__descriptors.get('dummy_mode' , False) self.__contexts = self.__descriptors.get('contexts' , []) self.__topologies = self.__descriptors.get('topologies' , []) self.__devices = self.__descriptors.get('devices' , []) self.__links = self.__descriptors.get('links' , []) self.__services = self.__descriptors.get('services' , []) self.__slices = self.__descriptors.get('slices' , []) self.__connections = self.__descriptors.get('connections', []) self.__contexts_add = None self.__devices = None self.__topologies_add = None self.__devices_add = None self.__devices_config = None self.__dummy_mode = None self.__links = None self.__services = None self.__services_add = None self.__slices = None self.__slices_add = None self.__topologies = None self.__topologies_add = None def get_results(self) -> TypeResults: return self.__results self.__ctx_cli = ContextClient() if context_client is None else context_client self.__dev_cli = DeviceClient() if device_client is None else device_client self.__svc_cli = ServiceClient() if service_client is None else service_client self.__slc_cli = SliceClient() if slice_client is None else slice_client def process_descriptors(self, descriptors : Dict) -> None: self.__dummy_mode = descriptors.get('dummy_mode' , False) self.__contexts = descriptors.get('contexts' , []) self.__topologies = descriptors.get('topologies' , []) self.__devices = descriptors.get('devices' , []) self.__links = descriptors.get('links' , []) self.__services = descriptors.get('services' , []) self.__slices = descriptors.get('slices' , []) self.__connections = descriptors.get('connections', []) self.__results : TypeResults = list() @property def contexts(self) -> List[Dict]: return self.__contexts @property def num_contexts(self) -> int: return len(self.__contexts) @property def topologies(self) -> Dict[str, List[Dict]]: _topologies = {} for topology in self.__topologies: context_uuid = topology.topology_id.context_id.context_uuid.uuid _topologies.setdefault(context_uuid, []).append(topology) return _topologies @property def num_topologies(self) -> Dict[str, int]: _num_topologies = {} for topology in self.__topologies: context_uuid = topology.topology_id.context_id.context_uuid.uuid _num_topologies[context_uuid] = _num_topologies.get(context_uuid, 0) + 1 return _num_topologies @property def devices(self) -> List[Dict]: return self.__devices @property def num_devices(self) -> int: return len(self.__devices) @property def links(self) -> List[Dict]: return self.__links @property def num_links(self) -> int: return len(self.__links) @property def services(self) -> Dict[str, List[Dict]]: _services = {} for service in self.__services: context_uuid = service.service_id.context_id.context_uuid.uuid _services.setdefault(context_uuid, []).append(service) return _services @property def num_services(self) -> Dict[str, int]: _num_services = {} for service in self.__services: context_uuid = service.service_id.context_id.context_uuid.uuid _num_services[context_uuid] = _num_services.get(context_uuid, 0) + 1 return _num_services @property def slices(self) -> Dict[str, List[Dict]]: _slices = {} for slice_ in self.__slices: context_uuid = slice_.slice_id.context_id.context_uuid.uuid _slices.setdefault(context_uuid, []).append(slice_) return _slices @property def num_slices(self) -> Dict[str, int]: _num_slices = {} for slice_ in self.__slices: context_uuid = slice_.slice_id.context_id.context_uuid.uuid _num_slices[context_uuid] = _num_slices.get(context_uuid, 0) + 1 return _num_slices @property def connections(self) -> List[Dict]: return self.__connections @property def num_connections(self) -> int: return len(self.__connections) def process(self) -> TypeResults: # Format CustomConfigRules in Devices, Services and Slices provided in JSON format self.__devices = [format_device_custom_config_rules (device ) for device in self.__devices ] self.__services = [format_service_custom_config_rules(service) for service in self.__services] Loading @@ -131,6 +195,8 @@ class DescriptorLoader: else: self._normal_mode() return self.__results def _dummy_mode(self) -> None: # Dummy Mode: used to pre-load databases (WebUI debugging purposes) with no smart or automated tasks. self.__ctx_cli.connect() Loading
