test_context_device.py

import logging, os, 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.tests.Assertions import validate_device_id, validate_link_id, validate_topology_has_devices, \
    validate_topology_has_links, validate_topology_is_empty
from context.client.ContextClient import ContextClient
from context.proto.context_pb2 import Device, Empty, Link
from device.client.DeviceClient import DeviceClient
from .definitions import DEVICE_DEV1, DEVICE_DEV2, DEVICE_DEV3, LINK_DEV1_DEV2, LINK_DEV1_DEV3, LINK_DEV2_DEV1, \
    LINK_DEV2_DEV3, LINK_DEV3_DEV1, LINK_DEV3_DEV2

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

def get_setting(name):
    value = os.environ.get(name)
    if value is None: raise Exception('Unable to find variable({})'.format(name))
    return value

@pytest.fixture(scope='session')
def redis_database():
    _database = get_database(engine=DatabaseEngineEnum.REDIS, REDIS_DATABASE_ID=0)
    return _database

@pytest.fixture(scope='session')
def context_client():
    service_host = get_setting('CONTEXTSERVICE_SERVICE_HOST')
    service_port = get_setting('CONTEXTSERVICE_SERVICE_PORT_GRPC')
    _client = ContextClient(address=service_host, port=service_port)
    yield _client
    _client.close()

@pytest.fixture(scope='session')
def device_client():
    service_host = get_setting('DEVICESERVICE_SERVICE_HOST')
    service_port = get_setting('DEVICESERVICE_SERVICE_PORT_GRPC')
    _client = DeviceClient(address=service_host, port=service_port)
    yield _client
    _client.close()

def test_clean_database(redis_database : Database):
    # should work
    redis_database.clear_all()

def test_get_topology_empty(context_client : ContextClient):
    # should work
    validate_topology_is_empty(MessageToDict(
        context_client.GetTopology(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_add_devices(context_client : ContextClient, device_client : DeviceClient):
    # should work
    for device in [DEVICE_DEV1, DEVICE_DEV2, DEVICE_DEV3]:
        validate_device_id(MessageToDict(
            device_client.AddDevice(Device(**device)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

    # should work
    validate_topology_has_devices(MessageToDict(
        context_client.GetTopology(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))

def test_add_links(context_client : ContextClient):
    # should work
    for link in [LINK_DEV1_DEV2, LINK_DEV1_DEV3, LINK_DEV2_DEV1, LINK_DEV2_DEV3, LINK_DEV3_DEV1, LINK_DEV3_DEV2]:
        validate_link_id(MessageToDict(
            context_client.AddLink(Link(**link)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

    # should work
    validate_topology_has_links(MessageToDict(
        context_client.GetTopology(Empty()),
        including_default_value_fields=True, preserving_proto_field_name=True,
        use_integers_for_enums=False))