1 # Copyright (C) 2021 Wind River Systems, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from unittest.mock import MagicMock
18 from o2ims.domain import ocloud, subscription_obj, configuration_obj
19 from o2ims.domain import resource_type as rt
20 from o2ims.views import ocloud_view
21 from o2common.config import config
25 ocloudid1 = str(uuid.uuid4())
26 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
27 config.get_api_url(), "ocloud for unit test", 1)
31 def test_new_ocloud():
32 ocloudid1 = str(uuid.uuid4())
33 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
34 config.get_api_url(), "ocloud for unit test", 1)
35 assert ocloudid1 is not None and ocloud1.oCloudId == ocloudid1
38 # def test_add_ocloud_with_dms():
39 # ocloud1 = setup_ocloud()
40 # dmsid = str(uuid.uuid4())
41 # dms = ocloud.DeploymentManager(
42 # dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
43 # ocloud1.addDeploymentManager(dms)
44 # ocloud1.addDeploymentManager(dms)
45 # assert len(ocloud1.deploymentManagers) == 1
46 # # repo.update(ocloud1.oCloudId, {
47 # # "deploymentManagers": ocloud1.deploymentManagers})
50 def test_new_resource_type():
51 ocloud1 = setup_ocloud()
52 resource_type_id1 = str(uuid.uuid4())
53 resource_type1 = ocloud.ResourceType(
54 resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
56 assert resource_type_id1 is not None and \
57 resource_type1.resourceTypeId == resource_type_id1
60 def test_new_resource_pool():
61 ocloud1 = setup_ocloud()
62 resource_pool_id1 = str(uuid.uuid4())
63 resource_pool1 = ocloud.ResourcePool(
64 resource_pool_id1, "resourcepool1", config.get_api_url(),
66 assert resource_pool_id1 is not None and \
67 resource_pool1.resourcePoolId == resource_pool_id1
70 def test_new_resource():
71 resource_id1 = str(uuid.uuid4())
72 resource_type_id1 = str(uuid.uuid4())
73 resource_pool_id1 = str(uuid.uuid4())
74 resource1 = ocloud.Resource(
75 resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
76 assert resource_id1 is not None and resource1.resourceId == resource_id1
79 def test_new_deployment_manager():
80 ocloud_id1 = str(uuid.uuid4())
81 deployment_manager_id1 = str(uuid.uuid4())
82 deployment_manager1 = ocloud.DeploymentManager(
83 deployment_manager_id1, "k8s1", ocloud_id1,
84 config.get_api_url()+"/k8s1")
85 assert deployment_manager_id1 is not None and deployment_manager1.\
86 deploymentManagerId == deployment_manager_id1
89 def test_new_subscription():
90 subscription_id1 = str(uuid.uuid4())
91 subscription1 = subscription_obj.Subscription(
92 subscription_id1, "https://callback/uri/write/here")
93 assert subscription_id1 is not None and\
94 subscription1.subscriptionId == subscription_id1
97 def test_new_configuration():
98 configuration_id1 = str(uuid.uuid4())
99 configuration1 = configuration_obj.Configuration(
100 configuration_id1, "https://callback/uri/write/here",
102 assert configuration_id1 is not None and\
103 configuration1.configurationId == configuration_id1
106 def test_view_olcouds(mock_uow):
107 session, uow = mock_uow
109 ocloud1_UUID = str(uuid.uuid4)
110 ocloud1 = MagicMock()
111 ocloud1.serialize.return_value = {
112 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
113 session.return_value.query.return_value = [ocloud1]
115 ocloud_list = ocloud_view.oclouds(uow)
116 # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
117 assert len(ocloud_list) == 1
120 def test_view_olcoud_one(mock_uow):
121 session, uow = mock_uow
123 ocloud1_UUID = str(uuid.uuid4)
124 session.return_value.query.return_value.filter_by.return_value.first.\
125 return_value.serialize.return_value = None
128 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
129 assert ocloud_res is None
131 session.return_value.query.return_value.filter_by.return_value.first.\
132 return_value.serialize.return_value = {
133 "oCloudId": ocloud1_UUID}
135 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
136 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
139 def test_view_resource_types(mock_uow):
140 session, uow = mock_uow
142 resource_type_id1 = str(uuid.uuid4())
143 restype1 = MagicMock()
144 restype1.serialize.return_value = {
145 "resourceTypeId": resource_type_id1}
146 session.return_value.query.return_value = [restype1]
148 resource_type_list = ocloud_view.resource_types(uow)
149 assert str(resource_type_list[0].get(
150 "resourceTypeId")) == resource_type_id1
153 def test_view_resource_type_one(mock_uow):
154 session, uow = mock_uow
156 resource_type_id1 = str(uuid.uuid4())
157 session.return_value.query.return_value.filter_by.return_value.first.\
158 return_value.serialize.return_value = None
161 resource_type_res = ocloud_view.resource_type_one(
162 resource_type_id1, uow)
163 assert resource_type_res is None
165 session.return_value.query.return_value.filter_by.return_value.first.\
166 return_value.serialize.return_value = {
167 "resourceTypeId": resource_type_id1}
169 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
170 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
173 def test_view_resource_pools(mock_uow):
174 session, uow = mock_uow
176 resource_pool_id1 = str(uuid.uuid4())
177 respool1 = MagicMock()
178 respool1.serialize.return_value = {
179 "resourcePoolId": resource_pool_id1}
180 session.return_value.query.return_value = [respool1]
182 resource_pool_list = ocloud_view.resource_pools(uow)
183 assert str(resource_pool_list[0].get(
184 "resourcePoolId")) == resource_pool_id1
187 def test_view_resource_pool_one(mock_uow):
188 session, uow = mock_uow
190 resource_pool_id1 = str(uuid.uuid4())
191 session.return_value.query.return_value.filter_by.return_value.first.\
192 return_value.serialize.return_value = None
195 resource_pool_res = ocloud_view.resource_pool_one(
196 resource_pool_id1, uow)
197 assert resource_pool_res is None
199 session.return_value.query.return_value.filter_by.return_value.first.\
200 return_value.serialize.return_value = {
201 "resourcePoolId": resource_pool_id1
204 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
205 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
208 def test_view_resources(mock_uow):
209 session, uow = mock_uow
211 resource_id1 = str(uuid.uuid4())
212 resource_pool_id1 = str(uuid.uuid4())
214 res1.serialize.return_value = {
215 "resourceId": resource_id1,
216 "resourcePoolId": resource_pool_id1
218 session.return_value.query.return_value.filter_by.return_value = [res1]
220 resource_list = ocloud_view.resources(resource_pool_id1, uow)
221 assert str(resource_list[0].get("resourceId")) == resource_id1
222 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
225 def test_view_resource_one(mock_uow):
226 session, uow = mock_uow
228 resource_id1 = str(uuid.uuid4())
229 resource_pool_id1 = str(uuid.uuid4())
230 session.return_value.query.return_value.filter_by.return_value.first.\
231 return_value.serialize.return_value = None
234 resource_res = ocloud_view.resource_one(resource_id1, uow)
235 assert resource_res is None
237 session.return_value.query.return_value.filter_by.return_value.first.\
238 return_value.serialize.return_value = {
239 "resourceId": resource_id1,
240 "resourcePoolId": resource_pool_id1
243 resource_res = ocloud_view.resource_one(resource_id1, uow)
244 assert str(resource_res.get("resourceId")) == resource_id1
247 def test_view_deployment_managers(mock_uow):
248 session, uow = mock_uow
250 deployment_manager_id1 = str(uuid.uuid4())
252 dm1.serialize.return_value = {
253 "deploymentManagerId": deployment_manager_id1,
255 session.return_value.query.return_value = [dm1]
257 deployment_manager_list = ocloud_view.deployment_managers(uow)
258 assert str(deployment_manager_list[0].get(
259 "deploymentManagerId")) == deployment_manager_id1
262 def test_view_deployment_manager_one(mock_uow):
263 session, uow = mock_uow
265 deployment_manager_id1 = str(uuid.uuid4())
266 session.return_value.query.return_value.filter_by.return_value.first.\
267 return_value.serialize.return_value = None
270 deployment_manager_res = ocloud_view.deployment_manager_one(
271 deployment_manager_id1, uow)
272 assert deployment_manager_res is None
274 session.return_value.query.return_value.filter_by.return_value.first.\
275 return_value.serialize.return_value = {
276 "deploymentManagerId": deployment_manager_id1,
279 deployment_manager_res = ocloud_view.deployment_manager_one(
280 deployment_manager_id1, uow)
281 assert str(deployment_manager_res.get(
282 "deploymentManagerId")) == deployment_manager_id1
285 def test_view_subscriptions(mock_uow):
286 session, uow = mock_uow
288 subscription_id1 = str(uuid.uuid4())
290 sub1.serialize.return_value = {
291 "subscriptionId": subscription_id1,
293 session.return_value.query.return_value = [sub1]
295 subscription_list = ocloud_view.subscriptions(uow)
296 assert str(subscription_list[0].get(
297 "subscriptionId")) == subscription_id1
300 def test_view_subscription_one(mock_uow):
301 session, uow = mock_uow
303 subscription_id1 = str(uuid.uuid4())
304 session.return_value.query.return_value.filter_by.return_value.first.\
305 return_value.serialize.return_value = None
308 subscription_res = ocloud_view.subscription_one(
309 subscription_id1, uow)
310 assert subscription_res is None
312 session.return_value.query.return_value.filter_by.return_value.first.\
313 return_value.serialize.return_value = {
314 "subscriptionId": subscription_id1,
317 subscription_res = ocloud_view.subscription_one(
318 subscription_id1, uow)
319 assert str(subscription_res.get(
320 "subscriptionId")) == subscription_id1
323 def test_flask_get_list(mock_flask_uow):
324 session, app = mock_flask_uow
325 session.query.return_value = []
326 apibase = config.get_o2ims_api_base()
328 with app.test_client() as client:
329 # Get list and return empty list
330 ##########################
331 resp = client.get(apibase+"/resourceTypes")
332 assert resp.get_data() == b'[]\n'
334 resp = client.get(apibase+"/resourcePools")
335 assert resp.get_data() == b'[]\n'
337 resource_pool_id1 = str(uuid.uuid4())
338 resp = client.get(apibase+"/resourcePools/" +
339 resource_pool_id1+"/resources")
340 assert resp.get_data() == b'[]\n'
342 resp = client.get(apibase+"/deploymentManagers")
343 assert resp.get_data() == b'[]\n'
345 resp = client.get(apibase+"/subscriptions")
346 assert resp.get_data() == b'[]\n'
349 def test_flask_get_one(mock_flask_uow):
350 session, app = mock_flask_uow
352 session.return_value.query.return_value.filter_by.return_value.\
353 first.return_value = None
354 apibase = config.get_o2ims_api_base()
356 with app.test_client() as client:
357 # Get one and return 404
358 ###########################
359 resp = client.get(apibase+"/")
360 assert resp.status_code == 404
362 resource_type_id1 = str(uuid.uuid4())
363 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
364 assert resp.status_code == 404
366 resource_pool_id1 = str(uuid.uuid4())
367 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
368 assert resp.status_code == 404
370 resource_id1 = str(uuid.uuid4())
371 resp = client.get(apibase+"/resourcePools/" +
372 resource_pool_id1+"/resources/"+resource_id1)
373 assert resp.status_code == 404
375 deployment_manager_id1 = str(uuid.uuid4())
376 resp = client.get(apibase+"/deploymentManagers/" +
377 deployment_manager_id1)
378 assert resp.status_code == 404
380 subscription_id1 = str(uuid.uuid4())
381 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
382 assert resp.status_code == 404
385 def test_flask_post(mock_flask_uow):
386 session, app = mock_flask_uow
387 apibase = config.get_o2ims_api_base()
389 with app.test_client() as client:
390 session.return_value.execute.return_value = []
392 sub_callback = 'http://subscription/callback/url'
393 resp = client.post(apibase+'/subscriptions', json={
394 'callback': sub_callback,
395 'consumerSubscriptionId': 'consumerSubId1',
398 assert resp.status_code == 201
399 assert 'subscriptionId' in resp.get_json()
402 def test_flask_delete(mock_flask_uow):
403 session, app = mock_flask_uow
404 apibase = config.get_o2ims_api_base()
406 with app.test_client() as client:
407 session.return_value.execute.return_value.first.return_value = {}
409 subscription_id1 = str(uuid.uuid4())
410 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
411 assert resp.status_code == 204
414 def test_flask_not_allowed(mock_flask_uow):
415 _, app = mock_flask_uow
416 apibase = config.get_o2ims_api_base()
418 with app.test_client() as client:
419 # Testing resource type not support method
420 ##########################
421 uri = apibase + "/resourceTypes"
422 resp = client.post(uri)
423 assert resp.status == '405 METHOD NOT ALLOWED'
424 resp = client.put(uri)
425 assert resp.status == '405 METHOD NOT ALLOWED'
426 resp = client.patch(uri)
427 assert resp.status == '405 METHOD NOT ALLOWED'
428 resp = client.delete(uri)
429 assert resp.status == '405 METHOD NOT ALLOWED'
431 resource_type_id1 = str(uuid.uuid4())
432 uri = apibase + "/resourceTypes/" + resource_type_id1
433 resp = client.post(uri)
434 assert resp.status == '405 METHOD NOT ALLOWED'
435 resp = client.put(uri)
436 assert resp.status == '405 METHOD NOT ALLOWED'
437 resp = client.patch(uri)
438 assert resp.status == '405 METHOD NOT ALLOWED'
439 resp = client.delete(uri)
440 assert resp.status == '405 METHOD NOT ALLOWED'
442 # Testing resource pool not support method
443 ##########################
444 uri = apibase + "/resourcePools"
445 resp = client.post(uri)
446 assert resp.status == '405 METHOD NOT ALLOWED'
447 resp = client.put(uri)
448 assert resp.status == '405 METHOD NOT ALLOWED'
449 resp = client.patch(uri)
450 assert resp.status == '405 METHOD NOT ALLOWED'
451 resp = client.delete(uri)
452 assert resp.status == '405 METHOD NOT ALLOWED'
454 resource_pool_id1 = str(uuid.uuid4())
455 uri = apibase + "/resourcePools/" + resource_pool_id1
456 resp = client.post(uri)
457 assert resp.status == '405 METHOD NOT ALLOWED'
458 resp = client.put(uri)
459 assert resp.status == '405 METHOD NOT ALLOWED'
460 resp = client.patch(uri)
461 assert resp.status == '405 METHOD NOT ALLOWED'
462 resp = client.delete(uri)
463 assert resp.status == '405 METHOD NOT ALLOWED'
465 # Testing resource not support method
466 ##########################
467 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
468 resp = client.post(uri)
469 assert resp.status == '405 METHOD NOT ALLOWED'
470 resp = client.put(uri)
471 assert resp.status == '405 METHOD NOT ALLOWED'
472 resp = client.patch(uri)
473 assert resp.status == '405 METHOD NOT ALLOWED'
474 resp = client.delete(uri)
475 assert resp.status == '405 METHOD NOT ALLOWED'
477 resource_id1 = str(uuid.uuid4())
478 uri = apibase + "/resourcePools/" + \
479 resource_pool_id1 + "/resources/" + resource_id1
480 resp = client.post(uri)
481 assert resp.status == '405 METHOD NOT ALLOWED'
482 resp = client.put(uri)
483 assert resp.status == '405 METHOD NOT ALLOWED'
484 resp = client.patch(uri)
485 assert resp.status == '405 METHOD NOT ALLOWED'
486 resp = client.delete(uri)
487 assert resp.status == '405 METHOD NOT ALLOWED'
489 # Testing deployment managers not support method
490 ##########################
491 uri = apibase + "/deploymentManagers"
492 resp = client.post(uri)
493 assert resp.status == '405 METHOD NOT ALLOWED'
494 resp = client.put(uri)
495 assert resp.status == '405 METHOD NOT ALLOWED'
496 resp = client.patch(uri)
497 assert resp.status == '405 METHOD NOT ALLOWED'
498 resp = client.delete(uri)
499 assert resp.status == '405 METHOD NOT ALLOWED'
501 deployment_manager_id1 = str(uuid.uuid4())
502 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
503 resp = client.post(uri)
504 assert resp.status == '405 METHOD NOT ALLOWED'
505 resp = client.put(uri)
506 assert resp.status == '405 METHOD NOT ALLOWED'
507 resp = client.patch(uri)
508 assert resp.status == '405 METHOD NOT ALLOWED'
509 resp = client.delete(uri)
510 assert resp.status == '405 METHOD NOT ALLOWED'
512 # Testing subscriptions not support method
513 ##########################
514 uri = apibase + "/subscriptions"
515 resp = client.put(uri)
516 assert resp.status == '405 METHOD NOT ALLOWED'
517 resp = client.patch(uri)
518 assert resp.status == '405 METHOD NOT ALLOWED'
519 resp = client.delete(uri)
520 assert resp.status == '405 METHOD NOT ALLOWED'
522 subscription_id1 = str(uuid.uuid4())
523 uri = apibase + "/subscriptions/" + subscription_id1
524 resp = client.post(uri)
525 assert resp.status == '405 METHOD NOT ALLOWED'
526 resp = client.put(uri)
527 assert resp.status == '405 METHOD NOT ALLOWED'
528 resp = client.patch(uri)
529 assert resp.status == '405 METHOD NOT ALLOWED'