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(
247 resource_id1, uow, resource_pool_id1)
248 assert resource_res is None
250 session.return_value.query.return_value.filter_by.return_value.first.\
251 return_value.serialize.return_value = {
252 "resourceId": resource_id1,
253 "resourcePoolId": resource_pool_id1
255 resource_res = ocloud_view.resource_one(
256 resource_id1, uow, resource_pool_id1)
257 assert str(resource_res.get("resourceId") == resource_id1)
258 assert str(resource_res.get("resourcePoolId") == resource_pool_id1)
261 def test_view_deployment_managers(mock_uow):
262 session, uow = mock_uow
264 deployment_manager_id1 = str(uuid.uuid4())
266 dm1.serialize.return_value = {
267 "deploymentManagerId": deployment_manager_id1,
270 order_by = MagicMock()
271 order_by.count.return_value = 1
272 order_by.limit.return_value.offset.return_value = [dm1]
273 session.return_value.query.return_value.filter.return_value.\
274 order_by.return_value = order_by
276 result = ocloud_view.deployment_managers(uow)
277 assert result['count'] == 1
278 ret_list = result['results']
279 assert str(ret_list[0].get("deploymentManagerId")
280 ) == deployment_manager_id1
283 def test_view_deployment_manager_one(mock_uow):
284 session, uow = mock_uow
286 deployment_manager_id1 = str(uuid.uuid4())
287 session.return_value.query.return_value.filter_by.return_value.first.\
288 return_value.serialize.return_value = None
291 deployment_manager_res = ocloud_view.deployment_manager_one(
292 deployment_manager_id1, uow)
293 assert deployment_manager_res is None
295 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
296 session.return_value.query.return_value.filter_by.return_value.first.\
297 return_value.serialize.return_value = {
298 "deploymentManagerId": deployment_manager_id1,
299 "serviceUri": dms_endpoint,
303 CONF.API.DMS_SUPPORT_PROFILES = 'native_k8sapi,sol018,sol018_helmcli'
304 cluster_endpoint = "https://test_k8s:6443"
305 session.return_value.query.return_value.filter_by.return_value.first.\
306 return_value.serialize.return_value['profile'] = {
307 "cluster_api_endpoint": cluster_endpoint
311 deployment_manager_res = ocloud_view.deployment_manager_one(
312 deployment_manager_id1, uow)
313 assert str(deployment_manager_res.get(
314 "deploymentManagerId")) == deployment_manager_id1
315 assert str(deployment_manager_res.get(
316 'serviceUri')) == cluster_endpoint
317 assert deployment_manager_res.get('profile') is None
320 profileName = ocloud.DeploymentManagerProfileSOL018
321 session.return_value.query.return_value.filter_by.return_value.first.\
322 return_value.serialize.return_value['profile'] = {
323 "cluster_api_endpoint": cluster_endpoint
325 deployment_manager_res = ocloud_view.deployment_manager_one(
326 deployment_manager_id1, uow, profile=profileName)
327 assert str(deployment_manager_res.get(
328 'serviceUri')) == cluster_endpoint
329 assert str(deployment_manager_res.get("extensions").get(
330 "profileName")) == profileName
333 profileName = 'wrong_profile'
334 deployment_manager_res = ocloud_view.deployment_manager_one(
335 deployment_manager_id1, uow, profile=profileName)
336 assert deployment_manager_res == ""
339 def test_view_subscriptions(mock_uow):
340 session, uow = mock_uow
342 subscription_id1 = str(uuid.uuid4())
344 sub1.serialize.return_value = {
345 "subscriptionId": subscription_id1,
348 order_by = MagicMock()
349 order_by.count.return_value = 1
350 order_by.limit.return_value.offset.return_value = [sub1]
351 session.return_value.query.return_value.filter.return_value.\
352 order_by.return_value = order_by
354 result = ocloud_view.subscriptions(uow)
355 assert result['count'] == 1
356 ret_list = result['results']
357 assert str(ret_list[0].get("subscriptionId")) == subscription_id1
360 def test_view_subscription_one(mock_uow):
361 session, uow = mock_uow
363 subscription_id1 = str(uuid.uuid4())
364 session.return_value.query.return_value.filter_by.return_value.first.\
365 return_value.serialize.return_value = None
368 subscription_res = ocloud_view.subscription_one(
369 subscription_id1, uow)
370 assert subscription_res is None
372 session.return_value.query.return_value.filter_by.return_value.first.\
373 return_value.serialize.return_value = {
374 "subscriptionId": subscription_id1,
377 subscription_res = ocloud_view.subscription_one(
378 subscription_id1, uow)
379 assert str(subscription_res.get(
380 "subscriptionId")) == subscription_id1
383 def test_flask_get_list(mock_flask_uow):
384 session, app = mock_flask_uow
385 order_by = MagicMock()
386 order_by.count.return_value = 0
387 order_by.limit.return_value.offset.return_value = []
388 session.return_value.query.return_value.filter.return_value.\
389 order_by.return_value = order_by
390 apibase = config.get_o2ims_api_base() + '/v1'
391 # TODO: workaround for sqlalchemy not mapping with resource object
392 setattr(ocloud.Resource, 'resourcePoolId', '')
394 with app.test_client() as client:
395 # Get list and return empty list
396 ##########################
397 resp = client.get(apibase+"/resourceTypes")
398 assert resp.get_data() == b'[]\n'
400 resp = client.get(apibase+"/resourcePools")
401 assert resp.get_data() == b'[]\n'
403 resource_pool_id1 = str(uuid.uuid4())
404 resp = client.get(apibase+"/resourcePools/" +
405 resource_pool_id1+"/resources")
406 assert resp.get_data() == b'[]\n'
408 resp = client.get(apibase+"/deploymentManagers")
409 assert resp.get_data() == b'[]\n'
411 resp = client.get(apibase+"/subscriptions")
412 assert resp.get_data() == b'[]\n'
415 def test_flask_get_one(mock_flask_uow):
416 session, app = mock_flask_uow
418 session.return_value.query.return_value.filter_by.return_value.\
419 first.return_value = None
420 apibase = config.get_o2ims_api_base() + '/v1'
422 with app.test_client() as client:
423 # Get one and return 404
424 ###########################
425 resp = client.get(apibase+"/")
426 assert resp.status_code == 404
428 resource_type_id1 = str(uuid.uuid4())
429 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
430 assert resp.status_code == 404
432 resource_pool_id1 = str(uuid.uuid4())
433 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
434 assert resp.status_code == 404
436 resource_id1 = str(uuid.uuid4())
437 resp = client.get(apibase+"/resourcePools/" +
438 resource_pool_id1+"/resources/"+resource_id1)
439 assert resp.status_code == 404
441 deployment_manager_id1 = str(uuid.uuid4())
442 resp = client.get(apibase+"/deploymentManagers/" +
443 deployment_manager_id1)
444 assert resp.status_code == 404
446 subscription_id1 = str(uuid.uuid4())
447 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
448 assert resp.status_code == 404
451 def test_flask_post(mock_flask_uow, mappers):
452 session, app = mock_flask_uow
453 apibase = config.get_o2ims_api_base() + '/v1'
455 order_by = MagicMock()
456 order_by.count.return_value = 0
457 order_by.limit.return_value.offset.return_value = []
458 session.return_value.query.return_value.filter.return_value.\
459 order_by.return_value = order_by
461 with app.test_client() as client:
462 session.return_value.execute.return_value = []
464 sub_callback = 'http://subscription/callback/url'
465 resp = client.post(apibase+'/subscriptions', json={
466 'callback': sub_callback,
467 'consumerSubscriptionId': 'consumerSubId1',
468 'filter': '(eq,resourceTypeId,xxx)'
470 assert resp.status_code == 201
471 assert 'subscriptionId' in resp.get_json()
474 def test_flask_delete(mock_flask_uow):
475 session, app = mock_flask_uow
476 apibase = config.get_o2ims_api_base() + '/v1'
478 with app.test_client() as client:
479 session.return_value.execute.return_value.first.return_value = {}
481 subscription_id1 = str(uuid.uuid4())
482 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
483 assert resp.status_code == 200
486 def test_flask_not_allowed(mock_flask_uow):
487 _, app = mock_flask_uow
488 apibase = config.get_o2ims_api_base() + '/v1'
490 with app.test_client() as client:
491 # Testing resource type not support method
492 ##########################
493 uri = apibase + "/resourceTypes"
494 resp = client.post(uri)
495 assert resp.status == '405 METHOD NOT ALLOWED'
496 resp = client.put(uri)
497 assert resp.status == '405 METHOD NOT ALLOWED'
498 resp = client.patch(uri)
499 assert resp.status == '405 METHOD NOT ALLOWED'
500 resp = client.delete(uri)
501 assert resp.status == '405 METHOD NOT ALLOWED'
503 resource_type_id1 = str(uuid.uuid4())
504 uri = apibase + "/resourceTypes/" + resource_type_id1
505 resp = client.post(uri)
506 assert resp.status == '405 METHOD NOT ALLOWED'
507 resp = client.put(uri)
508 assert resp.status == '405 METHOD NOT ALLOWED'
509 resp = client.patch(uri)
510 assert resp.status == '405 METHOD NOT ALLOWED'
511 resp = client.delete(uri)
512 assert resp.status == '405 METHOD NOT ALLOWED'
514 # Testing resource pool not support method
515 ##########################
516 uri = apibase + "/resourcePools"
517 resp = client.post(uri)
518 assert resp.status == '405 METHOD NOT ALLOWED'
519 resp = client.put(uri)
520 assert resp.status == '405 METHOD NOT ALLOWED'
521 resp = client.patch(uri)
522 assert resp.status == '405 METHOD NOT ALLOWED'
523 resp = client.delete(uri)
524 assert resp.status == '405 METHOD NOT ALLOWED'
526 resource_pool_id1 = str(uuid.uuid4())
527 uri = apibase + "/resourcePools/" + resource_pool_id1
528 resp = client.post(uri)
529 assert resp.status == '405 METHOD NOT ALLOWED'
530 resp = client.put(uri)
531 assert resp.status == '405 METHOD NOT ALLOWED'
532 resp = client.patch(uri)
533 assert resp.status == '405 METHOD NOT ALLOWED'
534 resp = client.delete(uri)
535 assert resp.status == '405 METHOD NOT ALLOWED'
537 # Testing resource not support method
538 ##########################
539 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
540 resp = client.post(uri)
541 assert resp.status == '405 METHOD NOT ALLOWED'
542 resp = client.put(uri)
543 assert resp.status == '405 METHOD NOT ALLOWED'
544 resp = client.patch(uri)
545 assert resp.status == '405 METHOD NOT ALLOWED'
546 resp = client.delete(uri)
547 assert resp.status == '405 METHOD NOT ALLOWED'
549 resource_id1 = str(uuid.uuid4())
550 uri = apibase + "/resourcePools/" + \
551 resource_pool_id1 + "/resources/" + resource_id1
552 resp = client.post(uri)
553 assert resp.status == '405 METHOD NOT ALLOWED'
554 resp = client.put(uri)
555 assert resp.status == '405 METHOD NOT ALLOWED'
556 resp = client.patch(uri)
557 assert resp.status == '405 METHOD NOT ALLOWED'
558 resp = client.delete(uri)
559 assert resp.status == '405 METHOD NOT ALLOWED'
561 # Testing deployment managers not support method
562 ##########################
563 uri = apibase + "/deploymentManagers"
564 resp = client.post(uri)
565 assert resp.status == '405 METHOD NOT ALLOWED'
566 resp = client.put(uri)
567 assert resp.status == '405 METHOD NOT ALLOWED'
568 resp = client.patch(uri)
569 assert resp.status == '405 METHOD NOT ALLOWED'
570 resp = client.delete(uri)
571 assert resp.status == '405 METHOD NOT ALLOWED'
573 deployment_manager_id1 = str(uuid.uuid4())
574 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
575 resp = client.post(uri)
576 assert resp.status == '405 METHOD NOT ALLOWED'
577 resp = client.put(uri)
578 assert resp.status == '405 METHOD NOT ALLOWED'
579 resp = client.patch(uri)
580 assert resp.status == '405 METHOD NOT ALLOWED'
581 resp = client.delete(uri)
582 assert resp.status == '405 METHOD NOT ALLOWED'
584 # Testing subscriptions not support method
585 ##########################
586 uri = apibase + "/subscriptions"
587 resp = client.put(uri)
588 assert resp.status == '405 METHOD NOT ALLOWED'
589 resp = client.patch(uri)
590 assert resp.status == '405 METHOD NOT ALLOWED'
591 resp = client.delete(uri)
592 assert resp.status == '405 METHOD NOT ALLOWED'
594 subscription_id1 = str(uuid.uuid4())
595 uri = apibase + "/subscriptions/" + subscription_id1
596 resp = client.post(uri)
597 assert resp.status == '405 METHOD NOT ALLOWED'
598 resp = client.put(uri)
599 assert resp.status == '405 METHOD NOT ALLOWED'
600 resp = client.patch(uri)
601 assert resp.status == '405 METHOD NOT ALLOWED'