Tools.py

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_TOPOLOGY_ID
from common.database.api.context.topology.device.Endpoint import Endpoint
from common.database.api.context.service.ServiceState import ServiceState, servicestate_enum_values, \
    to_servicestate_enum
from common.database.api.context.service.ServiceType import ServiceType, servicetype_enum_values, to_servicetype_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.ServiceCheckers import check_service_exists, check_service_not_exists
from service.proto.context_pb2 import Constraint
from service.proto.service_pb2 import Service, ServiceId

# For each method name, define acceptable service types. Empty set means accept all.
ACCEPTED_SERVICE_TYPES : Dict[str, Set[ServiceType]] = {
    'CreateService': set([ServiceType.L2NM, ServiceType.L3NM, ServiceType.TAPI_CONNECTIVITY_SERVICE]),
    'UpdateService': set([ServiceType.L2NM, ServiceType.L3NM, ServiceType.TAPI_CONNECTIVITY_SERVICE]),
}

# For each method name, define acceptable service states. Empty set means accept all.
ACCEPTED_SERVICE_STATES : Dict[str, Set[ServiceState]] = {
    'CreateService': set([ServiceState.PLANNED]),
    'UpdateService': set([ServiceState.PLANNED, ServiceState.ACTIVE, ServiceState.PENDING_REMOVAL]),
}

def _check_service_exists(method_name : str, database : Database, context_id : str, service_id : str):
    if method_name in ['CreateService']:
        check_service_not_exists(database, context_id, service_id)
    elif method_name in ['UpdateService', 'DeleteService', 'GetServiceById']:
        check_service_exists(database, context_id, service_id)
    else:                                       # pragma: no cover (test requires malforming the code)
        msg = 'Unexpected condition [_check_service_exists(method_name={}, context_id={}, service_id={})]'
        msg = msg.format(str(method_name), str(context_id), str(service_id))
        raise ServiceException(grpc.StatusCode.UNIMPLEMENTED, msg)

def check_service_type(method_name : str, value : str) -> ServiceType:
    return check_enum('ServiceType', method_name, value, to_servicetype_enum, ACCEPTED_SERVICE_TYPES)

def check_service_state(method_name : str, value : str) -> ServiceState:
    return check_enum('ServiceState', method_name, value, to_servicestate_enum, ACCEPTED_SERVICE_STATES)

def check_service_constraint(
    logger : logging.Logger, constraint_number : int, parent_name : str, constraint : Constraint,
    add_constraints : Dict[str, Dict[str, Set[str]]]) -> Tuple[str, str]:

    try:
        constraint_type  = chk_string('constraint[#{}].constraint_type'.format(constraint_number),
                                      constraint.constraint_type,
                                      allow_empty=False)
        constraint_value = chk_string('constraint[#{}].constraint_value'.format(constraint_number),
                                      constraint.constraint_value,
                                      allow_empty=False)
    except Exception as e:
        logger.exception('Invalid arguments:')
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

    if constraint_type in add_constraints:
        msg = 'Duplicated ConstraintType({}) in {}.'
        msg = msg.format(constraint_type, parent_name)
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    add_constraints[constraint_type] = constraint_value
    return constraint_type, constraint_value

def check_service_request(
    method_name : str, request : Service, database : Database, logger : logging.Logger
    ) -> Tuple[str, str, ServiceType, str, ServiceState, List[Endpoint], List[Tuple[str, str]]]:

    # ----- Parse attributes -------------------------------------------------------------------------------------------
    try:
        context_id     = chk_string ('service.cs_id.contextId.contextUuid.uuid',
                                    request.cs_id.contextId.contextUuid.uuid,
                                    allow_empty=False)
        service_id     = chk_string ('service.cs_id.cs_id.uuid',
                                    request.cs_id.cs_id.uuid,
                                    allow_empty=False)
        service_type   = chk_options('service.serviceType',
                                    request.serviceType,
                                    servicetype_enum_values())
        service_config = chk_string ('service.serviceConfig.serviceConfig',
                                    request.serviceConfig.serviceConfig,
                                    allow_empty=True)
        service_state  = chk_options('service.serviceState.serviceState',
                                    request.serviceState.serviceState,
                                    servicestate_enum_values())
    except Exception as e:
        logger.exception('Invalid arguments:')
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

    service_type = check_service_type(method_name, service_type)
    service_state = check_service_state(method_name, service_state)

    # ----- Check if service exists in database ------------------------------------------------------------------------
    _check_service_exists(method_name, database, context_id, service_id)

    # ----- Parse constraints ------------------------------------------------------------------------------------------
    add_constraints : Dict[str, str] = {}
    constraint_tuples : List[Tuple[str, str]] = []
    for constraint_number,constraint in enumerate(request.constraint):
        parent_name = 'Constraint(#{}) of Context({})/Service({})'.format(constraint_number, context_id, service_id)
        constraint_type, constraint_value = check_service_constraint(
            logger, constraint_number, parent_name, constraint, add_constraints)
        constraint_tuples.append((constraint_type, constraint_value))

    # ----- 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 : List[Endpoint] = []
    for endpoint_number,endpoint_id in enumerate(request.endpointList):
        parent_name = 'Endpoint(#{}) of Context({})/Service({})'.format(endpoint_number, context_id, service_id)

        ep_topology_id, ep_device_id, ep_port_id = check_endpoint_id(
            logger, endpoint_number, parent_name, endpoint_id, add_topology_devices_endpoints,
            predefined_context_id=context_id, acceptable_context_ids=set([context_id]))

        db_endpoint = check_device_endpoint_exists(
            database, parent_name, context_id, ep_topology_id, ep_device_id, ep_port_id)
        db_endpoints.append(db_endpoint)

    return context_id, service_id, service_type, service_config, service_state, db_endpoints, constraint_tuples

def check_service_id_request(
    method_name : str, request : ServiceId, database : Database, logger : logging.Logger) -> Tuple[str, str]:

    # ----- Parse attributes -------------------------------------------------------------------------------------------
    try:
        context_id     = chk_string ('service_id.contextId.contextUuid.uuid',
                                    request.contextId.contextUuid.uuid,
                                    allow_empty=False)
        service_id     = chk_string ('service_id.cs_id.uuid',
                                    request.cs_id.uuid,
                                    allow_empty=False)
    except Exception as e:
        logger.exception('Invalid arguments:')
        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

    # ----- Check if service exists in database ------------------------------------------------------------------------
    _check_service_exists(method_name, database, context_id, service_id)

    return context_id, service_id