test_unitary.py

import copy, grpc, logging, pytest
from google.protobuf.json_format import MessageToDict
from common.database.Factory import get_database, DatabaseEngineEnum
from common.database.api.Database import Database
from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
from common.database.tests.script import populate_example
from common.tests.Assertions import validate_empty, validate_service, validate_service_id, \
    validate_service_list_is_empty, validate_service_list_is_not_empty
from service.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
from service.client.ServiceClient import ServiceClient
from service.proto.context_pb2 import Empty
from service.proto.service_pb2 import Service, ServiceId, ServiceStateEnum, ServiceType
from service.service.ServiceService import ServiceService

port = 10000 + GRPC_SERVICE_PORT # avoid privileged ports

LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)

# use "copy.deepcopy" to prevent propagating forced changes during tests
CONTEXT_ID = {'contextUuid': {'uuid': DEFAULT_CONTEXT_ID}}
TOPOLOGY_ID = {'contextId': copy.deepcopy(CONTEXT_ID), 'topoId': {'uuid': DEFAULT_TOPOLOGY_ID}}
SERVICE_ID = {'contextId': copy.deepcopy(CONTEXT_ID), 'cs_id': {'uuid': 'DEV1'}}
SERVICE = {
    'cs_id': copy.deepcopy(SERVICE_ID),
    'serviceType': ServiceType.L3NM,
    'serviceConfig': {'serviceConfig': '<config/>'},
    'serviceState': {'serviceState': ServiceStateEnum.PLANNED},
    'constraint': [
        {'constraint_type': 'latency_ms', 'constraint_value': '100'},
        {'constraint_type': 'hops', 'constraint_value': '5'},
    ],
    'endpointList' : [
        {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': {'device_id': {'uuid': 'DEV1'}}, 'port_id': {'uuid' : 'EP5'}},
        {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': {'device_id': {'uuid': 'DEV2'}}, 'port_id': {'uuid' : 'EP5'}},
        {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': {'device_id': {'uuid': 'DEV3'}}, 'port_id': {'uuid' : 'EP5'}},
    ]
}

@pytest.fixture(scope='session')
def database():
    _database = get_database(engine=DatabaseEngineEnum.INMEMORY)
    populate_example(_database, add_services=False)
    return _database

@pytest.fixture(scope='session')
def service_service(database):
    _service = ServiceService(
        database, port=port, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
    _service.start()
    yield _service
    _service.stop()

@pytest.fixture(scope='session')
def service_client(service_service):
    _client = ServiceClient(address='127.0.0.1', port=port)
    yield _client
    _client.close()

def test_get_services_empty(service_client : ServiceClient):
    # should work
    validate_service_list_is_empty(MessageToDict(
        service_client.GetServiceList(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_create_service_wrong_service_attributes(service_client : ServiceClient):
    # should fail with wrong service context
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['cs_id']['contextId']['contextUuid']['uuid'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'service.cs_id.contextId.contextUuid.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with service context does not exist
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['cs_id']['contextId']['contextUuid']['uuid'] = 'wrong-context'
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(wrong-context) does not exist in the database.'
    assert e.value.details() == msg

    # should fail with wrong service id
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['cs_id']['cs_id']['uuid'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'service.cs_id.cs_id.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with wrong service type
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['serviceType'] = ServiceType.UNKNOWN
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = ' '.join([
        'Method(CreateService) does not accept ServiceType(UNKNOWN).',
        'Permitted values for Method(CreateService) are',
        'ServiceType([\'L2NM\', \'L3NM\', \'TAPI_CONNECTIVITY_SERVICE\']).',
    ])
    assert e.value.details() == msg

    # should fail with wrong service state
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['serviceState']['serviceState'] = ServiceStateEnum.PENDING_REMOVAL
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = ' '.join([
        'Method(CreateService) does not accept ServiceState(PENDING_REMOVAL).',
        'Permitted values for Method(CreateService) are',
        'ServiceState([\'PLANNED\']).',
    ])
    assert e.value.details() == msg

def test_create_service_wrong_constraint(service_client : ServiceClient):
    # should fail with wrong constraint type
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['constraint'][0]['constraint_type'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'constraint[#0].constraint_type() string is empty.'
    assert e.value.details() == msg

    # should fail with wrong constraint value
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['constraint'][0]['constraint_value'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'constraint[#0].constraint_value() string is empty.'
    assert e.value.details() == msg

    # should fail with dupplicated constraint type
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['constraint'][1] = copy_service['constraint'][0]
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Duplicated ConstraintType(latency_ms) in Constraint(#1) of Context(admin)/Service(DEV1).'
    assert e.value.details() == msg

def test_create_service_wrong_endpoint(service_client : ServiceClient, database : Database):
    # should fail with wrong endpoint context
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['topoId']['contextId']['contextUuid']['uuid'] = 'wrong-context'
        print(copy_service)
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = ' '.join([
        'Context(wrong-context) in Endpoint(#0) of Context(admin)/Service(DEV1)',
        'mismatches acceptable Contexts({\'admin\'}).',
        'Optionally, leave field empty to use predefined Context(admin).'
    ])
    assert e.value.details() == msg

    # should fail with wrong endpoint topology
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['topoId']['topoId']['uuid'] = 'wrong-topo'
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = ' '.join([
        'Context(admin)/Topology(wrong-topo) in Endpoint(#0) of Context(admin)/Service(DEV1)',
        'mismatches acceptable Topologies({\'admin\'}).',
        'Optionally, leave field empty to use predefined Topology(admin).',
    ])
    assert e.value.details() == msg

    # should fail with endpoint device is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['dev_id']['device_id']['uuid'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'endpoint_id[#0].dev_id.device_id.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with endpoint device not found
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['dev_id']['device_id']['uuid'] = 'wrong-device'
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = ' '.join([
        'Context(admin)/Topology(admin)/Device(wrong-device) in Endpoint(#0)',
        'of Context(admin)/Service(DEV1) does not exist in the database.',
    ])
    assert e.value.details() == msg

    # should fail with endpoint device duplicated
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][1] = copy_service['endpointList'][0]
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Duplicated Context(admin)/Topology(admin)/Device(DEV1) in Endpoint(#1) of Context(admin)/Service(DEV1).'
    assert e.value.details() == msg

    # should fail with endpoint port is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['port_id']['uuid'] = ''
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'endpoint_id[#0].port_id.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with endpoint port not found
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service = copy.deepcopy(SERVICE)
        copy_service['endpointList'][0]['port_id']['uuid'] = 'wrong-port'
        service_client.CreateService(Service(**copy_service))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = ' '.join([
        'Context(admin)/Topology(admin)/Device(DEV1)/Port(wrong-port) in Endpoint(#0)',
        'of Context(admin)/Service(DEV1) does not exist in the database.',
    ])
    assert e.value.details() == msg

def test_get_service_does_not_exist(service_client : ServiceClient):
    # should fail with service context does not exist
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service_id = copy.deepcopy(SERVICE_ID)
        copy_service_id['contextId']['contextUuid']['uuid'] = 'wrong-context'
        service_client.GetServiceById(ServiceId(**copy_service_id))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(wrong-context) does not exist in the database.'
    assert e.value.details() == msg

    # should fail with service does not exist
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        service_client.GetServiceById(ServiceId(**SERVICE_ID))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(admin)/Service(DEV1) does not exist in the database.'
    assert e.value.details() == msg

def test_update_service_does_not_exist(service_client : ServiceClient):
    # should fail with service does not exist
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        service_client.UpdateService(Service(**SERVICE))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(admin)/Service(DEV1) does not exist in the database.'
    assert e.value.details() == msg

def test_create_service(service_client : ServiceClient):
    # should work
    validate_service_id(MessageToDict(
        service_client.CreateService(Service(**SERVICE)),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_create_service_already_exists(service_client : ServiceClient):
    # should fail with service already exists
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        service_client.CreateService(Service(**SERVICE))
    assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
    msg = 'Context(admin)/Service(DEV1) already exists in the database.'
    assert e.value.details() == msg

def test_get_service(service_client : ServiceClient):
    # should work
    validate_service(MessageToDict(
        service_client.GetServiceById(ServiceId(**SERVICE_ID)),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_update_service(service_client : ServiceClient):
    # should work
    copy_service = copy.deepcopy(SERVICE)
    copy_service['serviceConfig']['serviceConfig'] = '<newconfig/>'
    copy_service['serviceState']['serviceState'] = ServiceStateEnum.ACTIVE
    copy_service['constraint'] = [
        {'constraint_type': 'latency_ms', 'constraint_value': '200'},
        {'constraint_type': 'bandwidth_gbps', 'constraint_value': '100'},
    ]
    copy_service['endpointList'] = [
        {
            'topoId': {'contextId': {'contextUuid': {'uuid': 'admin'}}, 'topoId': {'uuid': 'admin'}},
            'dev_id': {'device_id': {'uuid': 'DEV1'}},
            'port_id': {'uuid' : 'EP5'}
        },
        {
            'topoId': {'contextId': {'contextUuid': {'uuid': 'admin'}}, 'topoId': {'uuid': 'admin'}},
            'dev_id': {'device_id': {'uuid': 'DEV2'}},
            'port_id': {'uuid' : 'EP6'}
        },
    ]
    validate_service_id(MessageToDict(
        service_client.UpdateService(Service(**copy_service)),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_delete_service_wrong_service_id(service_client : ServiceClient):
    # should fail with service context is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service_id = copy.deepcopy(SERVICE_ID)
        copy_service_id['contextId']['contextUuid']['uuid'] = ''
        service_client.DeleteService(ServiceId(**copy_service_id))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'service_id.contextId.contextUuid.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with service context does not exist
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service_id = copy.deepcopy(SERVICE_ID)
        copy_service_id['contextId']['contextUuid']['uuid'] = 'wrong-context'
        service_client.DeleteService(ServiceId(**copy_service_id))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(wrong-context) does not exist in the database.'
    assert e.value.details() == msg

    # should fail with service id is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service_id = copy.deepcopy(SERVICE_ID)
        copy_service_id['cs_id']['uuid'] = ''
        service_client.DeleteService(ServiceId(**copy_service_id))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'service_id.cs_id.uuid() string is empty.'
    assert e.value.details() == msg

    # should fail with service id is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_service_id = copy.deepcopy(SERVICE_ID)
        copy_service_id['cs_id']['uuid'] = 'wrong-service'
        service_client.DeleteService(ServiceId(**copy_service_id))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(admin)/Service(wrong-service) does not exist in the database.'
    assert e.value.details() == msg

def test_delete_service(service_client : ServiceClient):
    # should work
    validate_empty(MessageToDict(
        service_client.DeleteService(ServiceId(**SERVICE_ID)),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_get_services_empty_2(service_client : ServiceClient):
    # should work
    validate_service_list_is_empty(MessageToDict(
        service_client.GetServiceList(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_create_service_empty_endpoints(service_client : ServiceClient):
    # should work
    copy_service = copy.deepcopy(SERVICE)
    copy_service['endpointList'][0]['topoId']['contextId']['contextUuid']['uuid'] = ''
    copy_service['endpointList'][0]['topoId']['topoId']['uuid'] = ''
    validate_service_id(MessageToDict(
        service_client.CreateService(Service(**copy_service)),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_get_services_full(service_client : ServiceClient):
    # should work
    validate_service_list_is_not_empty(MessageToDict(
        service_client.GetServiceList(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))