Merge branch 'develop' of https://labs.etsi.org/rep/tfs/controller into feat/context-scalability

# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

coverage==6.3

grpcio==1.47.*

grpcio-health-checking==1.47.*

import grpc, logging

from typing import Dict, List, Set, Tuple

from common.Checkers import chk_string

from common.exceptions.ServiceException import ServiceException

from service.proto.context_pb2 import Constraint

def check_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_constraints(logger : logging.Logger, parent_name : str, constraints):

    add_constraints : Dict[str, str] = {}

    constraint_tuples : List[Tuple[str, str]] = []

    for constraint_number,constraint in enumerate(constraints):

        _parent_name = 'Constraint(#{}) of {}'.format(constraint_number, parent_name)

        constraint_type, constraint_value = check_constraint(

            logger, constraint_number, _parent_name, constraint, add_constraints)

        constraint_tuples.append((constraint_type, constraint_value))

    return constraint_tuples

import grpc

from common.database.api.Database import Database

from common.database.api.context.slice.Slice import Slice

from common.exceptions.ServiceException import ServiceException

def check_slice_exists(database : Database, context_id : str, slice_id : str) -> Slice:

    db_context = database.context(context_id).create()

    if db_context.slices.contains(slice_id): return db_context.slice(slice_id)

    msg = 'Context({})/Slice({}) does not exist in the database.'

    msg = msg.format(context_id, slice_id)

    raise ServiceException(grpc.StatusCode.NOT_FOUND, msg)

def check_slice_not_exists(database : Database, context_id : str, slice_id : str):

    db_context = database.context(context_id).create()

    if not db_context.slices.contains(slice_id): return

    msg = 'Context({})/Slice({}) already exists in the database.'

    msg = msg.format(context_id, slice_id)

    raise ServiceException(grpc.StatusCode.ALREADY_EXISTS, msg)

from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method

from common.type_checkers.Checkers import chk_type

from device.service.driver_api._Driver import _Driver

from .cm.cm_connection import CmConnection

from .cm.cm_connection import CmConnection, ConsistencyMode

from .cm import tf

# Don't complain about non-verified SSL certificate. This driver is demo only

        self.__hub_module_name = settings["hub_module_name"]

        tls_verify = False # Currently using self signed certificates

        username = settings["username"] if "username" in settings else "xr-user-1"

        password = settings["password"] if "password" in settings else "xr-user-1"

        self.__cm_connection = CmConnection(address, int(port), username, password, self.__timeout, tls_verify = tls_verify)

        username = settings.get("username", "xr-user-1")

        password = settings.get("password", "xr-user-1")

        # Options are:

        #    asynchronous --> operation considered complete when IPM responds with suitable status code,

        #                     including "accepted", that only means request is semantically good and queued.

        #    synchronous  --> operation is considered complete once result is also reflected in GETs in REST API.

        #    lifecycle    --> operation is considered successfull once IPM has completed pluggaable configuration

        #                     or failed in it. This is typically unsuitable for production use

        #                     (as some optics may be transiently unreachable), but is convenient for demos and testin.

        consistency_mode = ConsistencyMode.from_str(settings.get("consistency-mode", "asynchronous"))

        self.__cm_connection = CmConnection(address, int(port), username, password, self.__timeout, tls_verify = tls_verify, consistency_mode=consistency_mode)

        self.__constellation = None

        LOGGER.info(f"XrDriver instantiated, cm {address}:{port}, {settings=}")

        LOGGER.info(f"XrDriver instantiated, cm {address}:{port}, consistency mode {str(consistency_mode)}, {settings=}")

    def __str__(self):

        return f"{self.__hub_module_name}@{self.__cm_address}"

import logging

import traceback

from typing import Tuple

from cm.cm_connection import CmConnection

from cm.cm_connection import CmConnection, ConsistencyMode

from cm.tf_service import TFService

from cm.transport_capacity import TransportCapacity

from cm.connection import Connection

parser.add_argument('--list-transport-capacities', action='store_true')

parser.add_argument('--create-transport-capacity', nargs='?', type=str, help="uuid;ifname;ifname;capacity")

parser.add_argument('--emulate-tf-set-config-service', nargs='?', type=str, help="hubmodule;uuid;ifname;ifname;capacity or hubmodule;uuid;ifname;ifname;capacity;FORCE-VTI-ON")

parser.add_argument('--consistency-mode', nargs='?', type=str, help="asynchronous|synchronous|lifecycle;RETRY_INTERVAL_FLOAT_AS_S")

parser.add_argument('--timeout', help='REST call timeout in seconds (per request and total for consistency validation)', type=int, default=60)

args = parser.parse_args()

    print("Invalid object create arguments. Expecting \"href;oid;ifname1;ifname2;bandwidthgbits\" or \"href;oid;ifname1;ifname2\", where ifname is form \"MODULE|PORT\"")

    exit(-1)

cm = CmConnection(args.ip, args.port, args.username, args.password, tls_verify=False)

if args.consistency_mode:

    ca = args.consistency_mode.split(";")

    if 2 != len(ca):

        print("Invalid consistency mode specification. Expecting \"asynchronous|synchronous|lifecycle;RETRY_INTERVAL_FLOAT_AS_S\"")

        exit(-1)

    consistency_mode = ConsistencyMode.from_str(ca[0])

    try:

        retry_interval = float(ca[1])

    except ValueError:

        print("Invalid consistency mode retry interval (non-float)")

        exit(-1)

else:

    consistency_mode = ConsistencyMode.lifecycle

    retry_interval = 0.2

cm = CmConnection(args.ip, args.port, args.username, args.password, timeout=args.timeout, tls_verify=False, consistency_mode=consistency_mode, retry_interval=retry_interval)

if not cm.Connect():

    exit(-1)