Preliminary unfinished version of device service and Context API based on...

Preliminary unfinished version of device service and Context API based on Redis. To be finished, it is just a backup.

src/device/Dockerfile

+FROM python:3-slim
+
+# Install dependencies
+RUN apt-get --yes --quiet --quiet update && \
+    apt-get --yes --quiet --quiet install wget g++ && \
+    rm -rf /var/lib/apt/lists/*
+
+# Set Python to show 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 generic Python packages
+RUN python3 -m pip install --upgrade pip setuptools wheel pip-tools
+
+# Set working directory
+WORKDIR /var/teraflow
+
+# Create module sub-folders
+RUN mkdir -p /var/teraflow/device
+
+# Get Python packages per module
+COPY device/requirements.in device/requirements.in
+RUN pip-compile --output-file=device/requirements.txt device/requirements.in
+RUN python3 -m pip install -r device/requirements.in
+
+# Add files into working directory
+COPY common/. common
+COPY device/. device
+
+# Start device service
+ENTRYPOINT ["python", "-m", "device.service"]

src/device/client/DeviceClient.py

+import grpc, logging
+from google.protobuf.json_format import MessageToDict
+from common.tools.RetryDecorator import retry, delay_exponential
+from device.proto.device_pb2_grpc import DeviceServiceStub
+
+LOGGER = logging.getLogger(__name__)
+MAX_RETRIES = 15
+DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
+
+class DeviceClient:
+    def __init__(self, address, port):
+        self.endpoint = '{}:{}'.format(address, port)
+        LOGGER.debug('Creating channel to {}...'.format(self.endpoint))
+        self.channel = None
+        self.stub = None
+        self.connect()
+        LOGGER.debug('Channel created')
+
+    def connect(self):
+        self.channel = grpc.insecure_channel(self.endpoint)
+        self.stub = DeviceServiceStub(self.channel)
+
+    def close(self):
+        if(self.channel is not None): self.channel.close()
+        self.channel = None
+        self.stub = None
+
+    @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
+    def AddDevice(self, request):
+        LOGGER.debug('AddDevice request: {}'.format(request))
+        response = self.stub.AddDevice(request)
+        LOGGER.debug('AddDevice result: {}'.format(response))
+        return MessageToDict(
+            response, including_default_value_fields=True, preserving_proto_field_name=True,
+            use_integers_for_enums=False)
+
+    @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
+    def ConfigureDevice(self, request):
+        LOGGER.debug('ConfigureDevice request: {}'.format(request))
+        response = self.stub.ConfigureDevice(request)
+        LOGGER.debug('ConfigureDevice result: {}'.format(response))
+        return MessageToDict(
+            response, including_default_value_fields=True, preserving_proto_field_name=True,
+            use_integers_for_enums=False)
+
+    @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
+    def DeleteDevice(self, request):
+        LOGGER.debug('DeleteDevice request: {}'.format(request))
+        response = self.stub.DeleteDevice(request)
+        LOGGER.debug('DeleteDevice result: {}'.format(response))
+        return MessageToDict(
+            response, including_default_value_fields=True, preserving_proto_field_name=True,
+            use_integers_for_enums=False)

src/device/genproto.sh

+#!/bin/bash -eu
+#
+# Copyright 2018 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#!/bin/bash -e
+
+# Make folder containing the script the root folder for its execution
+cd $(dirname $0)
+
+rm -rf proto/*.py
+touch proto/__init__.py
+
+python -m grpc_tools.protoc -I../../proto --python_out=proto --grpc_python_out=proto context.proto
+python -m grpc_tools.protoc -I../../proto --python_out=proto --grpc_python_out=proto device.proto
+
+sed -i -E 's/(import\ .*)_pb2/from device.proto \1_pb2/g' proto/context_pb2.py
+sed -i -E 's/(import\ .*)_pb2/from device.proto \1_pb2/g' proto/context_pb2_grpc.py
+sed -i -E 's/(import\ .*)_pb2/from device.proto \1_pb2/g' proto/device_pb2.py
+sed -i -E 's/(import\ .*)_pb2/from device.proto \1_pb2/g' proto/device_pb2_grpc.py

src/device/proto/context_pb2_grpc.py

+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+
+from device.proto import context_pb2 as context__pb2
+
+
+class ContextServiceStub(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.GetTopology = channel.unary_unary(
+                '/context.ContextService/GetTopology',
+                request_serializer=context__pb2.Empty.SerializeToString,
+                response_deserializer=context__pb2.Topology.FromString,
+                )
+
+
+class ContextServiceServicer(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def GetTopology(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_ContextServiceServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'GetTopology': grpc.unary_unary_rpc_method_handler(
+                    servicer.GetTopology,
+                    request_deserializer=context__pb2.Empty.FromString,
+                    response_serializer=context__pb2.Topology.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'context.ContextService', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+
+
+ # This class is part of an EXPERIMENTAL API.
+class ContextService(object):
+    """Missing associated documentation comment in .proto file."""
+
+    @staticmethod
+    def GetTopology(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/context.ContextService/GetTopology',
+            context__pb2.Empty.SerializeToString,
+            context__pb2.Topology.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

src/device/proto/device_pb2.py

+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# source: device.proto
+"""Generated protocol buffer code."""
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
+from google.protobuf import symbol_database as _symbol_database
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+from device.proto import context_pb2 as context__pb2
+
+
+DESCRIPTOR = _descriptor.FileDescriptor(
+  name='device.proto',
+  package='device',
+  syntax='proto3',
+  serialized_options=None,
+  create_key=_descriptor._internal_create_key,
+  serialized_pb=b'\n\x0c\x64\x65vice.proto\x12\x06\x64\x65vice\x1a\rcontext.proto2\xb6\x01\n\rDeviceService\x12\x31\n\tAddDevice\x12\x0f.context.Device\x1a\x11.context.DeviceId\"\x00\x12=\n\x0f\x43onfigureDevice\x12\x15.context.DeviceConfig\x1a\x11.context.DeviceId\"\x00\x12\x33\n\x0c\x44\x65leteDevice\x12\x11.context.DeviceId\x1a\x0e.context.Empty\"\x00\x62\x06proto3'
+  ,
+  dependencies=[context__pb2.DESCRIPTOR,])
+
+
+
+_sym_db.RegisterFileDescriptor(DESCRIPTOR)
+
+
+
+_DEVICESERVICE = _descriptor.ServiceDescriptor(
+  name='DeviceService',
+  full_name='device.DeviceService',
+  file=DESCRIPTOR,
+  index=0,
+  serialized_options=None,
+  create_key=_descriptor._internal_create_key,
+  serialized_start=40,
+  serialized_end=222,
+  methods=[
+  _descriptor.MethodDescriptor(
+    name='AddDevice',
+    full_name='device.DeviceService.AddDevice',
+    index=0,
+    containing_service=None,
+    input_type=context__pb2._DEVICE,
+    output_type=context__pb2._DEVICEID,
+    serialized_options=None,
+    create_key=_descriptor._internal_create_key,
+  ),
+  _descriptor.MethodDescriptor(
+    name='ConfigureDevice',
+    full_name='device.DeviceService.ConfigureDevice',
+    index=1,
+    containing_service=None,
+    input_type=context__pb2._DEVICECONFIG,
+    output_type=context__pb2._DEVICEID,
+    serialized_options=None,
+    create_key=_descriptor._internal_create_key,
+  ),
+  _descriptor.MethodDescriptor(
+    name='DeleteDevice',
+    full_name='device.DeviceService.DeleteDevice',
+    index=2,
+    containing_service=None,
+    input_type=context__pb2._DEVICEID,
+    output_type=context__pb2._EMPTY,
+    serialized_options=None,
+    create_key=_descriptor._internal_create_key,
+  ),
+])
+_sym_db.RegisterServiceDescriptor(_DEVICESERVICE)
+
+DESCRIPTOR.services_by_name['DeviceService'] = _DEVICESERVICE
+
+# @@protoc_insertion_point(module_scope)

src/device/proto/device_pb2_grpc.py

+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+
+from device.proto import context_pb2 as context__pb2
+
+
+class DeviceServiceStub(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.AddDevice = channel.unary_unary(
+                '/device.DeviceService/AddDevice',
+                request_serializer=context__pb2.Device.SerializeToString,
+                response_deserializer=context__pb2.DeviceId.FromString,
+                )
+        self.ConfigureDevice = channel.unary_unary(
+                '/device.DeviceService/ConfigureDevice',
+                request_serializer=context__pb2.DeviceConfig.SerializeToString,
+                response_deserializer=context__pb2.DeviceId.FromString,
+                )
+        self.DeleteDevice = channel.unary_unary(
+                '/device.DeviceService/DeleteDevice',
+                request_serializer=context__pb2.DeviceId.SerializeToString,
+                response_deserializer=context__pb2.Empty.FromString,
+                )
+
+
+class DeviceServiceServicer(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def AddDevice(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def ConfigureDevice(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def DeleteDevice(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_DeviceServiceServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'AddDevice': grpc.unary_unary_rpc_method_handler(
+                    servicer.AddDevice,
+                    request_deserializer=context__pb2.Device.FromString,
+                    response_serializer=context__pb2.DeviceId.SerializeToString,
+            ),
+            'ConfigureDevice': grpc.unary_unary_rpc_method_handler(
+                    servicer.ConfigureDevice,
+                    request_deserializer=context__pb2.DeviceConfig.FromString,
+                    response_serializer=context__pb2.DeviceId.SerializeToString,
+            ),
+            'DeleteDevice': grpc.unary_unary_rpc_method_handler(
+                    servicer.DeleteDevice,
+                    request_deserializer=context__pb2.DeviceId.FromString,
+                    response_serializer=context__pb2.Empty.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'device.DeviceService', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+
+
+ # This class is part of an EXPERIMENTAL API.
+class DeviceService(object):
+    """Missing associated documentation comment in .proto file."""
+
+    @staticmethod
+    def AddDevice(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/device.DeviceService/AddDevice',
+            context__pb2.Device.SerializeToString,
+            context__pb2.DeviceId.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def ConfigureDevice(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/device.DeviceService/ConfigureDevice',
+            context__pb2.DeviceConfig.SerializeToString,
+            context__pb2.DeviceId.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def DeleteDevice(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/device.DeviceService/DeleteDevice',
+            context__pb2.DeviceId.SerializeToString,
+            context__pb2.Empty.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

src/device/run_integration_tests.sh

+#!/bin/bash
+
+# Make folder containing the script the root folder for its execution
+cd $(dirname $0)
+
+ENDPOINT=($(kubectl --namespace teraflow-development get service deviceservice -o 'jsonpath={.spec.clusterIP} {.spec.ports[?(@.name=="grpc")].port}'))
+docker run -it --env TEST_TARGET_ADDRESS=${ENDPOINT[0]} --env TEST_TARGET_PORT=${ENDPOINT[1]} device_service:test

src/device/run_unitary_tests.sh

+#!/bin/bash
+
+# Make folder containing the script the root folder for its execution
+cd $(dirname $0)
+
+mkdir -p data
+pytest -v --log-level=DEBUG tests/test_unitary.py

src/device/service/DeviceService.py

+import grpc
+import logging
+from concurrent import futures
+from grpc_health.v1.health import HealthServicer, OVERALL_HEALTH
+from grpc_health.v1.health_pb2 import HealthCheckResponse
+from grpc_health.v1.health_pb2_grpc import add_HealthServicer_to_server
+from device.proto.device_pb2_grpc import add_DeviceServiceServicer_to_server
+from device.service.DeviceServiceServicerImpl import DeviceServiceServicerImpl
+from device.Config import SERVICE_PORT, MAX_WORKERS, GRACE_PERIOD
+
+BIND_ADDRESS = '0.0.0.0'
+LOGGER = logging.getLogger(__name__)
+
+class DeviceService:
+    def __init__(self, database, address=BIND_ADDRESS, port=SERVICE_PORT, max_workers=MAX_WORKERS,
+                 grace_period=GRACE_PERIOD):
+        self.database = database
+        self.address = address
+        self.port = port
+        self.endpoint = None
+        self.max_workers = max_workers
+        self.grace_period = grace_period
+        self.device_servicer = None
+        self.health_servicer = None
+        self.pool = None
+        self.server = None
+
+    def start(self):
+        self.endpoint = '{}:{}'.format(self.address, self.port)
+        LOGGER.debug('Starting Service (tentative endpoint: {}, max_workers: {})...'.format(
+            self.endpoint, self.max_workers))
+
+        self.pool = futures.ThreadPoolExecutor(max_workers=self.max_workers)
+        self.server = grpc.server(self.pool) # , interceptors=(tracer_interceptor,))
+
+        self.device_servicer = DeviceServiceServicerImpl(self.database)
+        add_DeviceServiceServicer_to_server(self.device_servicer, self.server)
+
+        self.health_servicer = HealthServicer(
+            experimental_non_blocking=True, experimental_thread_pool=futures.ThreadPoolExecutor(max_workers=1))
+        add_HealthServicer_to_server(self.health_servicer, self.server)
+
+        port = self.server.add_insecure_port(self.endpoint)
+        self.endpoint = '{}:{}'.format(self.address, port)
+        LOGGER.info('Listening on {}...'.format(self.endpoint))
+        self.server.start()
+        self.health_servicer.set(OVERALL_HEALTH, HealthCheckResponse.SERVING) # pylint: disable=maybe-no-member
+
+        LOGGER.debug('Service started')
+
+    def stop(self):
+        LOGGER.debug('Stopping service (grace period {} seconds)...'.format(self.grace_period))
+        self.health_servicer.enter_graceful_shutdown()
+        self.server.stop(self.grace_period)
+        LOGGER.debug('Service stopped')

src/device/service/DeviceServiceServicerImpl.py

+import grpc, logging
+from prometheus_client import Counter, Histogram
+from google.protobuf.json_format import MessageToDict
+from context.proto.context_pb2 import DeviceId, Empty
+from device.proto.device_pb2_grpc import DeviceServiceServicer
+
+LOGGER = logging.getLogger(__name__)
+
+ADDDEVICE_COUNTER_STARTED    = Counter  ('device_adddevice_counter_started',
+                                          'Device:AddDevice counter of requests started'  )
+ADDDEVICE_COUNTER_COMPLETED  = Counter  ('device_adddevice_counter_completed',
+                                          'Device:AddDevice counter of requests completed')
+ADDDEVICE_COUNTER_FAILED     = Counter  ('device_adddevice_counter_failed',
+                                          'Device:AddDevice counter of requests failed'   )
+ADDDEVICE_HISTOGRAM_DURATION = Histogram('device_adddevice_histogram_duration',
+                                          'Device:AddDevice histogram of request duration')
+
+CONFIGUREDEVICE_COUNTER_STARTED    = Counter  ('device_configuredevice_counter_started',
+                                               'Device:ConfigureDevice counter of requests started'  )
+CONFIGUREDEVICE_COUNTER_COMPLETED  = Counter  ('device_configuredevice_counter_completed',
+                                               'Device:ConfigureDevice counter of requests completed')
+CONFIGUREDEVICE_COUNTER_FAILED     = Counter  ('device_configuredevice_counter_failed',
+                                               'Device:ConfigureDevice counter of requests failed'   )
+CONFIGUREDEVICE_HISTOGRAM_DURATION = Histogram('device_configuredevice_histogram_duration',
+                                               'Device:ConfigureDevice histogram of request duration')
+
+DELETEDEVICE_COUNTER_STARTED    = Counter  ('device_deletedevice_counter_started',
+                                            'Device:DeleteDevice counter of requests started'  )
+DELETEDEVICE_COUNTER_COMPLETED  = Counter  ('device_deletedevice_counter_completed',
+                                            'Device:DeleteDevice counter of requests completed')
+DELETEDEVICE_COUNTER_FAILED     = Counter  ('device_deletedevice_counter_failed',
+                                            'Device:DeleteDevice counter of requests failed'   )
+DELETEDEVICE_HISTOGRAM_DURATION = Histogram('device_deletedevice_histogram_duration',
+                                            'Device:DeleteDevice histogram of request duration')
+
+class DeviceServiceServicerImpl(DeviceServiceServicer):
+    def __init__(self, database):
+        LOGGER.debug('Creating Servicer...')
+        self.database = database
+        LOGGER.debug('Servicer Created')
+
+    @ADDDEVICE_HISTOGRAM_DURATION.time()
+    def AddDevice(self, request, context):
+        # request=context.Device(), returns=context.DeviceId()
+        ADDDEVICE_COUNTER_STARTED.inc()
+        try:
+            LOGGER.info('AddDevice request: {}'.format(str(request)))
+            reply = DeviceId(**self.database.add_device(MessageToDict(request)))
+            LOGGER.info('AddDevice reply: {}'.format(str(reply)))
+            ADDDEVICE_COUNTER_COMPLETED.inc()
+            return reply
+        except:
+            LOGGER.exception('AddDevice exception')
+            ADDDEVICE_COUNTER_FAILED.inc()
+            context.set_code(grpc.StatusCode.INTERNAL)
+            return DeviceId()
+
+    @CONFIGUREDEVICE_HISTOGRAM_DURATION.time()
+    def ConfigureDevice(self, request, context):
+        # request=context.DeviceConfig(), returns=context.DeviceId()
+        CONFIGUREDEVICE_COUNTER_STARTED.inc()
+        try:
+            LOGGER.info('ConfigureDevice request: {}'.format(str(request)))
+            reply = DeviceId(**self.database.configure_device(MessageToDict(request)))
+            LOGGER.info('ConfigureDevice reply: {}'.format(str(reply)))
+            CONFIGUREDEVICE_COUNTER_COMPLETED.inc()
+            return reply
+        except:
+            LOGGER.exception('ConfigureDevice exception')
+            CONFIGUREDEVICE_COUNTER_FAILED.inc()
+            context.set_code(grpc.StatusCode.INTERNAL)
+            return DeviceId()
+
+    @DELETEDEVICE_HISTOGRAM_DURATION.time()
+    def DeleteDevice(self, request, context):
+        # request=context.DeviceId(), returns=context.Empty()
+        DELETEDEVICE_COUNTER_STARTED.inc()
+        try:
+            LOGGER.info('DeleteDevice request: {}'.format(str(request)))
+            reply = Empty(**self.database.delete_device(MessageToDict(request)))
+            LOGGER.info('DeleteDevice reply: {}'.format(str(reply)))
+            DELETEDEVICE_COUNTER_COMPLETED.inc()
+            return reply
+        except:
+            LOGGER.exception('DeleteDevice exception')
+            DELETEDEVICE_COUNTER_FAILED.inc()
+            context.set_code(grpc.StatusCode.INTERNAL)
+            return Empty()

src/device/service/__main__.py

+import logging, os, signal, sys, threading
+from prometheus_client import start_http_server
+from device.database.Factory import get_database
+from device.service.DeviceService import DeviceService
+from device.Config import SERVICE_PORT, MAX_WORKERS, GRACE_PERIOD, LOG_LEVEL, METRICS_PORT
+
+terminate = threading.Event()
+logger = None
+
+def signal_handler(signal, frame):
+    global terminate, logger
+    logger.warning('Terminate signal received')
+    terminate.set()
+
+def main():
+    global terminate, logger
+
+    service_port = os.environ.get('DEVICESERVICE_SERVICE_PORT_GRPC', SERVICE_PORT)
+    max_workers  = os.environ.get('MAX_WORKERS',  MAX_WORKERS )
+    grace_period = os.environ.get('GRACE_PERIOD', GRACE_PERIOD)
+    log_level    = os.environ.get('LOG_LEVEL',    LOG_LEVEL   )
+    metrics_port = os.environ.get('METRICS_PORT', METRICS_PORT)
+
+    logging.basicConfig(level=log_level)
+    logger = logging.getLogger(__name__)
+
+    signal.signal(signal.SIGINT,  signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    logger.info('Starting...')
+
+    # Start metrics server
+    start_http_server(metrics_port)
+
+    # Get database instance
+    database = get_database()
+
+    # Starting device service
+    service = DeviceService(database, port=service_port, max_workers=max_workers, grace_period=grace_period)
+    service.start()
+
+    # Wait for Ctrl+C or termination signal
+    while not terminate.wait(0.1): pass
+
+    logger.info('Terminating...')
+    service.stop()
+
+    logger.info('Bye')
+    return(0)
+
+if __name__ == '__main__':
+    sys.exit(main())

src/device/tests/Dockerfile

+FROM device_service:develop
+
+# Run integration tests
+ENTRYPOINT ["pytest", "-v", "--log-level=DEBUG", "device/tests/test_integration.py"]

src/device/tests/test_integration.py

+import logging, os, pytest, sys
+
+from pathlib import Path
+sys.path.append(__file__.split('src')[0] + 'src')
+print(sys.path)
+
+from context.client.ContextClient import ContextClient
+from context.proto.context_pb2 import Empty
+from .tools.ValidateTopology import validate_topology_dict
+
+LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(logging.DEBUG)
+
+@pytest.fixture(scope='session')
+def remote_context_client():
+    address = os.environ.get('TEST_TARGET_ADDRESS')
+    if(address is None): raise Exception('EnvironmentVariable(TEST_TARGET_ADDRESS) not specified')
+    port = os.environ.get('TEST_TARGET_PORT')
+    if(port is None): raise Exception('EnvironmentVariable(TEST_TARGET_PORT) not specified')
+    return ContextClient(address=address, port=port)
+
+def test_remote_get_topology(remote_context_client):
+    response = remote_context_client.GetTopology(Empty())
+    validate_topology_dict(response)

src/device/tests/test_unitary.py

+import logging, pytest, sys
+
+from pathlib import Path
+sys.path.append(__file__.split('src')[0] + 'src')
+print(sys.path)
+
+from context.client.ContextClient import ContextClient
+from context.database.Factory import get_database, DatabaseEngineEnum
+from context.proto.context_pb2 import Empty
+from context.service.ContextService import ContextService
+from context.Config import SERVICE_PORT, MAX_WORKERS, GRACE_PERIOD
+from context.tests.tools.ValidateTopology import validate_topology_dict
+
+LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(logging.DEBUG)
+
+@pytest.fixture(scope='session')
+def local_context_service():
+    database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json')
+    _service = ContextService(database, port=SERVICE_PORT, max_workers=MAX_WORKERS, grace_period=GRACE_PERIOD)
+    _service.start()
+    yield _service
+    _service.stop()
+
+@pytest.fixture(scope='session')
+def local_context_client(local_context_service):
+    return ContextClient(address='127.0.0.1', port=SERVICE_PORT)
+
+def test_local_get_topology(local_context_client):
+    response = local_context_client.GetTopology(Empty())
+    validate_topology_dict(response)