diff --git a/src/common/tools/object_factory/Device.py b/src/common/tools/object_factory/Device.py index 66c87b14dd866d44b5d48addf93d172aea962f8e..032aa4fb2781c175eb6f4421055143937750f398 100644 --- a/src/common/tools/object_factory/Device.py +++ b/src/common/tools/object_factory/Device.py @@ -13,7 +13,7 @@ # limitations under the License. import copy -from typing import Dict, List, Tuple +from typing import Dict, List, Optional, Tuple from common.DeviceTypes import DeviceTypeEnum from common.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusEnum from common.tools.object_factory.ConfigRule import json_config_rule_set @@ -50,10 +50,10 @@ def json_device_id(device_uuid : str): return {'device_uuid': {'uuid': device_uuid}} def json_device( - device_uuid : str, device_type : str, status : DeviceOperationalStatusEnum, endpoints : List[Dict] = [], - config_rules : List[Dict] = [], drivers : List[Dict] = [] + device_uuid : str, device_type : str, status : DeviceOperationalStatusEnum, name : Optional[str] = None, + endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = [] ): - return { + result = { 'device_id' : json_device_id(device_uuid), 'device_type' : device_type, 'device_config' : {'config_rules': copy.deepcopy(config_rules)}, @@ -61,74 +61,80 @@ def json_device( 'device_drivers' : copy.deepcopy(drivers), 'device_endpoints' : copy.deepcopy(endpoints), } + if name is not None: result['name'] = name + return result def json_device_emulated_packet_router_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_EMU_DRIVERS ): return json_device( - device_uuid, DEVICE_EMUPR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, + device_uuid, DEVICE_EMUPR_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_emulated_tapi_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_EMU_DRIVERS ): return json_device( - device_uuid, DEVICE_EMUOLS_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, + device_uuid, DEVICE_EMUOLS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_emulated_datacenter_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_EMU_DRIVERS ): return json_device( - device_uuid, DEVICE_EMUDC_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, + device_uuid, DEVICE_EMUDC_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_packetrouter_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_PR_DRIVERS ): return json_device( - device_uuid, DEVICE_PR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) + device_uuid, DEVICE_PR_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, + drivers=drivers) def json_device_tapi_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_TAPI_DRIVERS ): return json_device( - device_uuid, DEVICE_TAPI_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) + device_uuid, DEVICE_TAPI_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, + drivers=drivers) def json_device_xr_constellation_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_XR_CONSTELLATION_DRIVERS ): return json_device( - device_uuid, DEVICE_XR_CONSTELLATION_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, - drivers=drivers) + device_uuid, DEVICE_XR_CONSTELLATION_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, + config_rules=config_rules, drivers=drivers) def json_device_microwave_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_MICROWAVE_DRIVERS ): return json_device( - device_uuid, DEVICE_MICROWAVE_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, + device_uuid, DEVICE_MICROWAVE_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_p4_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_P4_DRIVERS ): return json_device( - device_uuid, DEVICE_P4_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) + device_uuid, DEVICE_P4_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, + drivers=drivers) def json_device_tfs_disabled( - device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], + device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_TFS_DRIVERS ): return json_device( - device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) + device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, + drivers=drivers) def json_device_connect_rules(address : str, port : int, settings : Dict = {}): return [ diff --git a/src/common/tools/object_factory/Link.py b/src/common/tools/object_factory/Link.py index dbb3d7fb15ac167e6b0a227d8bc8ab8002d93bba..5f8080d300d9d6d646b8d769ec5819b0bd26f789 100644 --- a/src/common/tools/object_factory/Link.py +++ b/src/common/tools/object_factory/Link.py @@ -13,7 +13,7 @@ # limitations under the License. import copy -from typing import Dict, List, Tuple +from typing import Dict, List, Optional, Tuple def get_link_uuid(a_endpoint_id : Dict, z_endpoint_id : Dict) -> str: return '{:s}/{:s}=={:s}/{:s}'.format( @@ -23,8 +23,10 @@ def get_link_uuid(a_endpoint_id : Dict, z_endpoint_id : Dict) -> str: def json_link_id(link_uuid : str) -> Dict: return {'link_uuid': {'uuid': link_uuid}} -def json_link(link_uuid : str, endpoint_ids : List[Dict]) -> Dict: - return {'link_id': json_link_id(link_uuid), 'link_endpoint_ids': copy.deepcopy(endpoint_ids)} +def json_link(link_uuid : str, endpoint_ids : List[Dict], name : Optional[str] = None) -> Dict: + result = {'link_id': json_link_id(link_uuid), 'link_endpoint_ids': copy.deepcopy(endpoint_ids)} + if name is not None: result['name'] = name + return result def compose_link(endpoint_a, endpoint_z) -> Tuple[Dict, Dict]: link_uuid = get_link_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id'])