import grpc, logging from typing import Dict, List, Set, Tuple from common.Checkers import chk_options, chk_string from common.database.api.Database import Database from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID from common.database.api.context.topology.device.Endpoint import Endpoint from common.database.api.context.topology.device.OperationalStatus import OperationalStatus, \ operationalstatus_enum_values, to_operationalstatus_enum from common.exceptions.ServiceException import ServiceException from common.tools.service.DeviceCheckers import check_device_endpoint_exists from common.tools.service.EndpointIdCheckers import check_endpoint_id from common.tools.service.EnumCheckers import check_enum from common.tools.service.DeviceCheckers import check_device_exists, check_device_not_exists from device.proto.context_pb2 import Device, DeviceId # For each method name, define acceptable device operational statuses. Empty set means accept all. ACCEPTED_DEVICE_OPERATIONAL_STATUSES : Dict[str, Set[OperationalStatus]] = { 'AddDevice': set([OperationalStatus.ENABLED, OperationalStatus.DISABLED]), 'UpdateDevice': set([OperationalStatus.KEEP_STATE, OperationalStatus.ENABLED, OperationalStatus.DISABLED]), } def _check_device_exists(method_name : str, database : Database, device_id : str): if method_name in ['AddDevice']: check_device_not_exists(database, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id) elif method_name in ['UpdateDevice', 'DeleteDevice']: check_device_exists(database, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id) else: # pragma: no cover (test requires malforming the code) msg = 'Unexpected condition: _check_device_exists(method_name={}, device_id={})' msg = msg.format(str(method_name), str(device_id)) raise ServiceException(grpc.StatusCode.UNIMPLEMENTED, msg) def _check_device_endpoint_exists_or_get_pointer( method_name : str, database : Database, parent_name : str, device_id : str, endpoint_id : str) -> Endpoint: if method_name in ['AddDevice']: db_context = database.context(DEFAULT_CONTEXT_ID) db_topology = db_context.topology(DEFAULT_TOPOLOGY_ID) db_device = db_topology.device(device_id) return db_device.endpoint(endpoint_id) elif method_name in ['UpdateDevice', 'DeleteDevice']: return check_device_endpoint_exists( database, parent_name, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id, endpoint_id) else: # pragma: no cover (test requires malforming the code) msg = 'Unexpected condition: _check_device_endpoint_exists_or_get_pointer(method_name={}, ' \ 'parent_name={}, device_id={}, endpoint_id={})' msg = msg.format(str(method_name), str(parent_name), str(device_id), str(endpoint_id)) raise ServiceException(grpc.StatusCode.UNIMPLEMENTED, msg) def check_device_operational_status(method_name : str, value : str) -> OperationalStatus: return check_enum( 'OperationalStatus', method_name, value, to_operationalstatus_enum, ACCEPTED_DEVICE_OPERATIONAL_STATUSES) def check_device_request( method_name : str, request : Device, database : Database, logger : logging.Logger ) -> Tuple[str, str, str, OperationalStatus, List[Tuple[Endpoint, str]]]: # ----- Parse attributes ------------------------------------------------------------------------------------------- try: device_id = chk_string ('device.device_id.device_id.uuid', request.device_id.device_id.uuid, allow_empty=False) device_type = chk_string ('device.device_type', request.device_type, allow_empty=False) device_config = chk_string ('device.device_config.device_config', request.device_config.device_config, allow_empty=True) device_opstat = chk_options('device.devOperationalStatus', request.devOperationalStatus, operationalstatus_enum_values()) except Exception as e: logger.exception('Invalid arguments:') raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e)) device_opstat = check_device_operational_status(method_name, device_opstat) # ----- Check if device exists in database ------------------------------------------------------------------------- _check_device_exists(method_name, database, device_id) # ----- Parse endpoints and check if they exist in the database as device endpoints -------------------------------- add_topology_devices_endpoints : Dict[str, Dict[str, Set[str]]] = {} db_endpoints__port_types : List[Tuple[Endpoint, str]] = [] for endpoint_number,endpoint in enumerate(request.endpointList): parent_name = 'Endpoint(#{}) of Context({})/Topology({})/Device({})' parent_name = parent_name.format(endpoint_number, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id) _, ep_device_id, ep_port_id = check_endpoint_id( logger, endpoint_number, parent_name, endpoint.port_id, add_topology_devices_endpoints, predefined_device_id=device_id, acceptable_device_ids=set([device_id]), prevent_same_device_multiple_times=False) try: ep_port_type = chk_string('endpoint[#{}].port_type'.format(endpoint_number), endpoint.port_type, allow_empty=False) except Exception as e: logger.exception('Invalid arguments:') raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e)) db_endpoint = _check_device_endpoint_exists_or_get_pointer( method_name, database, parent_name, ep_device_id, ep_port_id) db_endpoints__port_types.append((db_endpoint, ep_port_type)) return device_id, device_type, device_config, device_opstat, db_endpoints__port_types def check_device_id_request( method_name : str, request : DeviceId, database : Database, logger : logging.Logger) -> str: # ----- Parse attributes ------------------------------------------------------------------------------------------- try: device_id = chk_string('device_id.device_id.uuid', request.device_id.uuid, allow_empty=False) except Exception as e: logger.exception('Invalid arguments:') raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e)) # ----- Check if device exists in database --------------------------------------------------------------------------- _check_device_exists(method_name, database, device_id) return device_id