diff --git a/src/common/tests/LoadScenario.py b/src/common/tests/LoadScenario.py
index 1c531ed6018a9fbb1850992b8d9b8a71c2ed0007..3c3940e67b5772f3ba3ec0634c49f26b92bbc571 100644
--- a/src/common/tests/LoadScenario.py
+++ b/src/common/tests/LoadScenario.py
@@ -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
@@ -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):
@@ -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
diff --git a/src/common/tools/descriptor/Loader.py b/src/common/tools/descriptor/Loader.py
index 2674cdd3e8bf18077063d52b9a71857baa2bb374..468cd4851f50acebfc654267fd7fc537a65f2691 100644
--- a/src/common/tools/descriptor/Loader.py
+++ b/src/common/tools/descriptor/Loader.py
@@ -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
@@ -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.__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.__topologies_add = None
+ self.__devices_add = None
+ self.__devices_config = None
+ self.__services_add = None
+ self.__slices_add = 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.__contexts_add = None
- self.__devices = 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
-
- 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]
@@ -130,6 +194,8 @@ class DescriptorLoader:
self._dummy_mode()
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.