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
17 # from o2dms.domain import dms
19 from o2ims.domain import ocloud, subscription_obj, configuration_obj
20 from o2ims.domain import resource_type as rt
21 from o2ims.views import ocloud_view
22 from o2common.config import config
26 ocloudid1 = str(uuid.uuid4())
27 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
28 config.get_api_url(), "ocloud for unit test", 1)
32 def test_new_ocloud():
33 ocloudid1 = str(uuid.uuid4())
34 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
35 config.get_api_url(), "ocloud for unit test", 1)
36 assert ocloudid1 is not None and ocloud1.oCloudId == ocloudid1
39 # def test_add_ocloud_with_dms():
40 # ocloud1 = setup_ocloud()
41 # dmsid = str(uuid.uuid4())
42 # dms = ocloud.DeploymentManager(
43 # dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
44 # ocloud1.addDeploymentManager(dms)
45 # ocloud1.addDeploymentManager(dms)
46 # assert len(ocloud1.deploymentManagers) == 1
47 # # repo.update(ocloud1.oCloudId, {
48 # # "deploymentManagers": ocloud1.deploymentManagers})
51 def test_new_resource_type():
52 ocloud1 = setup_ocloud()
53 resource_type_id1 = str(uuid.uuid4())
54 resource_type1 = ocloud.ResourceType(
55 resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
57 assert resource_type_id1 is not None and \
58 resource_type1.resourceTypeId == resource_type_id1
61 def test_new_resource_pool():
62 ocloud1 = setup_ocloud()
63 resource_pool_id1 = str(uuid.uuid4())
64 resource_pool1 = ocloud.ResourcePool(
65 resource_pool_id1, "resourcepool1", config.get_api_url(),
67 assert resource_pool_id1 is not None and \
68 resource_pool1.resourcePoolId == resource_pool_id1
71 def test_new_resource():
72 resource_id1 = str(uuid.uuid4())
73 resource_type_id1 = str(uuid.uuid4())
74 resource_pool_id1 = str(uuid.uuid4())
75 resource1 = ocloud.Resource(
76 resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
77 assert resource_id1 is not None and resource1.resourceId == resource_id1
80 def test_new_deployment_manager():
81 ocloud_id1 = str(uuid.uuid4())
82 deployment_manager_id1 = str(uuid.uuid4())
83 deployment_manager1 = ocloud.DeploymentManager(
84 deployment_manager_id1, "k8s1", ocloud_id1,
85 config.get_api_url()+"/k8s1")
86 assert deployment_manager_id1 is not None and deployment_manager1.\
87 deploymentManagerId == deployment_manager_id1
90 def test_new_subscription():
91 subscription_id1 = str(uuid.uuid4())
92 subscription1 = subscription_obj.Subscription(
93 subscription_id1, "https://callback/uri/write/here")
94 assert subscription_id1 is not None and\
95 subscription1.subscriptionId == subscription_id1
98 def test_new_configuration():
99 configuration_id1 = str(uuid.uuid4())
100 configuration1 = configuration_obj.Configuration(
101 configuration_id1, "https://callback/uri/write/here",
103 assert configuration_id1 is not None and\
104 configuration1.configurationId == configuration_id1
107 def test_view_olcouds(mock_uow):
108 session, uow = mock_uow
110 ocloud1_UUID = str(uuid.uuid4)
111 ocloud1 = MagicMock()
112 ocloud1.serialize.return_value = {
113 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
114 session.return_value.query.return_value = [ocloud1]
116 ocloud_list = ocloud_view.oclouds(uow)
117 # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
118 assert len(ocloud_list) == 1
121 def test_view_olcoud_one(mock_uow):
122 session, uow = mock_uow
124 ocloud1_UUID = str(uuid.uuid4)
125 session.return_value.query.return_value.filter_by.return_value.first.\
126 return_value.serialize.return_value = None
129 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
130 assert ocloud_res is None
132 session.return_value.query.return_value.filter_by.return_value.first.\
133 return_value.serialize.return_value = {
134 "oCloudId": ocloud1_UUID}
136 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
137 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
140 def test_view_resource_types(mock_uow):
141 session, uow = mock_uow
143 resource_type_id1 = str(uuid.uuid4())
144 restype1 = MagicMock()
145 restype1.serialize.return_value = {
146 "resourceTypeId": resource_type_id1}
147 session.return_value.query.return_value = [restype1]
149 resource_type_list = ocloud_view.resource_types(uow)
150 assert str(resource_type_list[0].get(
151 "resourceTypeId")) == resource_type_id1
154 def test_view_resource_type_one(mock_uow):
155 session, uow = mock_uow
157 resource_type_id1 = str(uuid.uuid4())
158 session.return_value.query.return_value.filter_by.return_value.first.\
159 return_value.serialize.return_value = None
162 resource_type_res = ocloud_view.resource_type_one(
163 resource_type_id1, uow)
164 assert resource_type_res is None
166 session.return_value.query.return_value.filter_by.return_value.first.\
167 return_value.serialize.return_value = {
168 "resourceTypeId": resource_type_id1}
170 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
171 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
174 def test_view_resource_pools(mock_uow):
175 session, uow = mock_uow
177 resource_pool_id1 = str(uuid.uuid4())
178 respool1 = MagicMock()
179 respool1.serialize.return_value = {
180 "resourcePoolId": resource_pool_id1}
181 session.return_value.query.return_value = [respool1]
183 resource_pool_list = ocloud_view.resource_pools(uow)
184 assert str(resource_pool_list[0].get(
185 "resourcePoolId")) == resource_pool_id1
188 def test_view_resource_pool_one(mock_uow):
189 session, uow = mock_uow
191 resource_pool_id1 = str(uuid.uuid4())
192 session.return_value.query.return_value.filter_by.return_value.first.\
193 return_value.serialize.return_value = None
196 resource_pool_res = ocloud_view.resource_pool_one(
197 resource_pool_id1, uow)
198 assert resource_pool_res is None
200 session.return_value.query.return_value.filter_by.return_value.first.\
201 return_value.serialize.return_value = {
202 "resourcePoolId": resource_pool_id1
205 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
206 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
209 def test_view_resources(mock_uow):
210 session, uow = mock_uow
212 resource_id1 = str(uuid.uuid4())
213 resource_pool_id1 = str(uuid.uuid4())
215 res1.serialize.return_value = {
216 "resourceId": resource_id1,
217 "resourcePoolId": resource_pool_id1
219 session.return_value.query.return_value.filter_by.return_value = [res1]
221 resource_list = ocloud_view.resources(resource_pool_id1, uow)
222 assert str(resource_list[0].get("resourceId")) == resource_id1
223 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
226 def test_view_resource_one(mock_uow):
227 session, uow = mock_uow
229 resource_id1 = str(uuid.uuid4())
230 resource_pool_id1 = str(uuid.uuid4())
231 session.return_value.query.return_value.filter_by.return_value.first.\
232 return_value.serialize.return_value = None
235 resource_res = ocloud_view.resource_one(resource_id1, uow)
236 assert resource_res is None
238 session.return_value.query.return_value.filter_by.return_value.first.\
239 return_value.serialize.return_value = {
240 "resourceId": resource_id1,
241 "resourcePoolId": resource_pool_id1
244 resource_res = ocloud_view.resource_one(resource_id1, uow)
245 assert str(resource_res.get("resourceId")) == resource_id1
248 def test_view_deployment_managers(mock_uow):
249 session, uow = mock_uow
251 deployment_manager_id1 = str(uuid.uuid4())
253 dm1.serialize.return_value = {
254 "deploymentManagerId": deployment_manager_id1,
256 session.return_value.query.return_value = [dm1]
258 deployment_manager_list = ocloud_view.deployment_managers(uow)
259 assert str(deployment_manager_list[0].get(
260 "deploymentManagerId")) == deployment_manager_id1
263 def test_view_deployment_manager_one(mock_uow):
264 session, uow = mock_uow
266 deployment_manager_id1 = str(uuid.uuid4())
267 session.return_value.query.return_value.filter_by.return_value.first.\
268 return_value.serialize.return_value = None
271 deployment_manager_res = ocloud_view.deployment_manager_one(
272 deployment_manager_id1, uow)
273 assert deployment_manager_res is None
275 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
276 session.return_value.query.return_value.filter_by.return_value.first.\
277 return_value.serialize.return_value = {
278 "deploymentManagerId": deployment_manager_id1,
279 "deploymentManagementServiceEndpoint": dms_endpoint,
284 deployment_manager_res = ocloud_view.deployment_manager_one(
285 deployment_manager_id1, uow)
286 assert str(deployment_manager_res.get(
287 "deploymentManagerId")) == deployment_manager_id1
288 assert str(deployment_manager_res.get(
289 'deploymentManagementServiceEndpoint')) == dms_endpoint
290 assert deployment_manager_res.get('profile') is None
293 profileName = ocloud.DeploymentManagerProfileSOL018
294 cluster_endpoint = "https://test_k8s:6443"
295 session.return_value.query.return_value.filter_by.return_value.first.\
296 return_value.serialize.return_value['profile'] = {
297 "cluster_api_endpoint": cluster_endpoint
299 deployment_manager_res = ocloud_view.deployment_manager_one(
300 deployment_manager_id1, uow, profile=profileName)
301 assert str(deployment_manager_res.get(
302 'deploymentManagementServiceEndpoint')) == cluster_endpoint
303 assert str(deployment_manager_res.get(
304 "profileName")) == profileName
307 profileName = 'wrong_profile'
308 session.return_value.query.return_value.filter_by.return_value.first.\
309 return_value.serialize.return_value['profile'] = {
310 "cluster_api_endpoint": cluster_endpoint
312 deployment_manager_res = ocloud_view.deployment_manager_one(
313 deployment_manager_id1, uow, profile=profileName)
314 assert deployment_manager_res is None
317 def test_view_subscriptions(mock_uow):
318 session, uow = mock_uow
320 subscription_id1 = str(uuid.uuid4())
322 sub1.serialize.return_value = {
323 "subscriptionId": subscription_id1,
325 session.return_value.query.return_value = [sub1]
327 subscription_list = ocloud_view.subscriptions(uow)
328 assert str(subscription_list[0].get(
329 "subscriptionId")) == subscription_id1
332 def test_view_subscription_one(mock_uow):
333 session, uow = mock_uow
335 subscription_id1 = str(uuid.uuid4())
336 session.return_value.query.return_value.filter_by.return_value.first.\
337 return_value.serialize.return_value = None
340 subscription_res = ocloud_view.subscription_one(
341 subscription_id1, uow)
342 assert subscription_res is None
344 session.return_value.query.return_value.filter_by.return_value.first.\
345 return_value.serialize.return_value = {
346 "subscriptionId": subscription_id1,
349 subscription_res = ocloud_view.subscription_one(
350 subscription_id1, uow)
351 assert str(subscription_res.get(
352 "subscriptionId")) == subscription_id1
355 def test_flask_get_list(mock_flask_uow):
356 session, app = mock_flask_uow
357 session.query.return_value = []
358 apibase = config.get_o2ims_api_base()
360 with app.test_client() as client:
361 # Get list and return empty list
362 ##########################
363 resp = client.get(apibase+"/resourceTypes")
364 assert resp.get_data() == b'[]\n'
366 resp = client.get(apibase+"/resourcePools")
367 assert resp.get_data() == b'[]\n'
369 resource_pool_id1 = str(uuid.uuid4())
370 resp = client.get(apibase+"/resourcePools/" +
371 resource_pool_id1+"/resources")
372 assert resp.get_data() == b'[]\n'
374 resp = client.get(apibase+"/deploymentManagers")
375 assert resp.get_data() == b'[]\n'
377 resp = client.get(apibase+"/subscriptions")
378 assert resp.get_data() == b'[]\n'
381 def test_flask_get_one(mock_flask_uow):
382 session, app = mock_flask_uow
384 session.return_value.query.return_value.filter_by.return_value.\
385 first.return_value = None
386 apibase = config.get_o2ims_api_base()
388 with app.test_client() as client:
389 # Get one and return 404
390 ###########################
391 resp = client.get(apibase+"/")
392 assert resp.status_code == 404
394 resource_type_id1 = str(uuid.uuid4())
395 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
396 assert resp.status_code == 404
398 resource_pool_id1 = str(uuid.uuid4())
399 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
400 assert resp.status_code == 404
402 resource_id1 = str(uuid.uuid4())
403 resp = client.get(apibase+"/resourcePools/" +
404 resource_pool_id1+"/resources/"+resource_id1)
405 assert resp.status_code == 404
407 deployment_manager_id1 = str(uuid.uuid4())
408 resp = client.get(apibase+"/deploymentManagers/" +
409 deployment_manager_id1)
410 assert resp.status_code == 404
412 subscription_id1 = str(uuid.uuid4())
413 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
414 assert resp.status_code == 404
417 def test_flask_post(mock_flask_uow):
418 session, app = mock_flask_uow
419 apibase = config.get_o2ims_api_base()
421 with app.test_client() as client:
422 session.return_value.execute.return_value = []
424 sub_callback = 'http://subscription/callback/url'
425 resp = client.post(apibase+'/subscriptions', json={
426 'callback': sub_callback,
427 'consumerSubscriptionId': 'consumerSubId1',
430 assert resp.status_code == 201
431 assert 'subscriptionId' in resp.get_json()
434 def test_flask_delete(mock_flask_uow):
435 session, app = mock_flask_uow
436 apibase = config.get_o2ims_api_base()
438 with app.test_client() as client:
439 session.return_value.execute.return_value.first.return_value = {}
441 subscription_id1 = str(uuid.uuid4())
442 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
443 assert resp.status_code == 204
446 def test_flask_not_allowed(mock_flask_uow):
447 _, app = mock_flask_uow
448 apibase = config.get_o2ims_api_base()
450 with app.test_client() as client:
451 # Testing resource type not support method
452 ##########################
453 uri = apibase + "/resourceTypes"
454 resp = client.post(uri)
455 assert resp.status == '405 METHOD NOT ALLOWED'
456 resp = client.put(uri)
457 assert resp.status == '405 METHOD NOT ALLOWED'
458 resp = client.patch(uri)
459 assert resp.status == '405 METHOD NOT ALLOWED'
460 resp = client.delete(uri)
461 assert resp.status == '405 METHOD NOT ALLOWED'
463 resource_type_id1 = str(uuid.uuid4())
464 uri = apibase + "/resourceTypes/" + resource_type_id1
465 resp = client.post(uri)
466 assert resp.status == '405 METHOD NOT ALLOWED'
467 resp = client.put(uri)
468 assert resp.status == '405 METHOD NOT ALLOWED'
469 resp = client.patch(uri)
470 assert resp.status == '405 METHOD NOT ALLOWED'
471 resp = client.delete(uri)
472 assert resp.status == '405 METHOD NOT ALLOWED'
474 # Testing resource pool not support method
475 ##########################
476 uri = apibase + "/resourcePools"
477 resp = client.post(uri)
478 assert resp.status == '405 METHOD NOT ALLOWED'
479 resp = client.put(uri)
480 assert resp.status == '405 METHOD NOT ALLOWED'
481 resp = client.patch(uri)
482 assert resp.status == '405 METHOD NOT ALLOWED'
483 resp = client.delete(uri)
484 assert resp.status == '405 METHOD NOT ALLOWED'
486 resource_pool_id1 = str(uuid.uuid4())
487 uri = apibase + "/resourcePools/" + resource_pool_id1
488 resp = client.post(uri)
489 assert resp.status == '405 METHOD NOT ALLOWED'
490 resp = client.put(uri)
491 assert resp.status == '405 METHOD NOT ALLOWED'
492 resp = client.patch(uri)
493 assert resp.status == '405 METHOD NOT ALLOWED'
494 resp = client.delete(uri)
495 assert resp.status == '405 METHOD NOT ALLOWED'
497 # Testing resource not support method
498 ##########################
499 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
500 resp = client.post(uri)
501 assert resp.status == '405 METHOD NOT ALLOWED'
502 resp = client.put(uri)
503 assert resp.status == '405 METHOD NOT ALLOWED'
504 resp = client.patch(uri)
505 assert resp.status == '405 METHOD NOT ALLOWED'
506 resp = client.delete(uri)
507 assert resp.status == '405 METHOD NOT ALLOWED'
509 resource_id1 = str(uuid.uuid4())
510 uri = apibase + "/resourcePools/" + \
511 resource_pool_id1 + "/resources/" + resource_id1
512 resp = client.post(uri)
513 assert resp.status == '405 METHOD NOT ALLOWED'
514 resp = client.put(uri)
515 assert resp.status == '405 METHOD NOT ALLOWED'
516 resp = client.patch(uri)
517 assert resp.status == '405 METHOD NOT ALLOWED'
518 resp = client.delete(uri)
519 assert resp.status == '405 METHOD NOT ALLOWED'
521 # Testing deployment managers not support method
522 ##########################
523 uri = apibase + "/deploymentManagers"
524 resp = client.post(uri)
525 assert resp.status == '405 METHOD NOT ALLOWED'
526 resp = client.put(uri)
527 assert resp.status == '405 METHOD NOT ALLOWED'
528 resp = client.patch(uri)
529 assert resp.status == '405 METHOD NOT ALLOWED'
530 resp = client.delete(uri)
531 assert resp.status == '405 METHOD NOT ALLOWED'
533 deployment_manager_id1 = str(uuid.uuid4())
534 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
535 resp = client.post(uri)
536 assert resp.status == '405 METHOD NOT ALLOWED'
537 resp = client.put(uri)
538 assert resp.status == '405 METHOD NOT ALLOWED'
539 resp = client.patch(uri)
540 assert resp.status == '405 METHOD NOT ALLOWED'
541 resp = client.delete(uri)
542 assert resp.status == '405 METHOD NOT ALLOWED'
544 # Testing subscriptions not support method
545 ##########################
546 uri = apibase + "/subscriptions"
547 resp = client.put(uri)
548 assert resp.status == '405 METHOD NOT ALLOWED'
549 resp = client.patch(uri)
550 assert resp.status == '405 METHOD NOT ALLOWED'
551 resp = client.delete(uri)
552 assert resp.status == '405 METHOD NOT ALLOWED'
554 subscription_id1 = str(uuid.uuid4())
555 uri = apibase + "/subscriptions/" + subscription_id1
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'