Newer
Older
1
2
3
4
5
6
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
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import logging, grpc
import os
import sqlite3
import pytest
from typing import Tuple
from interdomain.proto import context_pb2, kpi_sample_types_pb2, monitoring_pb2
from interdomain.client.interdomain_client import InterdomainClient
from interdomain.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
from interdomain.service.InterdomainService import InterdomainService
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
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
###########################
# Tests Setup
###########################
SERVER_ADDRESS = '127.0.0.1'
LISTEN_ADDRESS = '[::]'
GRPC_PORT_MONITORING = 9090
GRPC_PORT_CONTEXT = 10000 + grpc_port_context # avoid privileged ports
SCENARIOS = [ # comment/uncomment scenarios to activate/deactivate them in the test unit
('all_inmemory', DatabaseBackendEnum.INMEMORY, {}, MessageBrokerBackendEnum.INMEMORY, {} ),
]
# This fixture will be requested by test cases and last during testing session
@pytest.fixture(scope='session')
def interdomain_service():
LOGGER.warning('interdomain_service begin')
interdomain_port = GRPC_INTERDOMAIN_PORT
max_workers = GRPC_MAX_WORKERS
grace_period = GRPC_GRACE_PERIOD
LOGGER.info('Initializing InterdomainService...')
grpc_service = InterdomainService(port=interdomain_port, max_workers=max_workers, grace_period=grace_period)
server = grpc_service.start()
# yield the server, when test finishes, execution will resume to stop it
LOGGER.warning('interdomain_service yielding')
yield server
LOGGER.info('Terminating InterdomainService...')
grpc_service.stop()
# 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 interdomain_client(interdomain_service):
LOGGER.warning('interdomain_client begin')
client = InterdomainClient(server=SERVER_ADDRESS, port=GRPC_PORT_INTERDOMAIN) # instantiate the client
LOGGER.warning('interdomain_client returning')
return client
# This fixture will be requested by test cases and last during testing session.
@pytest.fixture(scope='session')
def create_TeraFlowController():
LOGGER.warning('create_TeraFlowController begin')
# form request
tf_ctl = context_pb2.TeraFlowController()
tf_ctl.context_id = context_pb2.ContextId()
tf_ctl.context_id.context_uuid = context_pb2.Uuid()
tf_ctl.context_id.context_uuid.uuid = str(1)
tf_ctl.ip_address = "127.0.0.1"
tf_ctl.port = 9090
return tf_ctl
@pytest.fixture(scope='session')
def create_TransportSlice():
LOGGER.warning('create_TransportSlice begin')
# form request
slice_req = slice_pb2.TransportSlice()
slice_req.contextId = context_pb2.ContextId()
slice_req.contextId.context_uuid = context_pb2.Uuid()
slice_req.contextId.context_uuid.uuid = str(1)
slice_req.slice_id = context_pb2.Uuid()
slice_req.slice_id.context_uuid.uuid = str(1)
return slice_req
###########################
# Tests Implementation
###########################
# Test case that makes use of client fixture to test server's CreateKpi method
def test_Authenticate(interdomain_client,create_TeraFlowController):
# make call to server
LOGGER.warning('test_Authenticate requesting')
response = interdomain_client.Authenticate(create_TeraFlowController)
LOGGER.debug(str(response))
assert isinstance(response, context.AuthenticationResult)
# Test case that makes use of client fixture to test server's MonitorKpi method
def test_LookUpSlice(interdomain_client,create_TransportSlice):
LOGGER.warning('test_LookUpSlice begin')
response = interdomain_client.LookUpSlice(create_TransportSlice)
LOGGER.debug(str(response))
assert isinstance(response, slice.SliceId)
# Test case that makes use of client fixture to test server's GetStreamKpi method
def test_CreateSliceAndAddToCatalog(interdomain_client,create_TransportSlice):
LOGGER.warning('test_CreateSliceAndAddToCatalog begin')
response = interdomain_client.CreateSliceAndAddToCatalog(create_TransportSlice)
LOGGER.debug(str(response))
assert isinstance(response, slice.SliceId)
# Test case that makes use of client fixture to test server's IncludeKpi method
def test_OrderSliceFromCatalog(interdomain_client,create_TransportSlice):
# make call to server
LOGGER.warning('test_OrderSliceFromCatalog requesting')
response = interdomain_client.OrderSliceFromCatalog(create_TransportSlice)
LOGGER.debug(str(response))
assert isinstance(response, slice.SliceId)