test_context_device_service.py

import logging, pytest
from google.protobuf.json_format import MessageToDict
from common.Settings import get_setting
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_service_id, \
    validate_service_list_is_not_empty, 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 tester_functional.definitions import DEVICE_DEV1, DEVICE_DEV2, DEVICE_DEV3
from tester_functional.definitions import LINK_DEV1_DEV2, LINK_DEV1_DEV3, LINK_DEV2_DEV1, LINK_DEV2_DEV3, \
    LINK_DEV3_DEV1, LINK_DEV3_DEV2
from tester_functional.definitions import SERVICE_SVC1, SERVICE_SVC2, SERVICE_SVC3
from service.client.ServiceClient import ServiceClient
from service.proto.service_pb2 import Service

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

@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()

@pytest.fixture(scope='session')
def service_client():
    service_host = get_setting('SERVICESERVICE_SERVICE_HOST')
    service_port = get_setting('SERVICESERVICE_SERVICE_PORT_GRPC')
    _client = ServiceClient(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))

def test_add_services(service_client : ServiceClient):
    # should work
    for service in [SERVICE_SVC1, SERVICE_SVC2, SERVICE_SVC3]:
        validate_service_id(MessageToDict(
            service_client.CreateService(Service(**service)),
            including_default_value_fields=True, preserving_proto_field_name=True,
            use_integers_for_enums=False))

    # 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))