cm-cli.py

#!/usr/bin/env python3
#pylint: disable=invalid-name, missing-function-docstring, line-too-long, logging-fstring-interpolation, missing-class-docstring, missing-module-docstring
# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# 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.

# Test program for CmConnection
import argparse
import logging
import traceback
from typing import Tuple
from cm.cm_connection import CmConnection, ConsistencyMode
from cm.tf_service import TFService
from cm.transport_capacity import TransportCapacity
from cm.connection import Connection
import cm.tf as tf

logging.basicConfig(level=logging.INFO)

parser = argparse.ArgumentParser(description='CM Connectin Test Utility')
parser.add_argument('ip', help='CM IP address or domain name')
parser.add_argument('port', help='CM port', type=int)
parser.add_argument('username', help='Username')
parser.add_argument('password', help='Password')

parser.add_argument('--list-constellations', action='store_true')
parser.add_argument('--show-constellation-by-hub-name', nargs='?', type=str)
parser.add_argument('--create-connection', nargs='?', type=str, help="uuid;ifname;ifname;capacity")
parser.add_argument('--modify-connection', nargs='?', type=str, help="href;uuid;ifname;ifname;capacity")
parser.add_argument('--show-connection-by-name', nargs='?', type=str)
parser.add_argument('--list-connections', action='store_true')
parser.add_argument('--delete-connection', nargs='?', type=str, help="connection id, e.g. \"/network-connections/4505d5d3-b2f3-40b8-8ec2-4a5b28523c03\"")
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()

def cli_create_string_to_tf_service(cli_create_str: str) -> TFService:
    sargs = cli_create_str.split(";")
    if len(sargs) == 3:
        return TFService(*sargs, 0)
    if len(sargs) == 4:
        sargs[-1] = int(sargs[-1])
        return TFService(*sargs)
    print("Invalid object create arguments. Expecting \"oid;ifname1;ifname2;bandwidthgbits\" or \"oid;ifname1;ifname2\", where ifname is form \"MODULE|PORT\"")
    exit(-1)

def cli_modify_string_to_tf_service(cli_create_str: str) -> Tuple[str, TFService]:
    sargs = cli_create_str.split(";")
    if len(sargs) == 4:
        return (sargs[0], TFService(*sargs[1:], 0))
    if len(sargs) == 5:
        sargs[-1] = int(sargs[-1])
        return (sargs[0], TFService(*sargs[1:]))
    print("Invalid object create arguments. Expecting \"href;oid;ifname1;ifname2;bandwidthgbits\" or \"href;oid;ifname1;ifname2\", where ifname is form \"MODULE|PORT\"")
    exit(-1)

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)

if args.list_constellations:
    constellations = cm.list_constellations()
    for constellation in constellations:
        print("Constellation:", constellation.constellation_id)
        for if_name in constellation.ifnames():
            print(f"    {if_name}")

if args.show_constellation_by_hub_name:
    constellation = cm.get_constellation_by_hub_name(args.show_constellation_by_hub_name)
    if constellation:
        print(f"Constellation: {constellation.constellation_id},  traffic-mode: {constellation.traffic_mode}")
        for if_name in constellation.ifnames():
            print(f"    {if_name}")

if args.create_connection:
    tf_service = cli_create_string_to_tf_service(args.create_connection)
    connection = Connection(from_tf_service=tf_service)
    created_service = cm.create_connection(connection)
    if created_service:
        print(f"Created {created_service} for {connection}")
    else:
        print(f"Failed to create {connection}")

if args.modify_connection:
    href, tf_service = cli_modify_string_to_tf_service(args.modify_connection)
    mc_args = args.modify_connection.split(";")
    connection = Connection(from_tf_service=tf_service)
    result = cm.update_connection(href, connection)
    if result:
        print(f"Updated {href} for {connection}")
    else:
        print(f"Failed to update {href} for {connection}")

if args.show_connection_by_name:
    connection = cm.get_connection_by_name(args.show_connection_by_name)
    if connection:
        print(str(connection))

if args.list_connections:
    connections = cm.get_connections()
    for c in connections:
        print(str(c))

if args.delete_connection:
    was_deleted = cm.delete_connection(args.delete_connection)
    if was_deleted:
        print(f"Successfully deleted {args.delete_connection}")
    else:
        print(f"Failed to delete {args.delete_connection}")

if args.list_transport_capacities:
    tcs = cm.get_transport_capacities()
    for tc in tcs:
        print(str(tc))

if args.create_transport_capacity:
    tf_service = cli_create_string_to_tf_service(args.create_transport_capacity)
    tc = TransportCapacity(from_tf_service=tf_service)
    created_service = cm.create_transport_capacity(tc)
    if created_service:
        print(f"Created {created_service} for {tc}")
    else:
        print(f"Failed to create {tc}")

if args.emulate_tf_set_config_service:
    eargs = args.emulate_tf_set_config_service.split(";")
    if len(eargs) < 5:
        print("Mandatory tokens missing for --emulate-tf-set-config-service")
        exit(-1)

    hub_module_name, uuid, input_sip, output_sip, capacity_value  = eargs[0:5]
    capacity_value = int(capacity_value)
    config = {
        "input_sip": input_sip,
        "output_sip": output_sip,
        "capacity_value": capacity_value,
        "capacity_unit": "gigabit"
    }

    constellation = cm.get_constellation_by_hub_name(hub_module_name)

    # Allow testing some of the VTI code before we have CM that has VTI
    if len(eargs) > 5 and eargs[5] == "FORCE-VTI-ON":
        constellation.traffic_mode = "VTIMode"

    if constellation is None:
        print(f"Unable to find constellation for hub-module {hub_module_name}")
        exit(-1)
    result = tf.set_config_for_service(cm, constellation, uuid, config)
    print(f"Emulated SetConfig() for service result: {result}")
    if isinstance(result, Exception):
        traceback.print_exception(result)