Newer
Older
import copy, grpc, logging, pytest
from src.common.database.api.context.OperationalStatus import OperationalStatus
from google.protobuf.json_format import MessageToDict
from common.database.Factory import get_database, DatabaseEngineEnum
from common.tests.Assertions import validate_device_id, validate_empty
from device.client.DeviceClient import DeviceClient
from device.proto.context_pb2 import Device, DeviceId
from device.service.DeviceService import DeviceService
from device.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
Lluis Gifre Renom
committed
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
DEVICE_ID = {'device_id': {'uuid': 'test-device-001'}}
DEVICE = {
'device_id': {'device_id': {'uuid': 'test-device-001'}},
'device_type': 'ROADM',
'device_config': {'device_config': ''},
'devOperationalStatus': 1,
'endpointList' : [
{
'port_id': {
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'test-device-001'}},
'port_id': {'uuid' : 'port-101'}
},
'port_type': 'LINE'
},
{
'port_id': {
'topoId': {
'contextId': {'contextUuid': {'uuid': 'admin'}},
'topoId': {'uuid': 'admin'}
},
'dev_id': {'device_id': {'uuid': 'test-device-001'}},
'port_id': {'uuid' : 'port-102'}
},
'port_type': 'LINE'
},
]
}
Lluis Gifre Renom
committed
@pytest.fixture(scope='session')
Lluis Gifre Renom
committed
database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json')
_service = DeviceService(
database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
Lluis Gifre Renom
committed
_service.start()
yield _service
_service.stop()
@pytest.fixture(scope='session')
_client = DeviceClient(address='127.0.0.1', port=GRPC_SERVICE_PORT)
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
yield _client
_client.close()
def test_create_empty_device_uuid(client : DeviceClient):
# should fail with device uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['device_id']['device_id']['uuid'] = ''
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'device_uuid() string is empty.'
def test_create_empty_device_type(client : DeviceClient):
# should fail with device type is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['device_type'] = ''
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'device_type() string is empty.'
def test_create_wrong_device_operational_status(client : DeviceClient):
# should fail with wrong device operational status
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['devOperationalStatus'] = OperationalStatus.KEEP_STATE.value
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Device has to be created with either ENABLED/DISABLED Operational State.',
'Use KEEP_STATE only in configure Device methods.',
])
assert e.value.details() == msg
def test_create_endpoint_wrong_context(client : DeviceClient):
# should fail with unsupported context
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_id']['topoId']['contextId']['contextUuid']['uuid'] = 'wrong-context'
request = Device(**copy_device)
LOGGER.warning('request = {}'.format(request))
client.AddDevice(request)
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Unsupported context_id(wrong-context) in endpoint #0.',
'Only default context_id(admin) is currently supported.',
'Optionally, leave field empty to use default context_id.',
])
assert e.value.details() == msg
def test_create_endpoint_wrong_topology(client : DeviceClient):
# should fail with unsupported topology
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_id']['topoId']['topoId']['uuid'] = 'wrong-topo'
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Unsupported topology_id(wrong-topo) in endpoint #0.',
'Only default topology_id(admin) is currently supported.',
'Optionally, leave field empty to use default topology_id.',
])
assert e.value.details() == msg
def test_create_endpoint_wrong_device(client : DeviceClient):
# should fail with wrong endpoint device
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_id']['dev_id']['device_id']['uuid'] = 'wrong-device'
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
msg = ' '.join([
'Wrong device_id(wrong-device) in endpoint #0.',
'Parent specified in message is device_id(test-device-001).',
'Optionally, leave field empty to use parent device_id.',
])
assert e.value.details() == msg
def test_create_empty_port_uuid(client : DeviceClient):
# should fail with port uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_id']['port_id']['uuid'] = ''
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'port_uuid() string is empty.'
def test_create_empty_port_type(client : DeviceClient):
# should fail with port type is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_type'] = ''
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'port_type() string is empty.'
def test_create_duplicate_port(client : DeviceClient):
# should fail with uplicate port in device
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][1]['port_id']['port_id']['uuid'] = 'port-101'
client.AddDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
assert e.value.details() == 'Duplicated port_id(port-101) in device_id(test-device-001).'
def test_create(client : DeviceClient):
# should work
validate_device_id(MessageToDict(
client.AddDevice(Device(**DEVICE)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_create_duplicate(client : DeviceClient):
# should fail with device already exists
with pytest.raises(grpc._channel._InactiveRpcError) as e:
client.AddDevice(Device(**DEVICE))
assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
assert e.value.details() == 'device_uuid(test-device-001) already exists.'
def test_delete_empty_uuid(client : DeviceClient):
# should fail with device uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device_id = copy.deepcopy(DEVICE_ID)
copy_device_id['device_id']['uuid'] = ''
client.DeleteDevice(DeviceId(**copy_device_id))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'device_uuid() string is empty.'
def test_delete_device_not_found(client : DeviceClient):
# should fail with device not found
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device_id = copy.deepcopy(DEVICE_ID)
copy_device_id['device_id']['uuid'] = 'wrong-device-id'
client.DeleteDevice(DeviceId(**copy_device_id))
assert e.value.code() == grpc.StatusCode.NOT_FOUND
assert e.value.details() == 'device_uuid(wrong-device-id) does not exist.'
def test_delete(client : DeviceClient):
# should work
validate_empty(MessageToDict(
client.DeleteDevice(DeviceId(**DEVICE_ID)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_configure_empty_device_uuid(client : DeviceClient):
# should fail with device uuid is empty
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['device_id']['device_id']['uuid'] = ''
client.ConfigureDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'device_uuid() string is empty.'
def test_configure_device_not_found(client : DeviceClient):
# should fail with device not found
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['device_id']['device_id']['uuid'] = 'wrong-device-id'
client.ConfigureDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.NOT_FOUND
assert e.value.details() == 'device_uuid(wrong-device-id) does not exist.'
def test_create_device_default_endpoint_context_topology(client : DeviceClient):
# should work
copy_device = copy.deepcopy(DEVICE)
copy_device['endpointList'][0]['port_id']['topoId']['contextId']['contextUuid']['uuid'] = ''
copy_device['endpointList'][0]['port_id']['topoId']['topoId']['uuid'] = ''
copy_device['endpointList'][0]['port_id']['dev_id']['device_id']['uuid'] = ''
validate_device_id(MessageToDict(
client.AddDevice(Device(**copy_device)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))
def test_configure_wrong_device_type(client : DeviceClient):
# should fail with device type is wrong
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['device_type'] = 'wrong-type'
client.ConfigureDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'Device(test-device-001) has Type(ROADM). Cannot be changed to Type(wrong-type).'
def test_configure_with_endpoints(client : DeviceClient):
# should fail with endpoints cannot be modified
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
client.ConfigureDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert e.value.details() == 'Endpoints belonging to Device(test-device-001) cannot be modified.'
def test_configure_no_change(client : DeviceClient):
# should fail with any change detected
with pytest.raises(grpc._channel._InactiveRpcError) as e:
copy_device = copy.deepcopy(DEVICE)
copy_device['devOperationalStatus'] = OperationalStatus.KEEP_STATE.value
copy_device['endpointList'].clear()
client.ConfigureDevice(Device(**copy_device))
assert e.value.code() == grpc.StatusCode.ABORTED
msg = ' '.join([
'Any change has been requested for Device(test-device-001).',
'Either specify a new configuration or a new device state.',
])
assert e.value.details() == msg
Lluis Gifre Renom
committed
def test_configure(client : DeviceClient):
# should work
copy_device = copy.deepcopy(DEVICE)
copy_device['device_config']['device_config'] = '<new_config/>'
copy_device['devOperationalStatus'] = OperationalStatus.DISABLED.value
copy_device['endpointList'].clear()
validate_device_id(MessageToDict(
client.ConfigureDevice(Device(**copy_device)),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False))