Newer
Older
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.tests.script import populate_example
from common.tests.Assertions import validate_empty, validate_link_id, validate_topology
from context.client.ContextClient import ContextClient
from context.proto.context_pb2 import Empty, Link, LinkId
from context.service.ContextService import ContextService
from context.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
port = 10000 + GRPC_SERVICE_PORT # avoid first 1024 privileged ports to avoid evelating permissions for tests
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
LINK_ID = {'link_id': {'uuid': 'DEV1/EP2 ==> DEV2/EP1'}}
'link_id': {'link_id': {'uuid': 'DEV1/EP2 ==> DEV2/EP1'}},
'endpointList' : [
{
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'DEV1'}},
'port_id': {'uuid' : 'EP2'}
},
{
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'DEV2'}},
'port_id': {'uuid' : 'EP1'}
_database = get_database(engine=DatabaseEngineEnum.INMEMORY)
return _database
@pytest.fixture(scope='session')
context_database, port=port, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
_service.start()
yield _service
_service.stop()
@pytest.fixture(scope='session')
_client = ContextClient(address='127.0.0.1', port=port)
yield _client
_client.close()
def test_get_topology_empty(context_client : ContextClient, context_database : Database):
# should work
context_database.clear_all()
context_client.GetTopology(Empty()),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_get_topology_completed(context_client : ContextClient, context_database : Database):
# should work
populate_example(context_database)
validate_topology(MessageToDict(
context_client.GetTopology(Empty()),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_delete_link_empty_uuid(context_client : ContextClient):
# should fail with link not found
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link_id = copy.deepcopy(LINK_ID)
copy_link_id['link_id']['uuid'] = ''
context_client.DeleteLink(LinkId(**copy_link_id))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'link_id.link_id.uuid() string is empty.'
def test_add_link_already_exists(context_client : ContextClient):
# should fail with link already exists
with pytest.raises(grpc._channel._InactiveRpcError) as e:
context_client.AddLink(Link(**LINK))
assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
assert e.value.details() == 'Link(DEV1/EP2 ==> DEV2/EP1) already exists in the database.'
def test_delete_link(context_client : ContextClient):
# should work
validate_empty(MessageToDict(
context_client.DeleteLink(LinkId(**LINK_ID)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_delete_link_not_existing(context_client : ContextClient):
# should fail with link not found
with pytest.raises(grpc._channel._InactiveRpcError) as e:
context_client.DeleteLink(LinkId(**LINK_ID))
assert e.value.code() == grpc.StatusCode.NOT_FOUND
assert e.value.details() == 'Link(DEV1/EP2 ==> DEV2/EP1) does not exist in the database.'
def test_add_link_uuid_empty(context_client : ContextClient):
# should fail with link uuid empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['link_id']['link_id']['uuid'] = ''
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'link.link_id.link_id.uuid() string is empty.'
def test_add_link_endpoint_wrong_context(context_client : ContextClient):
# should fail with unsupported context
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['topoId']['contextId']['contextUuid']['uuid'] = 'wrong-context'
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Unsupported Context(wrong-context) in Endpoint(#0) of Link(DEV1/EP2 ==> DEV2/EP1).',
'Only default Context(admin) is currently supported.',
'Optionally, leave field empty to use default Context.',
])
assert e.value.details() == msg
def test_add_link_endpoint_wrong_topology(context_client : ContextClient):
# should fail with unsupported topology
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['topoId']['topoId']['uuid'] = 'wrong-topo'
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Unsupported Topology(wrong-topo) in Endpoint(#0) of Link(DEV1/EP2 ==> DEV2/EP1).',
'Only default Topology(admin) is currently supported.',
'Optionally, leave field empty to use default Topology.',
])
assert e.value.details() == msg
def test_add_link_empty_device_uuid(context_client : ContextClient):
# should fail with port uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['dev_id']['device_id']['uuid'] = ''
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'endpoint[#0].dev_id.device_id.uuid() string is empty.'
def test_add_link_endpoint_wrong_device(context_client : ContextClient):
# should fail with wrong endpoint device
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['dev_id']['device_id']['uuid'] = 'wrong-device'
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.NOT_FOUND
msg = 'Device(wrong-device) in Endpoint(#0) of Link(DEV1/EP2 ==> DEV2/EP1) does not exist in the database.'
assert e.value.details() == msg
def test_add_link_endpoint_wrong_port(context_client : ContextClient):
# should fail with wrong endpoint port
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['port_id']['uuid'] = 'wrong-port'
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.NOT_FOUND
msg = 'Device(DEV1)/Port(wrong-port) in Endpoint(#0) of Link(DEV1/EP2 ==> DEV2/EP1) does not exist in the database.'
assert e.value.details() == msg
def test_add_link_endpoint_duplicated_device(context_client : ContextClient):
# should fail with duplicated endpoint device
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][1]['dev_id']['device_id']['uuid'] = 'DEV1'
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = 'Duplicated Device(DEV1) in Endpoint(#1) of Link(DEV1/EP2 ==> DEV2/EP1).'
assert e.value.details() == msg
def test_add_link_empty_port_uuid(context_client : ContextClient):
# should fail with port uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['port_id']['uuid'] = ''
context_client.AddLink(Link(**copy_link))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'endpoint[#0].port_id.uuid() string is empty.'
def test_add_link(context_client : ContextClient):
# should work
validate_link_id(MessageToDict(
context_client.AddLink(Link(**LINK)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_delete_link_2(context_client : ContextClient):
# should work
validate_empty(MessageToDict(
context_client.DeleteLink(LinkId(**LINK_ID)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_add_link_default_endpoint_context_topology(context_client : ContextClient):
# should work
copy_link = copy.deepcopy(LINK)
copy_link['endpointList'][0]['topoId']['contextId']['contextUuid']['uuid'] = ''
copy_link['endpointList'][0]['topoId']['topoId']['uuid'] = ''
validate_link_id(MessageToDict(
context_client.AddLink(Link(**copy_link)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_get_topology_completed_2(context_client : ContextClient):
# should work
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))