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
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_registration():
98 registration_id1 = str(uuid.uuid4())
99 registration1 = subscription_obj.Registration(
100 registration_id1, "https://callback/uri/write/here")
101 assert registration_id1 is not None and\
102 registration1.registrationId == registration_id1
105 def test_view_olcouds(mock_uow):
106 session, uow = mock_uow
108 ocloud1_UUID = str(uuid.uuid4)
109 ocloud1 = MagicMock()
110 ocloud1.serialize.return_value = {
111 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
112 session.return_value.query.return_value = [ocloud1]
114 ocloud_list = ocloud_view.oclouds(uow)
115 # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
116 assert len(ocloud_list) == 1
119 def test_view_olcoud_one(mock_uow):
120 session, uow = mock_uow
122 ocloud1_UUID = str(uuid.uuid4)
123 session.return_value.query.return_value.filter_by.return_value.first.\
124 return_value.serialize.return_value = None
127 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
128 assert ocloud_res is None
130 session.return_value.query.return_value.filter_by.return_value.first.\
131 return_value.serialize.return_value = {
132 "oCloudId": ocloud1_UUID}
134 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
135 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
138 def test_view_resource_types(mock_uow):
139 session, uow = mock_uow
141 resource_type_id1 = str(uuid.uuid4())
142 restype1 = MagicMock()
143 restype1.serialize.return_value = {
144 "resourceTypeId": resource_type_id1}
145 session.return_value.query.return_value = [restype1]
147 resource_type_list = ocloud_view.resource_types(uow)
148 assert str(resource_type_list[0].get(
149 "resourceTypeId")) == resource_type_id1
152 def test_view_resource_type_one(mock_uow):
153 session, uow = mock_uow
155 resource_type_id1 = str(uuid.uuid4())
156 session.return_value.query.return_value.filter_by.return_value.first.\
157 return_value.serialize.return_value = None
160 resource_type_res = ocloud_view.resource_type_one(
161 resource_type_id1, uow)
162 assert resource_type_res is None
164 session.return_value.query.return_value.filter_by.return_value.first.\
165 return_value.serialize.return_value = {
166 "resourceTypeId": resource_type_id1}
168 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
169 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
172 def test_view_resource_pools(mock_uow):
173 session, uow = mock_uow
175 resource_pool_id1 = str(uuid.uuid4())
176 respool1 = MagicMock()
177 respool1.serialize.return_value = {
178 "resourcePoolId": resource_pool_id1}
179 session.return_value.query.return_value = [respool1]
181 resource_pool_list = ocloud_view.resource_pools(uow)
182 assert str(resource_pool_list[0].get(
183 "resourcePoolId")) == resource_pool_id1
186 def test_view_resource_pool_one(mock_uow):
187 session, uow = mock_uow
189 resource_pool_id1 = str(uuid.uuid4())
190 session.return_value.query.return_value.filter_by.return_value.first.\
191 return_value.serialize.return_value = None
194 resource_pool_res = ocloud_view.resource_pool_one(
195 resource_pool_id1, uow)
196 assert resource_pool_res is None
198 session.return_value.query.return_value.filter_by.return_value.first.\
199 return_value.serialize.return_value = {
200 "resourcePoolId": resource_pool_id1
203 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
204 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
207 def test_view_resources(mock_uow):
208 session, uow = mock_uow
210 resource_id1 = str(uuid.uuid4())
211 resource_pool_id1 = str(uuid.uuid4())
213 res1.serialize.return_value = {
214 "resourceId": resource_id1,
215 "resourcePoolId": resource_pool_id1
217 session.return_value.query.return_value.filter_by.return_value = [res1]
219 resource_list = ocloud_view.resources(resource_pool_id1, uow)
220 assert str(resource_list[0].get("resourceId")) == resource_id1
221 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
224 def test_view_resource_one(mock_uow):
225 session, uow = mock_uow
227 resource_id1 = str(uuid.uuid4())
228 resource_pool_id1 = str(uuid.uuid4())
229 session.return_value.query.return_value.filter_by.return_value.first.\
230 return_value.serialize.return_value = None
233 resource_res = ocloud_view.resource_one(resource_id1, uow)
234 assert resource_res is None
236 session.return_value.query.return_value.filter_by.return_value.first.\
237 return_value.serialize.return_value = {
238 "resourceId": resource_id1,
239 "resourcePoolId": resource_pool_id1
242 resource_res = ocloud_view.resource_one(resource_id1, uow)
243 assert str(resource_res.get("resourceId")) == resource_id1
246 def test_view_deployment_managers(mock_uow):
247 session, uow = mock_uow
249 deployment_manager_id1 = str(uuid.uuid4())
251 dm1.serialize.return_value = {
252 "deploymentManagerId": deployment_manager_id1,
254 session.return_value.query.return_value = [dm1]
256 deployment_manager_list = ocloud_view.deployment_managers(uow)
257 assert str(deployment_manager_list[0].get(
258 "deploymentManagerId")) == deployment_manager_id1
261 def test_view_deployment_manager_one(mock_uow):
262 session, uow = mock_uow
264 deployment_manager_id1 = str(uuid.uuid4())
265 session.return_value.query.return_value.filter_by.return_value.first.\
266 return_value.serialize.return_value = None
269 deployment_manager_res = ocloud_view.deployment_manager_one(
270 deployment_manager_id1, uow)
271 assert deployment_manager_res is None
273 session.return_value.query.return_value.filter_by.return_value.first.\
274 return_value.serialize.return_value = {
275 "deploymentManagerId": deployment_manager_id1,
278 deployment_manager_res = ocloud_view.deployment_manager_one(
279 deployment_manager_id1, uow)
280 assert str(deployment_manager_res.get(
281 "deploymentManagerId")) == deployment_manager_id1
284 def test_view_subscriptions(mock_uow):
285 session, uow = mock_uow
287 subscription_id1 = str(uuid.uuid4())
289 sub1.serialize.return_value = {
290 "subscriptionId": subscription_id1,
292 session.return_value.query.return_value = [sub1]
294 subscription_list = ocloud_view.subscriptions(uow)
295 assert str(subscription_list[0].get(
296 "subscriptionId")) == subscription_id1
299 def test_view_subscription_one(mock_uow):
300 session, uow = mock_uow
302 subscription_id1 = str(uuid.uuid4())
303 session.return_value.query.return_value.filter_by.return_value.first.\
304 return_value.serialize.return_value = None
307 subscription_res = ocloud_view.subscription_one(
308 subscription_id1, uow)
309 assert subscription_res is None
311 session.return_value.query.return_value.filter_by.return_value.first.\
312 return_value.serialize.return_value = {
313 "subscriptionId": subscription_id1,
316 subscription_res = ocloud_view.subscription_one(
317 subscription_id1, uow)
318 assert str(subscription_res.get(
319 "subscriptionId")) == subscription_id1
322 def test_view_registrations(mock_uow):
323 session, uow = mock_uow
325 registration_id1 = str(uuid.uuid4())
327 reg1.serialize.return_value = {
328 "registrationId": registration_id1,
330 session.return_value.query.return_value = [reg1]
332 registration_list = ocloud_view.registrations(uow)
333 assert str(registration_list[0].get(
334 "registrationId")) == registration_id1
337 def test_view_registration_one(mock_uow):
338 session, uow = mock_uow
340 registration_id1 = str(uuid.uuid4())
341 session.return_value.query.return_value.filter_by.return_value.first.\
342 return_value.serialize.return_value = None
345 registration_res = ocloud_view.registration_one(
346 registration_id1, uow)
347 assert registration_res is None
349 session.return_value.query.return_value.filter_by.return_value.first.\
350 return_value.serialize.return_value = {
351 "registrationId": registration_id1,
354 registration_res = ocloud_view.registration_one(
355 registration_id1, uow)
356 assert str(registration_res.get(
357 "registrationId")) == registration_id1
360 def test_flask_get_list(mock_flask_uow):
361 session, app = mock_flask_uow
362 session.query.return_value = []
363 apibase = config.get_o2ims_api_base()
365 with app.test_client() as client:
366 # Get list and return empty list
367 ##########################
368 resp = client.get(apibase+"/resourceTypes")
369 assert resp.get_data() == b'[]\n'
371 resp = client.get(apibase+"/resourcePools")
372 assert resp.get_data() == b'[]\n'
374 resource_pool_id1 = str(uuid.uuid4())
375 resp = client.get(apibase+"/resourcePools/" +
376 resource_pool_id1+"/resources")
377 assert resp.get_data() == b'[]\n'
379 resp = client.get(apibase+"/deploymentManagers")
380 assert resp.get_data() == b'[]\n'
382 resp = client.get(apibase+"/subscriptions")
383 assert resp.get_data() == b'[]\n'
385 resp = client.get(apibase+"/registrations")
386 assert resp.get_data() == b'[]\n'
389 def test_flask_get_one(mock_flask_uow):
390 session, app = mock_flask_uow
392 session.return_value.query.return_value.filter_by.return_value.\
393 first.return_value = None
394 apibase = config.get_o2ims_api_base()
396 with app.test_client() as client:
397 # Get one and return 404
398 ###########################
399 resp = client.get(apibase+"/")
400 assert resp.status_code == 404
402 resource_type_id1 = str(uuid.uuid4())
403 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
404 assert resp.status_code == 404
406 resource_pool_id1 = str(uuid.uuid4())
407 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
408 assert resp.status_code == 404
410 resource_id1 = str(uuid.uuid4())
411 resp = client.get(apibase+"/resourcePools/" +
412 resource_pool_id1+"/resources/"+resource_id1)
413 assert resp.status_code == 404
415 deployment_manager_id1 = str(uuid.uuid4())
416 resp = client.get(apibase+"/deploymentManagers/" +
417 deployment_manager_id1)
418 assert resp.status_code == 404
420 subscription_id1 = str(uuid.uuid4())
421 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
422 assert resp.status_code == 404
424 registration_id1 = str(uuid.uuid4())
425 resp = client.get(apibase+"/registrations/"+registration_id1)
426 assert resp.status_code == 404
429 def test_flask_post(mock_flask_uow):
430 session, app = mock_flask_uow
431 apibase = config.get_o2ims_api_base()
433 with app.test_client() as client:
434 session.return_value.execute.return_value = []
436 sub_callback = 'http://subscription/callback/url'
437 resp = client.post(apibase+'/subscriptions', json={
438 'callback': sub_callback,
439 'consumerSubscriptionId': 'consumerSubId1',
442 assert resp.status_code == 201
443 assert 'subscriptionId' in resp.get_json()
445 reg_callback = 'http://registration/callback/url'
446 resp = client.post(apibase+'/registrations', json={
447 'callback': reg_callback,
449 assert resp.status_code == 201
450 assert 'registrationId' in resp.get_json()
453 def test_flask_delete(mock_flask_uow):
454 session, app = mock_flask_uow
455 apibase = config.get_o2ims_api_base()
457 with app.test_client() as client:
458 session.return_value.execute.return_value.first.return_value = {}
460 subscription_id1 = str(uuid.uuid4())
461 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
462 assert resp.status_code == 204
464 registration_id1 = str(uuid.uuid4())
465 resp = client.delete(apibase+"/registrations/"+registration_id1)
466 assert resp.status_code == 204
469 def test_flask_not_allowed(mock_flask_uow):
470 _, app = mock_flask_uow
471 apibase = config.get_o2ims_api_base()
473 with app.test_client() as client:
474 # Testing resource type not support method
475 ##########################
476 uri = apibase + "/resourceTypes"
477 resp = client.post(uri)
478 assert resp.status == '405 METHOD NOT ALLOWED'
479 resp = client.put(uri)
480 assert resp.status == '405 METHOD NOT ALLOWED'
481 resp = client.patch(uri)
482 assert resp.status == '405 METHOD NOT ALLOWED'
483 resp = client.delete(uri)
484 assert resp.status == '405 METHOD NOT ALLOWED'
486 resource_type_id1 = str(uuid.uuid4())
487 uri = apibase + "/resourceTypes/" + resource_type_id1
488 resp = client.post(uri)
489 assert resp.status == '405 METHOD NOT ALLOWED'
490 resp = client.put(uri)
491 assert resp.status == '405 METHOD NOT ALLOWED'
492 resp = client.patch(uri)
493 assert resp.status == '405 METHOD NOT ALLOWED'
494 resp = client.delete(uri)
495 assert resp.status == '405 METHOD NOT ALLOWED'
497 # Testing resource pool not support method
498 ##########################
499 uri = apibase + "/resourcePools"
500 resp = client.post(uri)
501 assert resp.status == '405 METHOD NOT ALLOWED'
502 resp = client.put(uri)
503 assert resp.status == '405 METHOD NOT ALLOWED'
504 resp = client.patch(uri)
505 assert resp.status == '405 METHOD NOT ALLOWED'
506 resp = client.delete(uri)
507 assert resp.status == '405 METHOD NOT ALLOWED'
509 resource_pool_id1 = str(uuid.uuid4())
510 uri = apibase + "/resourcePools/" + resource_pool_id1
511 resp = client.post(uri)
512 assert resp.status == '405 METHOD NOT ALLOWED'
513 resp = client.put(uri)
514 assert resp.status == '405 METHOD NOT ALLOWED'
515 resp = client.patch(uri)
516 assert resp.status == '405 METHOD NOT ALLOWED'
517 resp = client.delete(uri)
518 assert resp.status == '405 METHOD NOT ALLOWED'
520 # Testing resource not support method
521 ##########################
522 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
523 resp = client.post(uri)
524 assert resp.status == '405 METHOD NOT ALLOWED'
525 resp = client.put(uri)
526 assert resp.status == '405 METHOD NOT ALLOWED'
527 resp = client.patch(uri)
528 assert resp.status == '405 METHOD NOT ALLOWED'
529 resp = client.delete(uri)
530 assert resp.status == '405 METHOD NOT ALLOWED'
532 resource_id1 = str(uuid.uuid4())
533 uri = apibase + "/resourcePools/" + \
534 resource_pool_id1 + "/resources/" + resource_id1
535 resp = client.post(uri)
536 assert resp.status == '405 METHOD NOT ALLOWED'
537 resp = client.put(uri)
538 assert resp.status == '405 METHOD NOT ALLOWED'
539 resp = client.patch(uri)
540 assert resp.status == '405 METHOD NOT ALLOWED'
541 resp = client.delete(uri)
542 assert resp.status == '405 METHOD NOT ALLOWED'
544 # Testing deployment managers not support method
545 ##########################
546 uri = apibase + "/deploymentManagers"
547 resp = client.post(uri)
548 assert resp.status == '405 METHOD NOT ALLOWED'
549 resp = client.put(uri)
550 assert resp.status == '405 METHOD NOT ALLOWED'
551 resp = client.patch(uri)
552 assert resp.status == '405 METHOD NOT ALLOWED'
553 resp = client.delete(uri)
554 assert resp.status == '405 METHOD NOT ALLOWED'
556 deployment_manager_id1 = str(uuid.uuid4())
557 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
558 resp = client.post(uri)
559 assert resp.status == '405 METHOD NOT ALLOWED'
560 resp = client.put(uri)
561 assert resp.status == '405 METHOD NOT ALLOWED'
562 resp = client.patch(uri)
563 assert resp.status == '405 METHOD NOT ALLOWED'
564 resp = client.delete(uri)
565 assert resp.status == '405 METHOD NOT ALLOWED'
567 # Testing subscriptions not support method
568 ##########################
569 uri = apibase + "/subscriptions"
570 resp = client.put(uri)
571 assert resp.status == '405 METHOD NOT ALLOWED'
572 resp = client.patch(uri)
573 assert resp.status == '405 METHOD NOT ALLOWED'
574 resp = client.delete(uri)
575 assert resp.status == '405 METHOD NOT ALLOWED'
577 subscription_id1 = str(uuid.uuid4())
578 uri = apibase + "/subscriptions/" + subscription_id1
579 resp = client.post(uri)
580 assert resp.status == '405 METHOD NOT ALLOWED'
581 resp = client.put(uri)
582 assert resp.status == '405 METHOD NOT ALLOWED'
583 resp = client.patch(uri)
584 assert resp.status == '405 METHOD NOT ALLOWED'
586 # Testing registrations not support method
587 ##########################
588 uri = apibase + "/registrations"
589 resp = client.put(uri)
590 assert resp.status == '405 METHOD NOT ALLOWED'
591 resp = client.patch(uri)
592 assert resp.status == '405 METHOD NOT ALLOWED'
593 resp = client.delete(uri)
594 assert resp.status == '405 METHOD NOT ALLOWED'
596 subscription_id1 = str(uuid.uuid4())
597 uri = apibase + "/registrations/" + subscription_id1
598 resp = client.post(uri)
599 assert resp.status == '405 METHOD NOT ALLOWED'
600 resp = client.put(uri)
601 assert resp.status == '405 METHOD NOT ALLOWED'
602 resp = client.patch(uri)
603 assert resp.status == '405 METHOD NOT ALLOWED'