test_monitoring.py

import logging
import pytest

from src.monitoring.proto import context_pb2
from src.monitoring.proto import monitoring_pb2
from src.monitoring.client.monitoring_client import MonitoringClient
from src.monitoring.service.monitoring_server import start_server, stop_server
# from src.monitoring.tests.test_monitoring import kpi_request

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

SERVER_ADDRESS = '127.0.0.1'
LISTEN_ADDRESS = '[::]'
PORT = 7070

###########################
# Tests Setup
###########################

# This fixture will be requested by test cases and last during testing session
@pytest.fixture(scope='session')
def monitoring_service():
    LOGGER.warning('monitoring_service begin')

    LOGGER.info('Initializing MonitoringService...')
    server = start_server(address=LISTEN_ADDRESS, port=PORT)

    # yield the server, when test finishes, execution will resume to stop it
    LOGGER.warning('monitoring_service yielding')
    yield server

    LOGGER.info('Terminating MonitoringService...')
    stop_server(server)

# This fixture will be requested by test cases and last during testing session.
# The client requires the server, so client fixture has the server as dependency.
@pytest.fixture(scope='session')
def monitoring_client(monitoring_service):
    LOGGER.warning('monitoring_client begin')
    client = MonitoringClient(server=SERVER_ADDRESS, port=PORT)  # instantiate the client
    LOGGER.warning('monitoring_client returning')
    return client

# This fixture will be requested by test cases and last during testing session.
@pytest.fixture(scope='session')
def kpi():
    LOGGER.warning('test_include_kpi begin')
    # form request
    kpi = monitoring_pb2.Kpi()
    kpi.kpi_id.kpi_id.uuid = 'KPIID0000'  # pylint: disable=maybe-no-member
    kpi.kpiDescription = 'KPI Desc'
    return kpi

@pytest.fixture(scope='session')
def kpi_id():
    LOGGER.warning('test_include_kpi begin')

    # form request
    kpi_id = monitoring_pb2.KpiId()
    kpi_id.kpi_id.uuid = 'KPIID0000'

    return kpi_id

@pytest.fixture(scope='session')
def create_kpi_request():
    LOGGER.warning('test_include_kpi begin')

    create_kpi_request = monitoring_pb2.CreateKpiRequest()
    create_kpi_request.device_id.device_id.uuid = 'DEV1'  # pylint: disable=maybe-no-member
    create_kpi_request.kpiDescription = 'KPI Description'
    create_kpi_request.kpi_sample_type = monitoring_pb2.KpiSampleType.PACKETS_TRANSMITTED

    return create_kpi_request

@pytest.fixture(scope='session')
def monitor_kpi_request():
    LOGGER.warning('test_monitor_kpi begin')

    monitor_kpi_request = monitoring_pb2.MonitorKpiRequest()
    monitor_kpi_request.kpi_id.kpi_id.uuid = 'KPIID0000'
    monitor_kpi_request.connexion_time_s = 120
    monitor_kpi_request.sample_rate_ms = 5

    return monitor_kpi_request

@pytest.fixture(scope='session')
def monitor_device_kpi_request():
    LOGGER.warning('test_monitor_kpi begin')

    monitor_device_kpi_request = monitoring_pb2.MonitorDeviceKpiRequest()
    monitor_device_kpi_request.connexion_time_s = 120
    monitor_device_kpi_request.sample_rate_ms = 5

    return monitor_device_kpi_request

###########################
# Tests Implementation
###########################

# Test case that makes use of client fixture to test server's CreateKpi method
def test_create_kpi(monitoring_client,create_kpi_request):
    # make call to server
    LOGGER.warning('test_create_kpi requesting')
    response = monitoring_client.CreateKpi(create_kpi_request)
    LOGGER.debug(str(response))
    assert isinstance(response, monitoring_pb2.KpiId)

# Test case that makes use of client fixture to test server's MonitorKpi method
def test_monitor_kpi(monitoring_client,monitor_kpi_request):
    LOGGER.warning('test_monitor_kpi begin')
    response = monitoring_client.MonitorKpi(monitor_kpi_request)
    LOGGER.debug(str(response))
    assert isinstance(response, context_pb2.Empty)

# Test case that makes use of client fixture to test server's MonitorDeviceKpi method
def test_monitor_device_kpi(monitoring_client,monitor_device_kpi_request):
    LOGGER.warning('test_monitor_device_kpi begin')
    response = monitoring_client.MonitorDeviceKpi(monitor_device_kpi_request)
    LOGGER.debug(str(response))
    assert isinstance(response, context_pb2.Empty)

# Test case that makes use of client fixture to test server's IncludeKpi method
def test_include_kpi(monitoring_client,kpi_request):
    # make call to server
    LOGGER.warning('test_include_kpi requesting')
    response = monitoring_client.IncludeKpi(kpi_request)
    LOGGER.debug(str(response))
    assert isinstance(response, context_pb2.Empty)


# Test case that makes use of client fixture to test server's GetStreamKpi method
def test_getstream_kpi(monitoring_client,kpi):
    LOGGER.warning('test_getstream_kpi begin')
    response = monitoring_client.GetStreamKpi(kpi)
    LOGGER.debug(str(response))
    #assert isinstance(response, monitoring_pb2.Kpi)

# Test case that makes use of client fixture to test server's GetInstantKpi method
def test_getinstant_kpi(monitoring_client,kpi_id):
    LOGGER.warning('test_getinstant_kpi begin')
    response = monitoring_client.GetInstantKpi(kpi_id)
    LOGGER.debug(str(response))
    assert isinstance(response, monitoring_pb2.Kpi)