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_view_olcouds(mock_uow):
98 session, uow = mock_uow
100 ocloud1_UUID = str(uuid.uuid4)
101 ocloud1 = MagicMock()
102 ocloud1.serialize.return_value = {
103 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
104 session.return_value.query.return_value = [ocloud1]
106 ocloud_list = ocloud_view.oclouds(uow)
107 # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
108 assert len(ocloud_list) == 1
111 def test_view_olcoud_one(mock_uow):
112 session, uow = mock_uow
114 ocloud1_UUID = str(uuid.uuid4)
115 session.return_value.query.return_value.filter_by.return_value.first.\
116 return_value.serialize.return_value = None
119 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
120 assert ocloud_res is None
122 session.return_value.query.return_value.filter_by.return_value.first.\
123 return_value.serialize.return_value = {
124 "oCloudId": ocloud1_UUID}
126 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
127 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
130 def test_view_resource_types(mock_uow):
131 session, uow = mock_uow
133 resource_type_id1 = str(uuid.uuid4())
134 restype1 = MagicMock()
135 restype1.serialize.return_value = {
136 "resourceTypeId": resource_type_id1}
138 order_by = MagicMock()
139 order_by.count.return_value = 1
140 order_by.limit.return_value.offset.return_value = [restype1]
141 session.return_value.query.return_value.filter_by.return_value.\
142 order_by.return_value = order_by
144 result = ocloud_view.resource_types(uow)
145 assert result['count'] == 1
146 ret_list = result['results']
147 assert str(ret_list[0].get("resourceTypeId")) == resource_type_id1
150 def test_view_resource_type_one(mock_uow):
151 session, uow = mock_uow
153 resource_type_id1 = str(uuid.uuid4())
154 session.return_value.query.return_value.filter_by.return_value.first.\
155 return_value.serialize.return_value = None
158 resource_type_res = ocloud_view.resource_type_one(
159 resource_type_id1, uow)
160 assert resource_type_res is None
162 session.return_value.query.return_value.filter_by.return_value.first.\
163 return_value.serialize.return_value = {
164 "resourceTypeId": resource_type_id1}
166 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
167 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
170 def test_view_resource_pools(mock_uow):
171 session, uow = mock_uow
173 resource_pool_id1 = str(uuid.uuid4())
174 respool1 = MagicMock()
175 respool1.serialize.return_value = {
176 "resourcePoolId": resource_pool_id1}
178 order_by = MagicMock()
179 order_by.count.return_value = 1
180 order_by.limit.return_value.offset.return_value = [respool1]
181 session.return_value.query.return_value.filter_by.return_value.\
182 order_by.return_value = order_by
184 result = ocloud_view.resource_pools(uow)
185 assert result['count'] == 1
186 ret_list = result['results']
187 assert str(ret_list[0].get("resourcePoolId")) == resource_pool_id1
190 def test_view_resource_pool_one(mock_uow):
191 session, uow = mock_uow
193 resource_pool_id1 = str(uuid.uuid4())
194 session.return_value.query.return_value.filter_by.return_value.first.\
195 return_value.serialize.return_value = None
198 resource_pool_res = ocloud_view.resource_pool_one(
199 resource_pool_id1, uow)
200 assert resource_pool_res is None
202 session.return_value.query.return_value.filter_by.return_value.first.\
203 return_value.serialize.return_value = {
204 "resourcePoolId": resource_pool_id1
207 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
208 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
211 def test_view_resources(mock_uow):
212 session, uow = mock_uow
214 resource_id1 = str(uuid.uuid4())
215 resource_pool_id1 = str(uuid.uuid4())
217 res1.serialize.return_value = {
218 "resourceId": resource_id1,
219 "resourcePoolId": resource_pool_id1
222 order_by = MagicMock()
223 order_by.count.return_value = 1
224 order_by.limit.return_value.offset.return_value = [res1]
225 session.return_value.query.return_value.filter_by.return_value.\
226 order_by.return_value = order_by
228 result = ocloud_view.resources(resource_pool_id1, uow)
229 assert result['count'] == 1
230 resource_list = result['results']
231 assert str(resource_list[0].get("resourceId")) == resource_id1
232 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
235 def test_view_resource_one(mock_uow):
236 session, uow = mock_uow
238 resource_id1 = str(uuid.uuid4())
239 resource_pool_id1 = str(uuid.uuid4())
240 session.return_value.query.return_value.filter_by.return_value.first.\
241 return_value.serialize.return_value = None
244 resource_res = ocloud_view.resource_one(resource_id1, uow)
245 assert resource_res is None
247 session.return_value.query.return_value.filter_by.return_value.first.\
248 return_value.serialize.return_value = {
249 "resourceId": resource_id1,
250 "resourcePoolId": resource_pool_id1
253 resource_res = ocloud_view.resource_one(resource_id1, uow)
254 assert str(resource_res.get("resourceId")) == resource_id1
257 def test_view_deployment_managers(mock_uow):
258 session, uow = mock_uow
260 deployment_manager_id1 = str(uuid.uuid4())
262 dm1.serialize.return_value = {
263 "deploymentManagerId": deployment_manager_id1,
266 order_by = MagicMock()
267 order_by.count.return_value = 1
268 order_by.limit.return_value.offset.return_value = [dm1]
269 session.return_value.query.return_value.filter_by.return_value.\
270 order_by.return_value = order_by
272 result = ocloud_view.deployment_managers(uow)
273 assert result['count'] == 1
274 ret_list = result['results']
275 assert str(ret_list[0].get("deploymentManagerId")
276 ) == deployment_manager_id1
279 def test_view_deployment_manager_one(mock_uow):
280 session, uow = mock_uow
282 deployment_manager_id1 = str(uuid.uuid4())
283 session.return_value.query.return_value.filter_by.return_value.first.\
284 return_value.serialize.return_value = None
287 deployment_manager_res = ocloud_view.deployment_manager_one(
288 deployment_manager_id1, uow)
289 assert deployment_manager_res is None
291 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
292 session.return_value.query.return_value.filter_by.return_value.first.\
293 return_value.serialize.return_value = {
294 "deploymentManagerId": deployment_manager_id1,
295 "deploymentManagementServiceEndpoint": dms_endpoint,
300 deployment_manager_res = ocloud_view.deployment_manager_one(
301 deployment_manager_id1, uow)
302 assert str(deployment_manager_res.get(
303 "deploymentManagerId")) == deployment_manager_id1
304 assert str(deployment_manager_res.get(
305 'deploymentManagementServiceEndpoint')) == dms_endpoint
306 assert deployment_manager_res.get('profile') is None
309 profileName = ocloud.DeploymentManagerProfileSOL018
310 cluster_endpoint = "https://test_k8s:6443"
311 session.return_value.query.return_value.filter_by.return_value.first.\
312 return_value.serialize.return_value['profile'] = {
313 "cluster_api_endpoint": cluster_endpoint
315 deployment_manager_res = ocloud_view.deployment_manager_one(
316 deployment_manager_id1, uow, profile=profileName)
317 assert str(deployment_manager_res.get(
318 'deploymentManagementServiceEndpoint')) == cluster_endpoint
319 assert str(deployment_manager_res.get(
320 "profileName")) == profileName
323 profileName = 'wrong_profile'
324 session.return_value.query.return_value.filter_by.return_value.first.\
325 return_value.serialize.return_value['profile'] = {
326 "cluster_api_endpoint": cluster_endpoint
328 deployment_manager_res = ocloud_view.deployment_manager_one(
329 deployment_manager_id1, uow, profile=profileName)
330 assert deployment_manager_res is None
333 def test_view_subscriptions(mock_uow):
334 session, uow = mock_uow
336 subscription_id1 = str(uuid.uuid4())
338 sub1.serialize.return_value = {
339 "subscriptionId": subscription_id1,
342 order_by = MagicMock()
343 order_by.count.return_value = 1
344 order_by.limit.return_value.offset.return_value = [sub1]
345 session.return_value.query.return_value.filter_by.return_value.\
346 order_by.return_value = order_by
348 result = ocloud_view.subscriptions(uow)
349 assert result['count'] == 1
350 ret_list = result['results']
351 assert str(ret_list[0].get("subscriptionId")) == subscription_id1
354 def test_view_subscription_one(mock_uow):
355 session, uow = mock_uow
357 subscription_id1 = str(uuid.uuid4())
358 session.return_value.query.return_value.filter_by.return_value.first.\
359 return_value.serialize.return_value = None
362 subscription_res = ocloud_view.subscription_one(
363 subscription_id1, uow)
364 assert subscription_res is None
366 session.return_value.query.return_value.filter_by.return_value.first.\
367 return_value.serialize.return_value = {
368 "subscriptionId": subscription_id1,
371 subscription_res = ocloud_view.subscription_one(
372 subscription_id1, uow)
373 assert str(subscription_res.get(
374 "subscriptionId")) == subscription_id1
377 def test_flask_get_list(mock_flask_uow):
378 session, app = mock_flask_uow
379 order_by = MagicMock()
380 order_by.count.return_value = 0
381 order_by.limit.return_value.offset.return_value = []
382 session.return_value.query.return_value.filter_by.return_value.\
383 order_by.return_value = order_by
384 apibase = config.get_o2ims_api_base()
386 with app.test_client() as client:
387 # Get list and return empty list
388 ##########################
389 resp = client.get(apibase+"/resourceTypes")
390 assert resp.get_data() == b'[]\n'
392 resp = client.get(apibase+"/resourcePools")
393 assert resp.get_data() == b'[]\n'
395 resource_pool_id1 = str(uuid.uuid4())
396 resp = client.get(apibase+"/resourcePools/" +
397 resource_pool_id1+"/resources")
398 assert resp.get_data() == b'[]\n'
400 resp = client.get(apibase+"/deploymentManagers")
401 assert resp.get_data() == b'[]\n'
403 resp = client.get(apibase+"/subscriptions")
404 assert resp.get_data() == b'[]\n'
407 def test_flask_get_one(mock_flask_uow):
408 session, app = mock_flask_uow
410 session.return_value.query.return_value.filter_by.return_value.\
411 first.return_value = None
412 apibase = config.get_o2ims_api_base()
414 with app.test_client() as client:
415 # Get one and return 404
416 ###########################
417 resp = client.get(apibase+"/")
418 assert resp.status_code == 404
420 resource_type_id1 = str(uuid.uuid4())
421 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
422 assert resp.status_code == 404
424 resource_pool_id1 = str(uuid.uuid4())
425 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
426 assert resp.status_code == 404
428 resource_id1 = str(uuid.uuid4())
429 resp = client.get(apibase+"/resourcePools/" +
430 resource_pool_id1+"/resources/"+resource_id1)
431 assert resp.status_code == 404
433 deployment_manager_id1 = str(uuid.uuid4())
434 resp = client.get(apibase+"/deploymentManagers/" +
435 deployment_manager_id1)
436 assert resp.status_code == 404
438 subscription_id1 = str(uuid.uuid4())
439 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
440 assert resp.status_code == 404
443 def test_flask_post(mock_flask_uow):
444 session, app = mock_flask_uow
445 apibase = config.get_o2ims_api_base()
447 with app.test_client() as client:
448 session.return_value.execute.return_value = []
450 sub_callback = 'http://subscription/callback/url'
451 resp = client.post(apibase+'/subscriptions', json={
452 'callback': sub_callback,
453 'consumerSubscriptionId': 'consumerSubId1',
456 assert resp.status_code == 201
457 assert 'subscriptionId' in resp.get_json()
460 def test_flask_delete(mock_flask_uow):
461 session, app = mock_flask_uow
462 apibase = config.get_o2ims_api_base()
464 with app.test_client() as client:
465 session.return_value.execute.return_value.first.return_value = {}
467 subscription_id1 = str(uuid.uuid4())
468 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
469 assert resp.status_code == 204
472 def test_flask_not_allowed(mock_flask_uow):
473 _, app = mock_flask_uow
474 apibase = config.get_o2ims_api_base()
476 with app.test_client() as client:
477 # Testing resource type not support method
478 ##########################
479 uri = apibase + "/resourceTypes"
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 resource_type_id1 = str(uuid.uuid4())
490 uri = apibase + "/resourceTypes/" + resource_type_id1
491 resp = client.post(uri)
492 assert resp.status == '405 METHOD NOT ALLOWED'
493 resp = client.put(uri)
494 assert resp.status == '405 METHOD NOT ALLOWED'
495 resp = client.patch(uri)
496 assert resp.status == '405 METHOD NOT ALLOWED'
497 resp = client.delete(uri)
498 assert resp.status == '405 METHOD NOT ALLOWED'
500 # Testing resource pool not support method
501 ##########################
502 uri = apibase + "/resourcePools"
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 resource_pool_id1 = str(uuid.uuid4())
513 uri = apibase + "/resourcePools/" + resource_pool_id1
514 resp = client.post(uri)
515 assert resp.status == '405 METHOD NOT ALLOWED'
516 resp = client.put(uri)
517 assert resp.status == '405 METHOD NOT ALLOWED'
518 resp = client.patch(uri)
519 assert resp.status == '405 METHOD NOT ALLOWED'
520 resp = client.delete(uri)
521 assert resp.status == '405 METHOD NOT ALLOWED'
523 # Testing resource not support method
524 ##########################
525 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
526 resp = client.post(uri)
527 assert resp.status == '405 METHOD NOT ALLOWED'
528 resp = client.put(uri)
529 assert resp.status == '405 METHOD NOT ALLOWED'
530 resp = client.patch(uri)
531 assert resp.status == '405 METHOD NOT ALLOWED'
532 resp = client.delete(uri)
533 assert resp.status == '405 METHOD NOT ALLOWED'
535 resource_id1 = str(uuid.uuid4())
536 uri = apibase + "/resourcePools/" + \
537 resource_pool_id1 + "/resources/" + resource_id1
538 resp = client.post(uri)
539 assert resp.status == '405 METHOD NOT ALLOWED'
540 resp = client.put(uri)
541 assert resp.status == '405 METHOD NOT ALLOWED'
542 resp = client.patch(uri)
543 assert resp.status == '405 METHOD NOT ALLOWED'
544 resp = client.delete(uri)
545 assert resp.status == '405 METHOD NOT ALLOWED'
547 # Testing deployment managers not support method
548 ##########################
549 uri = apibase + "/deploymentManagers"
550 resp = client.post(uri)
551 assert resp.status == '405 METHOD NOT ALLOWED'
552 resp = client.put(uri)
553 assert resp.status == '405 METHOD NOT ALLOWED'
554 resp = client.patch(uri)
555 assert resp.status == '405 METHOD NOT ALLOWED'
556 resp = client.delete(uri)
557 assert resp.status == '405 METHOD NOT ALLOWED'
559 deployment_manager_id1 = str(uuid.uuid4())
560 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
561 resp = client.post(uri)
562 assert resp.status == '405 METHOD NOT ALLOWED'
563 resp = client.put(uri)
564 assert resp.status == '405 METHOD NOT ALLOWED'
565 resp = client.patch(uri)
566 assert resp.status == '405 METHOD NOT ALLOWED'
567 resp = client.delete(uri)
568 assert resp.status == '405 METHOD NOT ALLOWED'
570 # Testing subscriptions not support method
571 ##########################
572 uri = apibase + "/subscriptions"
573 resp = client.put(uri)
574 assert resp.status == '405 METHOD NOT ALLOWED'
575 resp = client.patch(uri)
576 assert resp.status == '405 METHOD NOT ALLOWED'
577 resp = client.delete(uri)
578 assert resp.status == '405 METHOD NOT ALLOWED'
580 subscription_id1 = str(uuid.uuid4())
581 uri = apibase + "/subscriptions/" + subscription_id1
582 resp = client.post(uri)
583 assert resp.status == '405 METHOD NOT ALLOWED'
584 resp = client.put(uri)
585 assert resp.status == '405 METHOD NOT ALLOWED'
586 resp = client.patch(uri)
587 assert resp.status == '405 METHOD NOT ALLOWED'