Commit 205d0633 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

DLT component: 

- Added requirements.in
- Added default config file
- Created objects for the unitary test
- Created scenario settings for the unitary test
parent ea50d792

src/dlt/connector/Config.py

0 → 100644
+14 −0
Original line number Diff line number Diff line 
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

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

src/dlt/connector/requirements.in

0 → 100644
+6 −0
Original line number Diff line number Diff line 
grpcio==1.43.0

grpcio-health-checking==1.43.0

prometheus-client==0.13.0

protobuf==3.19.3

pytest==6.2.5

pytest-benchmark==3.4.1

src/dlt/connector/tests/Objects.py

0 → 100644
+53 −0
Original line number Diff line number Diff line 
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

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



from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID

from common.tools.object_factory.Context import json_context, json_context_id

from common.tools.object_factory.Device import json_device_emulated_packet_router_disabled, json_device_id

from common.tools.object_factory.EndPoint import json_endpoints

from common.tools.object_factory.Link import compose_link

from common.tools.object_factory.Topology import json_topology, json_topology_id



def compose_device(

    device_uuid, endpoint_uuids, endpoint_type='copper', endpoint_topology_id=None, endpoint_sample_types=[]

):

    device_id = json_device_id(device_uuid)

    endpoints = [(endpoint_uuid, endpoint_type, endpoint_sample_types) for endpoint_uuid in endpoint_uuids]

    endpoints = json_endpoints(device_id, endpoints, topology_id=endpoint_topology_id)

    device = json_device_emulated_packet_router_disabled(device_uuid, endpoints=endpoints)

    return device_id, endpoints, device



# ----- Context --------------------------------------------------------------------------------------------------------

CONTEXT_ADMIN_ID = json_context_id(DEFAULT_CONTEXT_UUID)

CONTEXT_ADMIN    = json_context(DEFAULT_CONTEXT_UUID)



# ----- Topology -------------------------------------------------------------------------------------------------------

TOPOLOGY_ADMIN_ID = json_topology_id(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ADMIN_ID)

TOPOLOGY_ADMIN    = json_topology(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ADMIN_ID)



# ----- Devices --------------------------------------------------------------------------------------------------------

DEVICE_DEV1_ID, DEVICE_DEV1_ENDPOINTS, DEVICE_DEV1 = compose_device('DEV1', ['1', '2'])

DEVICE_DEV2_ID, DEVICE_DEV2_ENDPOINTS, DEVICE_DEV2 = compose_device('DEV2', ['1', '2'])

DEVICE_DEV3_ID, DEVICE_DEV3_ENDPOINTS, DEVICE_DEV3 = compose_device('DEV3', ['1', '2'])



# ----- Links ----------------------------------------------------------------------------------------------------------

LINK_DEV1_DEV2_ID, LINK_DEV1_DEV2 = compose_link(DEVICE_DEV1_ENDPOINTS[0], DEVICE_DEV2_ENDPOINTS[0])

LINK_DEV1_DEV3_ID, LINK_DEV1_DEV3 = compose_link(DEVICE_DEV1_ENDPOINTS[1], DEVICE_DEV3_ENDPOINTS[0])

LINK_DEV2_DEV3_ID, LINK_DEV2_DEV3 = compose_link(DEVICE_DEV2_ENDPOINTS[1], DEVICE_DEV3_ENDPOINTS[1])



# ----- Containers -----------------------------------------------------------------------------------------------------

CONTEXTS   = [CONTEXT_ADMIN]

TOPOLOGIES = [TOPOLOGY_ADMIN]

DEVICES    = [DEVICE_DEV1, DEVICE_DEV2, DEVICE_DEV3]

LINKS      = [LINK_DEV1_DEV2, LINK_DEV1_DEV3, LINK_DEV2_DEV3]

src/dlt/connector/tests/PrepareTestScenario.py

0 → 100644
+50 −0
Original line number Diff line number Diff line 
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

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



import os, pytest

from typing import Tuple

from common.Constants import ServiceNameEnum

from common.Settings import (

    ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_service_port_grpc)

from common.orm.Database import Database

from common.orm.Factory import get_database_backend, BackendEnum as DatabaseBackendEnum

from common.message_broker.Factory import get_messagebroker_backend, BackendEnum as MessageBrokerBackendEnum

from common.message_broker.MessageBroker import MessageBroker

from context.client.ContextClient import ContextClient

from context.service.grpc_server.ContextService import ContextService



LOCAL_HOST = '127.0.0.1'

GRPC_PORT = 10000 + get_service_port_grpc(ServiceNameEnum.CONTEXT) # avoid privileged ports

os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST     )] = str(LOCAL_HOST)

os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(GRPC_PORT)



@pytest.fixture(scope='session')

def context_db_mb() -> Tuple[Database, MessageBroker]:

    _database = Database(get_database_backend(backend=DatabaseBackendEnum.INMEMORY))

    _message_broker = MessageBroker(get_messagebroker_backend(backend=MessageBrokerBackendEnum.INMEMORY))

    yield _database, _message_broker

    _message_broker.terminate()



@pytest.fixture(scope='session')

def context_service(context_db_mb : Tuple[Database, MessageBroker]): # pylint: disable=redefined-outer-name

    _service = ContextService(context_db_mb[0], context_db_mb[1])

    _service.start()

    yield _service

    _service.stop()



@pytest.fixture(scope='session')

def context_client(context_service : ContextService): # pylint: disable=redefined-outer-name

    _client = ContextClient()

    yield _client

    _client.close()