Newer
Older
import logging, os, pytest
from google.protobuf.json_format import MessageToDict
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_topology_has_devices, \
validate_topology_has_links, validate_topology_is_empty
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from context.client.ContextClient import ContextClient
from context.proto.context_pb2 import Device, Empty, Link
from device.client.DeviceClient import DeviceClient
from .definitions import DEVICE_DEV1, DEVICE_DEV2, DEVICE_DEV3, LINK_DEV1_DEV2, LINK_DEV1_DEV3, LINK_DEV2_DEV1, \
LINK_DEV2_DEV3, LINK_DEV3_DEV1, LINK_DEV3_DEV2
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
def get_setting(name):
value = os.environ.get(name)
if value is None: raise Exception('Unable to find variable({})'.format(name))
return value
@pytest.fixture(scope='session')
def redis_database():
_database = get_database(engine=DatabaseEngineEnum.REDIS)
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()
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(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))