Newer
Older
from google.protobuf.json_format import MessageToDict
from common.database.Factory import get_database, DatabaseEngineEnum
from common.database.api.Database import Database
from common.database.tests.script import populate_example
from common.tests.Assertions import validate_empty, validate_service, validate_service_id, validate_service_list
from service.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
from service.client.ServiceClient import ServiceClient
from service.proto.context_pb2 import Empty
from service.proto.service_pb2 import Service, ServiceId, ServiceStateEnum, ServiceType
from service.service.ServiceService import ServiceService
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
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
SERVICE_ID = {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'cs_id': {'uuid': 'DEV1'},
}
SERVICE = {
'cs_id': SERVICE_ID,
'serviceType': ServiceType.L3NM,
'serviceConfig': {'serviceConfig': '<config/>'},
'serviceState': {'serviceState': ServiceStateEnum.PLANNED},
'endpointList' : [
{
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'DEV1'}},
'port_id': {'uuid' : 'EP5'}
},
{
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'DEV2'}},
'port_id': {'uuid' : 'EP5'}
},
{
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'DEV3'}},
'port_id': {'uuid' : 'EP5'}
},
]
}
_database = get_database(engine=DatabaseEngineEnum.INMEMORY)
return _database
@pytest.fixture(scope='session')
database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
_service.start()
yield _service
_service.stop()
@pytest.fixture(scope='session')
def service_client(service_service):
_client = ServiceClient(address='127.0.0.1', port=GRPC_SERVICE_PORT)
yield _client
_client.close()
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
def test_get_services_empty(service_client : ServiceClient, database : Database):
# should work
populate_example(database, add_services=False)
validate_service_list(MessageToDict(
service_client.GetServiceList(Empty()),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_create_service(service_client : ServiceClient):
# should work
validate_service_id(MessageToDict(
service_client.CreateService(Service(**SERVICE)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_get_service(service_client : ServiceClient):
# should work
validate_service(MessageToDict(
service_client.GetServiceById(ServiceId(**SERVICE_ID)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_delete_service(service_client : ServiceClient):
# should work
validate_empty(MessageToDict(
service_client.DeleteService(ServiceId(**SERVICE_ID)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_get_services_full(service_client : ServiceClient, database : Database):
# should work
database.clear_all()
populate_example(database)
validate_service_list(MessageToDict(
service_client.GetServiceList(Empty()),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))