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, mappers):
450 session, app = mock_flask_uow
451 apibase = config.get_o2ims_api_base() + '/v1'
453 order_by = MagicMock()
454 order_by.count.return_value = 0
455 order_by.limit.return_value.offset.return_value = []
456 session.return_value.query.return_value.filter.return_value.\
457 order_by.return_value = order_by
459 with app.test_client() as client:
460 session.return_value.execute.return_value = []
462 sub_callback = 'http://subscription/callback/url'
463 resp = client.post(apibase+'/subscriptions', json={
464 'callback': sub_callback,
465 'consumerSubscriptionId': 'consumerSubId1',
466 'filter': '(eq,resourceTypeId,xxx)'
468 assert resp.status_code == 201
469 assert 'subscriptionId' in resp.get_json()
472 def test_flask_delete(mock_flask_uow):
473 session, app = mock_flask_uow
474 apibase = config.get_o2ims_api_base() + '/v1'
476 with app.test_client() as client:
477 session.return_value.execute.return_value.first.return_value = {}
479 subscription_id1 = str(uuid.uuid4())
480 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
481 assert resp.status_code == 200
484 def test_flask_not_allowed(mock_flask_uow):
485 _, app = mock_flask_uow
486 apibase = config.get_o2ims_api_base() + '/v1'
488 with app.test_client() as client:
489 # Testing resource type not support method
490 ##########################
491 uri = apibase + "/resourceTypes"
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 resource_type_id1 = str(uuid.uuid4())
502 uri = apibase + "/resourceTypes/" + resource_type_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 resource pool not support method
513 ##########################
514 uri = apibase + "/resourcePools"
515 resp = client.post(uri)
516 assert resp.status == '405 METHOD NOT ALLOWED'
517 resp = client.put(uri)
518 assert resp.status == '405 METHOD NOT ALLOWED'
519 resp = client.patch(uri)
520 assert resp.status == '405 METHOD NOT ALLOWED'
521 resp = client.delete(uri)
522 assert resp.status == '405 METHOD NOT ALLOWED'
524 resource_pool_id1 = str(uuid.uuid4())
525 uri = apibase + "/resourcePools/" + resource_pool_id1
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 # Testing resource not support method
536 ##########################
537 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
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 resource_id1 = str(uuid.uuid4())
548 uri = apibase + "/resourcePools/" + \
549 resource_pool_id1 + "/resources/" + resource_id1
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 # Testing deployment managers not support method
560 ##########################
561 uri = apibase + "/deploymentManagers"
562 resp = client.post(uri)
563 assert resp.status == '405 METHOD NOT ALLOWED'
564 resp = client.put(uri)
565 assert resp.status == '405 METHOD NOT ALLOWED'
566 resp = client.patch(uri)
567 assert resp.status == '405 METHOD NOT ALLOWED'
568 resp = client.delete(uri)
569 assert resp.status == '405 METHOD NOT ALLOWED'
571 deployment_manager_id1 = str(uuid.uuid4())
572 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
573 resp = client.post(uri)
574 assert resp.status == '405 METHOD NOT ALLOWED'
575 resp = client.put(uri)
576 assert resp.status == '405 METHOD NOT ALLOWED'
577 resp = client.patch(uri)
578 assert resp.status == '405 METHOD NOT ALLOWED'
579 resp = client.delete(uri)
580 assert resp.status == '405 METHOD NOT ALLOWED'
582 # Testing subscriptions not support method
583 ##########################
584 uri = apibase + "/subscriptions"
585 resp = client.put(uri)
586 assert resp.status == '405 METHOD NOT ALLOWED'
587 resp = client.patch(uri)
588 assert resp.status == '405 METHOD NOT ALLOWED'
589 resp = client.delete(uri)
590 assert resp.status == '405 METHOD NOT ALLOWED'
592 subscription_id1 = str(uuid.uuid4())
593 uri = apibase + "/subscriptions/" + subscription_id1
594 resp = client.post(uri)
595 assert resp.status == '405 METHOD NOT ALLOWED'
596 resp = client.put(uri)
597 assert resp.status == '405 METHOD NOT ALLOWED'
598 resp = client.patch(uri)
599 assert resp.status == '405 METHOD NOT ALLOWED'