Initial commit of the branch aimed at adding new tests to monitoring (WIP)

FROM python:3-slim

RUN apt-get update -qqy && \

	apt-get -qqy install wget g++ && \

	rm -rf /var/lib/apt/lists/*

# show python logs as they occur

ENV PYTHONUNBUFFERED=0

# download the grpc health probe

RUN GRPC_HEALTH_PROBE_VERSION=v0.2.0 && \

    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \

    chmod +x /bin/grpc_health_probe

# get packages

WORKDIR /monitoring

COPY monitoring/requirements.in requirements.in

RUN python3 -m pip install pip-tools

RUN pip-compile --output-file=requirements.txt requirements.in

RUN python3 -m pip install -r requirements.txt

# add files into working directory

COPY /monitoring/. .

COPY /common/logger.py .

# set listen port

ENTRYPOINT ["python", "/monitoring/monitoring_client.py"]

import logging, pytest, time

from . import context_pb2

from . import monitoring_pb2

from .monitoring_server import start_server, stop_server

from .monitoring_client import MonitoringClient

LOGGER = logging.getLogger(__name__)

LOGGER.setLevel(logging.DEBUG)

SERVER_ADDRESS = '127.0.0.1'

LISTEN_ADDRESS = '[::]'

PORT = 7070

# 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

# Test case that makes use of client fixture to test server's IncludeKpi method

def test_include_kpi(monitoring_client):

    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'

    # make call to server

    LOGGER.warning('test_include_kpi requesting')

    response = monitoring_client.IncludeKpi(kpi)

    LOGGER.debug(str(response))

    assert isinstance(response, context_pb2.Empty)

# You can add as many tests as you want. Just copy the "def test_include_kpi(monitoring_client):" and implement

# appropriate tests. monitoring_client and monitoring_service fixtures and their connection are reused along tests.