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, conf as CONF
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.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.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.return_value.\
226 order_by.return_value = order_by
227 # TODO: workaround for sqlalchemy not mapping with resource object
228 setattr(ocloud.Resource, 'resourcePoolId', '')
230 result = ocloud_view.resources(resource_pool_id1, uow)
231 assert result['count'] == 1
232 resource_list = result['results']
233 assert str(resource_list[0].get("resourceId")) == resource_id1
234 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
237 def test_view_resource_one(mock_uow):
238 session, uow = mock_uow
240 resource_id1 = str(uuid.uuid4())
241 resource_pool_id1 = str(uuid.uuid4())
242 session.return_value.query.return_value.filter_by.return_value.first.\
243 return_value.serialize.return_value = None
246 resource_res = ocloud_view.resource_one(resource_id1, uow)
247 assert resource_res is None
249 session.return_value.query.return_value.filter_by.return_value.first.\
250 return_value.serialize.return_value = {
251 "resourceId": resource_id1,
252 "resourcePoolId": resource_pool_id1
255 resource_res = ocloud_view.resource_one(resource_id1, uow)
256 assert str(resource_res.get("resourceId")) == resource_id1
259 def test_view_deployment_managers(mock_uow):
260 session, uow = mock_uow
262 deployment_manager_id1 = str(uuid.uuid4())
264 dm1.serialize.return_value = {
265 "deploymentManagerId": deployment_manager_id1,
268 order_by = MagicMock()
269 order_by.count.return_value = 1
270 order_by.limit.return_value.offset.return_value = [dm1]
271 session.return_value.query.return_value.filter.return_value.\
272 order_by.return_value = order_by
274 result = ocloud_view.deployment_managers(uow)
275 assert result['count'] == 1
276 ret_list = result['results']
277 assert str(ret_list[0].get("deploymentManagerId")
278 ) == deployment_manager_id1
281 def test_view_deployment_manager_one(mock_uow):
282 session, uow = mock_uow
284 deployment_manager_id1 = str(uuid.uuid4())
285 session.return_value.query.return_value.filter_by.return_value.first.\
286 return_value.serialize.return_value = None
289 deployment_manager_res = ocloud_view.deployment_manager_one(
290 deployment_manager_id1, uow)
291 assert deployment_manager_res is None
293 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
294 session.return_value.query.return_value.filter_by.return_value.first.\
295 return_value.serialize.return_value = {
296 "deploymentManagerId": deployment_manager_id1,
297 "serviceUri": dms_endpoint,
301 CONF.API.DMS_SUPPORT_PROFILES = 'native_k8sapi,sol018,sol018_helmcli'
302 cluster_endpoint = "https://test_k8s:6443"
303 session.return_value.query.return_value.filter_by.return_value.first.\
304 return_value.serialize.return_value['profile'] = {
305 "cluster_api_endpoint": cluster_endpoint
309 deployment_manager_res = ocloud_view.deployment_manager_one(
310 deployment_manager_id1, uow)
311 assert str(deployment_manager_res.get(
312 "deploymentManagerId")) == deployment_manager_id1
313 assert str(deployment_manager_res.get(
314 'serviceUri')) == cluster_endpoint
315 assert deployment_manager_res.get('profile') is None
318 profileName = ocloud.DeploymentManagerProfileSOL018
319 session.return_value.query.return_value.filter_by.return_value.first.\
320 return_value.serialize.return_value['profile'] = {
321 "cluster_api_endpoint": cluster_endpoint
323 deployment_manager_res = ocloud_view.deployment_manager_one(
324 deployment_manager_id1, uow, profile=profileName)
325 assert str(deployment_manager_res.get(
326 'serviceUri')) == cluster_endpoint
327 assert str(deployment_manager_res.get("extensions").get(
328 "profileName")) == profileName
331 profileName = 'wrong_profile'
332 deployment_manager_res = ocloud_view.deployment_manager_one(
333 deployment_manager_id1, uow, profile=profileName)
334 assert deployment_manager_res == ""
337 def test_view_subscriptions(mock_uow):
338 session, uow = mock_uow
340 subscription_id1 = str(uuid.uuid4())
342 sub1.serialize.return_value = {
343 "subscriptionId": subscription_id1,
346 order_by = MagicMock()
347 order_by.count.return_value = 1
348 order_by.limit.return_value.offset.return_value = [sub1]
349 session.return_value.query.return_value.filter.return_value.\
350 order_by.return_value = order_by
352 result = ocloud_view.subscriptions(uow)
353 assert result['count'] == 1
354 ret_list = result['results']
355 assert str(ret_list[0].get("subscriptionId")) == subscription_id1
358 def test_view_subscription_one(mock_uow):
359 session, uow = mock_uow
361 subscription_id1 = str(uuid.uuid4())
362 session.return_value.query.return_value.filter_by.return_value.first.\
363 return_value.serialize.return_value = None
366 subscription_res = ocloud_view.subscription_one(
367 subscription_id1, uow)
368 assert subscription_res is None
370 session.return_value.query.return_value.filter_by.return_value.first.\
371 return_value.serialize.return_value = {
372 "subscriptionId": subscription_id1,
375 subscription_res = ocloud_view.subscription_one(
376 subscription_id1, uow)
377 assert str(subscription_res.get(
378 "subscriptionId")) == subscription_id1
381 def test_flask_get_list(mock_flask_uow):
382 session, app = mock_flask_uow
383 order_by = MagicMock()
384 order_by.count.return_value = 0
385 order_by.limit.return_value.offset.return_value = []
386 session.return_value.query.return_value.filter.return_value.\
387 order_by.return_value = order_by
388 apibase = config.get_o2ims_api_base() + '/v1'
389 # TODO: workaround for sqlalchemy not mapping with resource object
390 setattr(ocloud.Resource, 'resourcePoolId', '')
392 with app.test_client() as client:
393 # Get list and return empty list
394 ##########################
395 resp = client.get(apibase+"/resourceTypes")
396 assert resp.get_data() == b'[]\n'
398 resp = client.get(apibase+"/resourcePools")
399 assert resp.get_data() == b'[]\n'
401 resource_pool_id1 = str(uuid.uuid4())
402 resp = client.get(apibase+"/resourcePools/" +
403 resource_pool_id1+"/resources")
404 assert resp.get_data() == b'[]\n'
406 resp = client.get(apibase+"/deploymentManagers")
407 assert resp.get_data() == b'[]\n'
409 resp = client.get(apibase+"/subscriptions")
410 assert resp.get_data() == b'[]\n'
413 def test_flask_get_one(mock_flask_uow):
414 session, app = mock_flask_uow
416 session.return_value.query.return_value.filter_by.return_value.\
417 first.return_value = None
418 apibase = config.get_o2ims_api_base() + '/v1'
420 with app.test_client() as client:
421 # Get one and return 404
422 ###########################
423 resp = client.get(apibase+"/")
424 assert resp.status_code == 404
426 resource_type_id1 = str(uuid.uuid4())
427 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
428 assert resp.status_code == 404
430 resource_pool_id1 = str(uuid.uuid4())
431 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
432 assert resp.status_code == 404
434 resource_id1 = str(uuid.uuid4())
435 resp = client.get(apibase+"/resourcePools/" +
436 resource_pool_id1+"/resources/"+resource_id1)
437 assert resp.status_code == 404
439 deployment_manager_id1 = str(uuid.uuid4())
440 resp = client.get(apibase+"/deploymentManagers/" +
441 deployment_manager_id1)
442 assert resp.status_code == 404
444 subscription_id1 = str(uuid.uuid4())
445 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
446 assert resp.status_code == 404
449 def test_flask_post(mock_flask_uow):
450 session, app = mock_flask_uow
451 apibase = config.get_o2ims_api_base() + '/v1'
453 with app.test_client() as client:
454 session.return_value.execute.return_value = []
456 sub_callback = 'http://subscription/callback/url'
457 resp = client.post(apibase+'/subscriptions', json={
458 'callback': sub_callback,
459 'consumerSubscriptionId': 'consumerSubId1',
462 assert resp.status_code == 201
463 assert 'subscriptionId' in resp.get_json()
466 def test_flask_delete(mock_flask_uow):
467 session, app = mock_flask_uow
468 apibase = config.get_o2ims_api_base() + '/v1'
470 with app.test_client() as client:
471 session.return_value.execute.return_value.first.return_value = {}
473 subscription_id1 = str(uuid.uuid4())
474 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
475 assert resp.status_code == 200
478 def test_flask_not_allowed(mock_flask_uow):
479 _, app = mock_flask_uow
480 apibase = config.get_o2ims_api_base() + '/v1'
482 with app.test_client() as client:
483 # Testing resource type not support method
484 ##########################
485 uri = apibase + "/resourceTypes"
486 resp = client.post(uri)
487 assert resp.status == '405 METHOD NOT ALLOWED'
488 resp = client.put(uri)
489 assert resp.status == '405 METHOD NOT ALLOWED'
490 resp = client.patch(uri)
491 assert resp.status == '405 METHOD NOT ALLOWED'
492 resp = client.delete(uri)
493 assert resp.status == '405 METHOD NOT ALLOWED'
495 resource_type_id1 = str(uuid.uuid4())
496 uri = apibase + "/resourceTypes/" + resource_type_id1
497 resp = client.post(uri)
498 assert resp.status == '405 METHOD NOT ALLOWED'
499 resp = client.put(uri)
500 assert resp.status == '405 METHOD NOT ALLOWED'
501 resp = client.patch(uri)
502 assert resp.status == '405 METHOD NOT ALLOWED'
503 resp = client.delete(uri)
504 assert resp.status == '405 METHOD NOT ALLOWED'
506 # Testing resource pool not support method
507 ##########################
508 uri = apibase + "/resourcePools"
509 resp = client.post(uri)
510 assert resp.status == '405 METHOD NOT ALLOWED'
511 resp = client.put(uri)
512 assert resp.status == '405 METHOD NOT ALLOWED'
513 resp = client.patch(uri)
514 assert resp.status == '405 METHOD NOT ALLOWED'
515 resp = client.delete(uri)
516 assert resp.status == '405 METHOD NOT ALLOWED'
518 resource_pool_id1 = str(uuid.uuid4())
519 uri = apibase + "/resourcePools/" + resource_pool_id1
520 resp = client.post(uri)
521 assert resp.status == '405 METHOD NOT ALLOWED'
522 resp = client.put(uri)
523 assert resp.status == '405 METHOD NOT ALLOWED'
524 resp = client.patch(uri)
525 assert resp.status == '405 METHOD NOT ALLOWED'
526 resp = client.delete(uri)
527 assert resp.status == '405 METHOD NOT ALLOWED'
529 # Testing resource not support method
530 ##########################
531 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
532 resp = client.post(uri)
533 assert resp.status == '405 METHOD NOT ALLOWED'
534 resp = client.put(uri)
535 assert resp.status == '405 METHOD NOT ALLOWED'
536 resp = client.patch(uri)
537 assert resp.status == '405 METHOD NOT ALLOWED'
538 resp = client.delete(uri)
539 assert resp.status == '405 METHOD NOT ALLOWED'
541 resource_id1 = str(uuid.uuid4())
542 uri = apibase + "/resourcePools/" + \
543 resource_pool_id1 + "/resources/" + resource_id1
544 resp = client.post(uri)
545 assert resp.status == '405 METHOD NOT ALLOWED'
546 resp = client.put(uri)
547 assert resp.status == '405 METHOD NOT ALLOWED'
548 resp = client.patch(uri)
549 assert resp.status == '405 METHOD NOT ALLOWED'
550 resp = client.delete(uri)
551 assert resp.status == '405 METHOD NOT ALLOWED'
553 # Testing deployment managers not support method
554 ##########################
555 uri = apibase + "/deploymentManagers"
556 resp = client.post(uri)
557 assert resp.status == '405 METHOD NOT ALLOWED'
558 resp = client.put(uri)
559 assert resp.status == '405 METHOD NOT ALLOWED'
560 resp = client.patch(uri)
561 assert resp.status == '405 METHOD NOT ALLOWED'
562 resp = client.delete(uri)
563 assert resp.status == '405 METHOD NOT ALLOWED'
565 deployment_manager_id1 = str(uuid.uuid4())
566 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
567 resp = client.post(uri)
568 assert resp.status == '405 METHOD NOT ALLOWED'
569 resp = client.put(uri)
570 assert resp.status == '405 METHOD NOT ALLOWED'
571 resp = client.patch(uri)
572 assert resp.status == '405 METHOD NOT ALLOWED'
573 resp = client.delete(uri)
574 assert resp.status == '405 METHOD NOT ALLOWED'
576 # Testing subscriptions not support method
577 ##########################
578 uri = apibase + "/subscriptions"
579 resp = client.put(uri)
580 assert resp.status == '405 METHOD NOT ALLOWED'
581 resp = client.patch(uri)
582 assert resp.status == '405 METHOD NOT ALLOWED'
583 resp = client.delete(uri)
584 assert resp.status == '405 METHOD NOT ALLOWED'
586 subscription_id1 = str(uuid.uuid4())
587 uri = apibase + "/subscriptions/" + subscription_id1
588 resp = client.post(uri)
589 assert resp.status == '405 METHOD NOT ALLOWED'
590 resp = client.put(uri)
591 assert resp.status == '405 METHOD NOT ALLOWED'
592 resp = client.patch(uri)
593 assert resp.status == '405 METHOD NOT ALLOWED'