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.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
from common.database.api.context.topology.device.OperationalStatus import OperationalStatus
from common.tests.Assertions import validate_device_id, validate_empty
from device.client.DeviceClient import DeviceClient
from device.proto.context_pb2 import Device, DeviceId
from device.service.DeviceService import DeviceService
from device.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD

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}}
DEVICE_ID = {'device_id': {'uuid': 'DEV1'}}
DEVICE = {
    'device_id': copy.deepcopy(DEVICE_ID),
    'device_type': 'ROADM',
    'device_config': {'device_config': '<config/>'},
    'devOperationalStatus': OperationalStatus.ENABLED.value,
    'endpointList' : [
        {
            'port_id': {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': copy.deepcopy(DEVICE_ID), 'port_id': {'uuid' : 'EP2'}},
            'port_type': 'WDM'
        },
        {
            'port_id': {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': copy.deepcopy(DEVICE_ID), 'port_id': {'uuid' : 'EP3'}},
            'port_type': 'WDM'
        },
        {
            'port_id': {'topoId': copy.deepcopy(TOPOLOGY_ID), 'dev_id': copy.deepcopy(DEVICE_ID), 'port_id': {'uuid' : 'EP4'}},
            'port_type': 'WDM'
        },
    ]
}

@pytest.fixture(scope='session')
def device_database():
    _database = get_database(engine=DatabaseEngineEnum.INMEMORY)
    return _database

@pytest.fixture(scope='session')
def device_service(device_database):
    _service = DeviceService(
        device_database, port=port, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
    _service.start()
    yield _service
    _service.stop()

@pytest.fixture(scope='session')
def device_client(device_service):
    _client = DeviceClient(address='127.0.0.1', port=port)
    yield _client
    _client.close()

def test_add_device_wrong_attributes(device_client : DeviceClient):
    # should fail with device uuid is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_id']['device_id']['uuid'] = ''
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'device.device_id.device_id.uuid() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

    # should fail with device type is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_type'] = ''
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'device.device_type() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

    # should fail with wrong device operational status
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['devOperationalStatus'] = OperationalStatus.KEEP_STATE.value
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Method(AddDevice) does not accept OperationalStatus(KEEP_STATE). '\
          'Permitted values for Method(AddDevice) are OperationalStatus([\'DISABLED\', \'ENABLED\']).'
    assert e.value.details() == msg

def test_add_device_wrong_endpoint(device_client : DeviceClient):
    # should fail with unsupported endpoint context
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['endpointList'][0]['port_id']['topoId']['contextId']['contextUuid']['uuid'] = 'wrong-context'
        request = Device(**copy_device)
        device_client.AddDevice(request)
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Context(wrong-context) in Endpoint(#0) of '\
          'Context(admin)/Topology(admin)/Device(DEV1) mismatches acceptable Contexts({\'admin\'}). '\
          'Optionally, leave field empty to use predefined Context(admin).'
    assert e.value.details() == msg

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

    # should fail with wrong endpoint device
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['endpointList'][0]['port_id']['dev_id']['device_id']['uuid'] = 'wrong-device'
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Context(admin)/Topology(admin)/Device(wrong-device) in Endpoint(#0) of '\
          'Context(admin)/Topology(admin)/Device(DEV1) mismatches acceptable Devices({\'DEV1\'}). '\
          'Optionally, leave field empty to use predefined Device(DEV1).'
    assert e.value.details() == msg

    # should fail with endpoint port uuid is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['endpointList'][0]['port_id']['port_id']['uuid'] = ''
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'endpoint_id[#0].port_id.uuid() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

    # should fail with endpoint port type is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['endpointList'][0]['port_type'] = ''
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'endpoint[#0].port_type() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

    # should fail with duplicate port in device
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['endpointList'][1]['port_id']['port_id']['uuid'] = 'EP2'
        device_client.AddDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Duplicated Context(admin)/Topology(admin)/Device(DEV1)/Port(EP2) in Endpoint(#1) of '\
          'Context(admin)/Topology(admin)/Device(DEV1).'
    assert e.value.details() == msg

def test_add_device(device_client : DeviceClient):
    # should work
    validate_device_id(MessageToDict(
            device_client.AddDevice(Device(**DEVICE)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

def test_add_device_duplicate(device_client : DeviceClient):
    # should fail with device already exists
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        device_client.AddDevice(Device(**DEVICE))
    assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
    msg = 'Context(admin)/Topology(admin)/Device(DEV1) already exists in the database.'
    assert e.value.details() == msg

def test_delete_device_empty_uuid(device_client : DeviceClient):
    # should fail with device uuid is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device_id = copy.deepcopy(DEVICE_ID)
        copy_device_id['device_id']['uuid'] = ''
        device_client.DeleteDevice(DeviceId(**copy_device_id))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'device_id.device_id.uuid() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

def test_delete_device_not_found(device_client : DeviceClient):
    # should fail with device not found
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device_id = copy.deepcopy(DEVICE_ID)
        copy_device_id['device_id']['uuid'] = 'wrong-device-id'
        device_client.DeleteDevice(DeviceId(**copy_device_id))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(admin)/Topology(admin)/Device(wrong-device-id) does not exist in the database.'
    assert e.value.details() == msg

def test_delete_device(device_client : DeviceClient):
    # should work
    validate_empty(MessageToDict(
            device_client.DeleteDevice(DeviceId(**DEVICE_ID)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

def test_configure_device_empty_device_uuid(device_client : DeviceClient):
    # should fail with device uuid is empty
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_id']['device_id']['uuid'] = ''
        device_client.ConfigureDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'device.device_id.device_id.uuid() is out of range: '\
          'allow_empty(False) min_length(None) max_length(None) allowed_lengths(None).'
    assert e.value.details() == msg

def test_configure_device_not_found(device_client : DeviceClient):
    # should fail with device not found
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_id']['device_id']['uuid'] = 'wrong-device-id'
        device_client.ConfigureDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
    msg = 'Context(admin)/Topology(admin)/Device(wrong-device-id) does not exist in the database.'
    assert e.value.details() == msg

def test_add_device_default_endpoint_context_topology_device(device_client : DeviceClient):
    # should work
    copy_device = copy.deepcopy(DEVICE)
    copy_device['endpointList'][0]['port_id']['topoId']['contextId']['contextUuid']['uuid'] = ''
    copy_device['endpointList'][0]['port_id']['topoId']['topoId']['uuid'] = ''
    copy_device['endpointList'][0]['port_id']['dev_id']['device_id']['uuid'] = ''
    validate_device_id(MessageToDict(
            device_client.AddDevice(Device(**copy_device)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

def test_configure_device_wrong_attributes(device_client : DeviceClient):
    # should fail with device type is wrong
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_type'] = 'wrong-type'
        device_client.ConfigureDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    msg = 'Device(DEV1) has Type(ROADM) in the database. Cannot be changed to Type(wrong-type).'
    assert e.value.details() == msg

    # should fail with endpoints cannot be modified
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        device_client.ConfigureDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    assert e.value.details() == 'Endpoints belonging to Device(DEV1) cannot be modified.'

    # should fail with any change detected
    with pytest.raises(grpc._channel._InactiveRpcError) as e:
        copy_device = copy.deepcopy(DEVICE)
        copy_device['device_config']['device_config'] = ''
        copy_device['devOperationalStatus'] = OperationalStatus.KEEP_STATE.value
        copy_device['endpointList'].clear()
        device_client.ConfigureDevice(Device(**copy_device))
    assert e.value.code() == grpc.StatusCode.ABORTED
    msg = 'Any change has been requested for Device(DEV1). '\
          'Either specify a new configuration or a new device operational status.'
    assert e.value.details() == msg

def test_configure_device(device_client : DeviceClient):
    # should work
    copy_device = copy.deepcopy(DEVICE)
    copy_device['device_config']['device_config'] = '<new_config/>'
    copy_device['devOperationalStatus'] = OperationalStatus.DISABLED.value
    copy_device['endpointList'].clear()
    validate_device_id(MessageToDict(
            device_client.ConfigureDevice(Device(**copy_device)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))