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.filter.return_value.\
105 order_by.return_value = [ocloud1]
107 ocloud_list = ocloud_view.oclouds(uow)
108 # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
109 assert len(ocloud_list) == 1
112 def test_view_olcoud_one(mock_uow):
113 session, uow = mock_uow
115 ocloud1_UUID = str(uuid.uuid4)
116 session.return_value.query.return_value.filter_by.return_value.first.\
117 return_value.serialize.return_value = None
120 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
121 assert ocloud_res is None
123 session.return_value.query.return_value.filter_by.return_value.first.\
124 return_value.serialize.return_value = {
125 "oCloudId": ocloud1_UUID}
127 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
128 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
131 def test_view_resource_types(mock_uow):
132 session, uow = mock_uow
134 resource_type_id1 = str(uuid.uuid4())
135 restype1 = MagicMock()
136 restype1.serialize.return_value = {
137 "resourceTypeId": resource_type_id1}
139 order_by = MagicMock()
140 order_by.count.return_value = 1
141 order_by.limit.return_value.offset.return_value = [restype1]
142 session.return_value.query.return_value.filter.return_value.\
143 order_by.return_value = order_by
145 result = ocloud_view.resource_types(uow)
146 assert result['count'] == 1
147 ret_list = result['results']
148 assert str(ret_list[0].get("resourceTypeId")) == resource_type_id1
151 def test_view_resource_type_one(mock_uow):
152 session, uow = mock_uow
154 resource_type_id1 = str(uuid.uuid4())
155 session.return_value.query.return_value.filter_by.return_value.first.\
156 return_value.serialize.return_value = None
159 resource_type_res = ocloud_view.resource_type_one(
160 resource_type_id1, uow)
161 assert resource_type_res is None
163 session.return_value.query.return_value.filter_by.return_value.first.\
164 return_value.serialize.return_value = {
165 "resourceTypeId": resource_type_id1}
167 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
168 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
171 def test_view_resource_pools(mock_uow):
172 session, uow = mock_uow
174 resource_pool_id1 = str(uuid.uuid4())
175 respool1 = MagicMock()
176 respool1.serialize.return_value = {
177 "resourcePoolId": resource_pool_id1}
179 order_by = MagicMock()
180 order_by.count.return_value = 1
181 order_by.limit.return_value.offset.return_value = [respool1]
182 session.return_value.query.return_value.filter.return_value.\
183 order_by.return_value = order_by
185 result = ocloud_view.resource_pools(uow)
186 assert result['count'] == 1
187 ret_list = result['results']
188 assert str(ret_list[0].get("resourcePoolId")) == resource_pool_id1
191 def test_view_resource_pool_one(mock_uow):
192 session, uow = mock_uow
194 resource_pool_id1 = str(uuid.uuid4())
195 session.return_value.query.return_value.filter_by.return_value.first.\
196 return_value.serialize.return_value = None
199 resource_pool_res = ocloud_view.resource_pool_one(
200 resource_pool_id1, uow)
201 assert resource_pool_res is None
203 session.return_value.query.return_value.filter_by.return_value.first.\
204 return_value.serialize.return_value = {
205 "resourcePoolId": resource_pool_id1
208 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
209 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
212 def test_view_resources(mock_uow):
213 session, uow = mock_uow
215 resource_id1 = str(uuid.uuid4())
216 resource_pool_id1 = str(uuid.uuid4())
218 res1.serialize.return_value = {
219 "resourceId": resource_id1,
220 "resourcePoolId": resource_pool_id1
223 order_by = MagicMock()
224 order_by.count.return_value = 1
225 order_by.limit.return_value.offset.return_value = [res1]
226 session.return_value.query.return_value.filter.return_value.\
227 order_by.return_value = order_by
228 # TODO: workaround for sqlalchemy not mapping with resource object
229 setattr(ocloud.Resource, 'resourcePoolId', '')
231 result = ocloud_view.resources(resource_pool_id1, uow)
232 assert result['count'] == 1
233 resource_list = result['results']
234 assert str(resource_list[0].get("resourceId")) == resource_id1
235 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
238 def test_view_resource_one(mock_uow):
239 session, uow = mock_uow
241 resource_id1 = str(uuid.uuid4())
242 resource_pool_id1 = str(uuid.uuid4())
243 session.return_value.query.return_value.filter_by.return_value.first.\
244 return_value.serialize.return_value = None
247 resource_res = ocloud_view.resource_one(
248 resource_id1, uow, resource_pool_id1)
249 assert resource_res is None
251 session.return_value.query.return_value.filter_by.return_value.first.\
252 return_value.serialize.return_value = {
253 "resourceId": resource_id1,
254 "resourcePoolId": resource_pool_id1
256 resource_res = ocloud_view.resource_one(
257 resource_id1, uow, resource_pool_id1)
258 assert str(resource_res.get("resourceId") == resource_id1)
259 assert str(resource_res.get("resourcePoolId") == resource_pool_id1)
262 def test_view_deployment_managers(mock_uow):
263 session, uow = mock_uow
265 deployment_manager_id1 = str(uuid.uuid4())
267 dm1.serialize.return_value = {
268 "deploymentManagerId": deployment_manager_id1,
271 order_by = MagicMock()
272 order_by.count.return_value = 1
273 order_by.limit.return_value.offset.return_value = [dm1]
274 session.return_value.query.return_value.filter.return_value.\
275 order_by.return_value = order_by
277 result = ocloud_view.deployment_managers(uow)
278 assert result['count'] == 1
279 ret_list = result['results']
280 assert str(ret_list[0].get("deploymentManagerId")
281 ) == deployment_manager_id1
284 def test_view_deployment_manager_one(mock_uow):
285 session, uow = mock_uow
287 deployment_manager_id1 = str(uuid.uuid4())
288 session.return_value.query.return_value.filter_by.return_value.first.\
289 return_value.serialize.return_value = None
292 deployment_manager_res = ocloud_view.deployment_manager_one(
293 deployment_manager_id1, uow)
294 assert deployment_manager_res is None
296 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
297 session.return_value.query.return_value.filter_by.return_value.first.\
298 return_value.serialize.return_value = {
299 "deploymentManagerId": deployment_manager_id1,
300 "serviceUri": dms_endpoint,
304 CONF.API.DMS_SUPPORT_PROFILES = 'native_k8sapi,sol018,sol018_helmcli'
305 cluster_endpoint = "https://test_k8s:6443"
306 session.return_value.query.return_value.filter_by.return_value.first.\
307 return_value.serialize.return_value['profile'] = {
308 "cluster_api_endpoint": cluster_endpoint
312 deployment_manager_res = ocloud_view.deployment_manager_one(
313 deployment_manager_id1, uow)
314 assert str(deployment_manager_res.get(
315 "deploymentManagerId")) == deployment_manager_id1
316 assert str(deployment_manager_res.get(
317 'serviceUri')) == cluster_endpoint
318 assert deployment_manager_res.get('profile') is None
321 profileName = ocloud.DeploymentManagerProfileSOL018
322 session.return_value.query.return_value.filter_by.return_value.first.\
323 return_value.serialize.return_value['profile'] = {
324 "cluster_api_endpoint": cluster_endpoint
326 deployment_manager_res = ocloud_view.deployment_manager_one(
327 deployment_manager_id1, uow, profile=profileName)
328 assert str(deployment_manager_res.get(
329 'serviceUri')) == cluster_endpoint
330 assert str(deployment_manager_res.get("extensions").get(
331 "profileName")) == profileName
334 profileName = 'wrong_profile'
335 deployment_manager_res = ocloud_view.deployment_manager_one(
336 deployment_manager_id1, uow, profile=profileName)
337 assert deployment_manager_res == ""
340 def test_view_subscriptions(mock_uow):
341 session, uow = mock_uow
343 subscription_id1 = str(uuid.uuid4())
345 sub1.serialize.return_value = {
346 "subscriptionId": subscription_id1,
349 order_by = MagicMock()
350 order_by.count.return_value = 1
351 order_by.limit.return_value.offset.return_value = [sub1]
352 session.return_value.query.return_value.filter.return_value.\
353 order_by.return_value = order_by
355 result = ocloud_view.subscriptions(uow)
356 assert result['count'] == 1
357 ret_list = result['results']
358 assert str(ret_list[0].get("subscriptionId")) == subscription_id1
361 def test_view_subscription_one(mock_uow):
362 session, uow = mock_uow
364 subscription_id1 = str(uuid.uuid4())
365 session.return_value.query.return_value.filter_by.return_value.first.\
366 return_value.serialize.return_value = None
369 subscription_res = ocloud_view.subscription_one(
370 subscription_id1, uow)
371 assert subscription_res is None
373 session.return_value.query.return_value.filter_by.return_value.first.\
374 return_value.serialize.return_value = {
375 "subscriptionId": subscription_id1,
378 subscription_res = ocloud_view.subscription_one(
379 subscription_id1, uow)
380 assert str(subscription_res.get(
381 "subscriptionId")) == subscription_id1
384 def test_flask_get_list(mock_flask_uow):
385 session, app = mock_flask_uow
386 order_by = MagicMock()
387 order_by.count.return_value = 0
388 order_by.limit.return_value.offset.return_value = []
389 session.return_value.query.return_value.filter.return_value.\
390 order_by.return_value = order_by
391 apibase = config.get_o2ims_api_base() + '/v1'
392 # TODO: workaround for sqlalchemy not mapping with resource object
393 setattr(ocloud.Resource, 'resourcePoolId', '')
395 with app.test_client() as client:
396 # Get list and return empty list
397 ##########################
398 resp = client.get(apibase+"/resourceTypes")
399 assert resp.get_data() == b'[]\n'
401 resp = client.get(apibase+"/resourcePools")
402 assert resp.get_data() == b'[]\n'
404 resource_pool_id1 = str(uuid.uuid4())
405 resp = client.get(apibase+"/resourcePools/" +
406 resource_pool_id1+"/resources")
407 assert resp.get_data() == b'[]\n'
409 resp = client.get(apibase+"/deploymentManagers")
410 assert resp.get_data() == b'[]\n'
412 resp = client.get(apibase+"/subscriptions")
413 assert resp.get_data() == b'[]\n'
416 def test_flask_get_one(mock_flask_uow):
417 session, app = mock_flask_uow
419 session.return_value.query.return_value.filter_by.return_value.\
420 first.return_value = None
421 apibase = config.get_o2ims_api_base() + '/v1'
423 with app.test_client() as client:
424 # Get one and return 404
425 ###########################
426 resp = client.get(apibase+"/")
427 assert resp.status_code == 404
429 resource_type_id1 = str(uuid.uuid4())
430 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
431 assert resp.status_code == 404
433 resource_pool_id1 = str(uuid.uuid4())
434 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
435 assert resp.status_code == 404
437 resource_id1 = str(uuid.uuid4())
438 resp = client.get(apibase+"/resourcePools/" +
439 resource_pool_id1+"/resources/"+resource_id1)
440 assert resp.status_code == 404
442 deployment_manager_id1 = str(uuid.uuid4())
443 resp = client.get(apibase+"/deploymentManagers/" +
444 deployment_manager_id1)
445 assert resp.status_code == 404
447 subscription_id1 = str(uuid.uuid4())
448 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
449 assert resp.status_code == 404
452 def test_flask_post(mock_flask_uow, mappers):
453 session, app = mock_flask_uow
454 apibase = config.get_o2ims_api_base() + '/v1'
456 order_by = MagicMock()
457 order_by.count.return_value = 0
458 order_by.limit.return_value.offset.return_value = []
459 session.return_value.query.return_value.filter.return_value.\
460 order_by.return_value = order_by
462 with app.test_client() as client:
463 session.return_value.execute.return_value = []
465 sub_callback = 'http://subscription/callback/url'
466 resp = client.post(apibase+'/subscriptions', json={
467 'callback': sub_callback,
468 'consumerSubscriptionId': 'consumerSubId1',
469 'filter': '(eq,resourceTypeId,xxx)'
471 assert resp.status_code == 201
472 assert 'subscriptionId' in resp.get_json()
475 def test_flask_delete(mock_flask_uow):
476 session, app = mock_flask_uow
477 apibase = config.get_o2ims_api_base() + '/v1'
479 with app.test_client() as client:
480 session.return_value.execute.return_value.first.return_value = {}
482 subscription_id1 = str(uuid.uuid4())
483 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
484 assert resp.status_code == 200
487 def test_flask_not_allowed(mock_flask_uow):
488 _, app = mock_flask_uow
489 apibase = config.get_o2ims_api_base() + '/v1'
491 with app.test_client() as client:
492 # Testing resource type not support method
493 ##########################
494 uri = apibase + "/resourceTypes"
495 resp = client.post(uri)
496 assert resp.status == '405 METHOD NOT ALLOWED'
497 resp = client.put(uri)
498 assert resp.status == '405 METHOD NOT ALLOWED'
499 resp = client.patch(uri)
500 assert resp.status == '405 METHOD NOT ALLOWED'
501 resp = client.delete(uri)
502 assert resp.status == '405 METHOD NOT ALLOWED'
504 resource_type_id1 = str(uuid.uuid4())
505 uri = apibase + "/resourceTypes/" + resource_type_id1
506 resp = client.post(uri)
507 assert resp.status == '405 METHOD NOT ALLOWED'
508 resp = client.put(uri)
509 assert resp.status == '405 METHOD NOT ALLOWED'
510 resp = client.patch(uri)
511 assert resp.status == '405 METHOD NOT ALLOWED'
512 resp = client.delete(uri)
513 assert resp.status == '405 METHOD NOT ALLOWED'
515 # Testing resource pool not support method
516 ##########################
517 uri = apibase + "/resourcePools"
518 resp = client.post(uri)
519 assert resp.status == '405 METHOD NOT ALLOWED'
520 resp = client.put(uri)
521 assert resp.status == '405 METHOD NOT ALLOWED'
522 resp = client.patch(uri)
523 assert resp.status == '405 METHOD NOT ALLOWED'
524 resp = client.delete(uri)
525 assert resp.status == '405 METHOD NOT ALLOWED'
527 resource_pool_id1 = str(uuid.uuid4())
528 uri = apibase + "/resourcePools/" + resource_pool_id1
529 resp = client.post(uri)
530 assert resp.status == '405 METHOD NOT ALLOWED'
531 resp = client.put(uri)
532 assert resp.status == '405 METHOD NOT ALLOWED'
533 resp = client.patch(uri)
534 assert resp.status == '405 METHOD NOT ALLOWED'
535 resp = client.delete(uri)
536 assert resp.status == '405 METHOD NOT ALLOWED'
538 # Testing resource not support method
539 ##########################
540 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
541 resp = client.post(uri)
542 assert resp.status == '405 METHOD NOT ALLOWED'
543 resp = client.put(uri)
544 assert resp.status == '405 METHOD NOT ALLOWED'
545 resp = client.patch(uri)
546 assert resp.status == '405 METHOD NOT ALLOWED'
547 resp = client.delete(uri)
548 assert resp.status == '405 METHOD NOT ALLOWED'
550 resource_id1 = str(uuid.uuid4())
551 uri = apibase + "/resourcePools/" + \
552 resource_pool_id1 + "/resources/" + resource_id1
553 resp = client.post(uri)
554 assert resp.status == '405 METHOD NOT ALLOWED'
555 resp = client.put(uri)
556 assert resp.status == '405 METHOD NOT ALLOWED'
557 resp = client.patch(uri)
558 assert resp.status == '405 METHOD NOT ALLOWED'
559 resp = client.delete(uri)
560 assert resp.status == '405 METHOD NOT ALLOWED'
562 # Testing deployment managers not support method
563 ##########################
564 uri = apibase + "/deploymentManagers"
565 resp = client.post(uri)
566 assert resp.status == '405 METHOD NOT ALLOWED'
567 resp = client.put(uri)
568 assert resp.status == '405 METHOD NOT ALLOWED'
569 resp = client.patch(uri)
570 assert resp.status == '405 METHOD NOT ALLOWED'
571 resp = client.delete(uri)
572 assert resp.status == '405 METHOD NOT ALLOWED'
574 deployment_manager_id1 = str(uuid.uuid4())
575 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
576 resp = client.post(uri)
577 assert resp.status == '405 METHOD NOT ALLOWED'
578 resp = client.put(uri)
579 assert resp.status == '405 METHOD NOT ALLOWED'
580 resp = client.patch(uri)
581 assert resp.status == '405 METHOD NOT ALLOWED'
582 resp = client.delete(uri)
583 assert resp.status == '405 METHOD NOT ALLOWED'
585 # Testing subscriptions not support method
586 ##########################
587 uri = apibase + "/subscriptions"
588 resp = client.put(uri)
589 assert resp.status == '405 METHOD NOT ALLOWED'
590 resp = client.patch(uri)
591 assert resp.status == '405 METHOD NOT ALLOWED'
592 resp = client.delete(uri)
593 assert resp.status == '405 METHOD NOT ALLOWED'
595 subscription_id1 = str(uuid.uuid4())
596 uri = apibase + "/subscriptions/" + subscription_id1
597 resp = client.post(uri)
598 assert resp.status == '405 METHOD NOT ALLOWED'
599 resp = client.put(uri)
600 assert resp.status == '405 METHOD NOT ALLOWED'
601 resp = client.patch(uri)
602 assert resp.status == '405 METHOD NOT ALLOWED'