EndpointIdCheckers.py

import grpc, logging
from typing import Dict, Set, Tuple, Union
from common.Checkers import chk_string
from common.exceptions.ServiceException import ServiceException
from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
from context.proto.context_pb2 import EndPointId

def check_endpoint_id(
    logger : logging.Logger, endpoint_number : int, parent_name : str, endpoint_id : 'EndPointId',
    add_topology_devices_endpoints : Dict[str, Dict[str, Set[str]]],
    predefined_context_id : str = DEFAULT_CONTEXT_ID, acceptable_context_ids : Set[str] = set([DEFAULT_CONTEXT_ID]),
    predefined_topology_id : str = DEFAULT_TOPOLOGY_ID, acceptable_topology_ids : Set[str] = set([DEFAULT_TOPOLOGY_ID]),
    predefined_device_id : Union[str, None] = None, acceptable_device_ids : Set[str] = set(),
    prevent_same_device_multiple_times : bool = True) -> Tuple[str, str, str]:

    try:
        ep_context_id  = chk_string('endpoint_id[#{}].topoId.contextId.contextUuid.uuid'.format(endpoint_number),
                                    endpoint_id.topoId.contextId.contextUuid.uuid,
                                    allow_empty=True)
        ep_topology_id = chk_string('endpoint_id[#{}].topoId.topoId.uuid'.format(endpoint_number),
                                    endpoint_id.topoId.topoId.uuid,
                                    allow_empty=True)
        ep_device_id   = chk_string('endpoint_id[#{}].dev_id.device_id.uuid'.format(endpoint_number),
                                    endpoint_id.dev_id.device_id.uuid,
                                    allow_empty=(predefined_device_id is not None))
        ep_port_id     = chk_string('endpoint_id[#{}].port_id.uuid'.format(endpoint_number),
                                    endpoint_id.port_id.uuid,
                                    allow_empty=False)
    except Exception as e:
        logger.exception('Invalid arguments:')
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

    if len(ep_context_id) == 0:
        # Assumption: if no context is specified for an endpoint_id, use predefined context
        ep_context_id = predefined_context_id
    elif (len(acceptable_context_ids) > 0) and (ep_context_id not in acceptable_context_ids):
        # Assumption: parent and endpoints should belong to the same context
        msg = ' '.join([
            'Context({}) in {} mismatches acceptable Contexts({}).',
            'Optionally, leave field empty to use predefined Context({}).',
        ])
        msg = msg.format(
            ep_context_id, parent_name, str(acceptable_context_ids), predefined_context_id)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    if len(ep_topology_id) == 0:
        # Assumption: if no topology is specified for an endpoint_id, use predefined topology
        ep_topology_id = predefined_topology_id
    elif (len(acceptable_topology_ids) > 0) and (ep_topology_id not in acceptable_topology_ids):
        msg = ' '.join([
            'Context({})/Topology({}) in {} mismatches acceptable Topologies({}).',
            'Optionally, leave field empty to use predefined Topology({}).',
        ])
        msg = msg.format(
            ep_context_id, ep_topology_id, parent_name, str(acceptable_topology_ids), predefined_topology_id)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    if (predefined_device_id is not None) and (len(ep_device_id) == 0):
        # Assumption: if no device is specified for an endpoint_id, use predefined device, if available
        ep_device_id = predefined_device_id
    elif (len(acceptable_device_ids) > 0) and (ep_device_id not in acceptable_device_ids):
        msg = ' '.join([
            'Context({})/Topology({})/Device({}) in {} mismatches acceptable Devices({}).',
            'Optionally, leave field empty to use predefined Device({}).',
        ])
        msg = msg.format(
            ep_context_id, ep_topology_id, ep_device_id, parent_name, str(acceptable_device_ids), predefined_device_id)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    add_devices = add_topology_devices_endpoints.setdefault(ep_topology_id, dict())
    if prevent_same_device_multiple_times and (ep_device_id in add_devices):
        msg = 'Duplicated Context({})/Topology({})/Device({}) in {}.'
        msg = msg.format(ep_context_id, ep_topology_id, ep_device_id, parent_name)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    add_device_and_endpoints = add_devices.setdefault(ep_device_id, set())

    # Implicit validation: same device cannot appear 2 times in the list of endpoints
    if ep_port_id in add_device_and_endpoints:                                # pragma: no cover
        msg = 'Duplicated Context({})/Topology({})/Device({})/Port({}) in {}.'
        msg = msg.format(ep_context_id, ep_topology_id, ep_device_id, ep_port_id, parent_name)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    add_device_and_endpoints.add(ep_port_id)
    return ep_topology_id, ep_device_id, ep_port_id