Skip to content
Snippets Groups Projects
Commit 3e5ef106 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Intermediate backup

parent 1e34eec7
No related branches found
No related tags found
1 merge request!54Release 2.0.0
......@@ -27,7 +27,7 @@ rm -f $COVERAGEFILE
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
device/tests/test_unitary_driverapi.py \
device/tests/test_unitary_service.py
device/tests/test_unitary.py
#coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
# service/tests/test_unitary.py
......
......@@ -48,29 +48,31 @@ class DeviceServiceServicerImpl(DeviceServiceServicer):
def ConfigureDevice(self, request : Device, context : grpc.ServicerContext) -> DeviceId:
device_id = request.device_id
device_uuid = device_id.device_uuid.uuid
config_name = 'running'
#config_name = 'running'
self.data_cache.sync_device_from_context(device_uuid)
db_device,_ = self.data_cache.set_device(request)
#resources_to_set = List[Tuple[str: resource_key, any: resource_value]]
#resources_to_delete = List[Tuple[str: resource_key]]
#resources_to_subscribe = List[str: resource_key, float: sampling_rate]
#resources_to_unsubscribe = List[Tuple[str: resource_key]]
# Compute "difference" between config field in request and config from Context
# Compute list of changes between device_config in context, and device_config in request
set_changes : List[Tuple[str, Any]] = []
delete_changes : List[Tuple[str, Any]] = []
subscriptions : List[Tuple[str, Any]] = []
unsubscriptions : List[Tuple[str, Any]] = []
driver_filter_fields = self.data_cache.get_device_driver_filter_fields(device_uuid)
driver : _Driver = self.driver_instance_cache.get(device_uuid, **driver_filter_fields)
driver.Connect()
result = driver.SetConfig(set_changes)
# check result
result = driver.DeleteConfig(delete_changes)
# check result
result = driver.SubscribeState(subscriptions)
# check result
result = driver.UnsubscribeState(unsubscriptions)
# check result
#results_setconfig = driver.SetConfig(resources_to_set)
## check result
#results_deleteconfig = driver.DeleteConfig(resources_to_delete)
## check result
#results_subscribestate = driver.SubscribeState(resources_to_subscribe)
## check result
#results_unsubscribestate = driver.UnsubscribeState(resources_to_unsubscribe)
## check result
self.data_cache.sync_device_to_context(device_uuid)
return DeviceId(**db_device.dump_id())
......
......@@ -40,6 +40,8 @@ def main():
# Initialize DataCache
data_cache = DataCache(context_service_host, context_service_port)
# TODO: start monitoring loops to periodically report to Monitoring the collected data
# Initialize Driver framework
driver_factory = DriverFactory(DRIVERS)
driver_instance_cache = DriverInstanceCache(driver_factory)
......
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))
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment