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
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_view_olcouds(mock_uow):
99 session, uow = mock_uow
101 ocloud1_UUID = str(uuid.uuid4)
102 ocloud1 = MagicMock()
103 ocloud1.serialize.return_value = {
104 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
105 session.return_value.query.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}
138 session.return_value.query.return_value = [restype1]
140 resource_type_list = ocloud_view.resource_types(uow)
141 assert str(resource_type_list[0].get(
142 "resourceTypeId")) == resource_type_id1
145 def test_view_resource_type_one(mock_uow):
146 session, uow = mock_uow
148 resource_type_id1 = str(uuid.uuid4())
149 session.return_value.query.return_value.filter_by.return_value.first.\
150 return_value.serialize.return_value = None
153 resource_type_res = ocloud_view.resource_type_one(
154 resource_type_id1, uow)
155 assert resource_type_res is None
157 session.return_value.query.return_value.filter_by.return_value.first.\
158 return_value.serialize.return_value = {
159 "resourceTypeId": resource_type_id1}
161 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
162 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
165 def test_view_resource_pools(mock_uow):
166 session, uow = mock_uow
168 resource_pool_id1 = str(uuid.uuid4())
169 respool1 = MagicMock()
170 respool1.serialize.return_value = {
171 "resourcePoolId": resource_pool_id1}
172 session.return_value.query.return_value = [respool1]
174 resource_pool_list = ocloud_view.resource_pools(uow)
175 assert str(resource_pool_list[0].get(
176 "resourcePoolId")) == resource_pool_id1
179 def test_view_resource_pool_one(mock_uow):
180 session, uow = mock_uow
182 resource_pool_id1 = str(uuid.uuid4())
183 session.return_value.query.return_value.filter_by.return_value.first.\
184 return_value.serialize.return_value = None
187 resource_pool_res = ocloud_view.resource_pool_one(
188 resource_pool_id1, uow)
189 assert resource_pool_res is None
191 session.return_value.query.return_value.filter_by.return_value.first.\
192 return_value.serialize.return_value = {
193 "resourcePoolId": resource_pool_id1
196 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
197 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
200 def test_view_resources(mock_uow):
201 session, uow = mock_uow
203 resource_id1 = str(uuid.uuid4())
204 resource_pool_id1 = str(uuid.uuid4())
206 res1.serialize.return_value = {
207 "resourceId": resource_id1,
208 "resourcePoolId": resource_pool_id1
210 session.return_value.query.return_value.filter_by.return_value = [res1]
212 resource_list = ocloud_view.resources(resource_pool_id1, uow)
213 assert str(resource_list[0].get("resourceId")) == resource_id1
214 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
217 def test_view_resource_one(mock_uow):
218 session, uow = mock_uow
220 resource_id1 = str(uuid.uuid4())
221 resource_pool_id1 = str(uuid.uuid4())
222 session.return_value.query.return_value.filter_by.return_value.first.\
223 return_value.serialize.return_value = None
226 resource_res = ocloud_view.resource_one(resource_id1, uow)
227 assert resource_res is None
229 session.return_value.query.return_value.filter_by.return_value.first.\
230 return_value.serialize.return_value = {
231 "resourceId": resource_id1,
232 "resourcePoolId": resource_pool_id1
235 resource_res = ocloud_view.resource_one(resource_id1, uow)
236 assert str(resource_res.get("resourceId")) == resource_id1
239 def test_view_deployment_managers(mock_uow):
240 session, uow = mock_uow
242 deployment_manager_id1 = str(uuid.uuid4())
244 dm1.serialize.return_value = {
245 "deploymentManagerId": deployment_manager_id1,
247 session.return_value.query.return_value = [dm1]
249 deployment_manager_list = ocloud_view.deployment_managers(uow)
250 assert str(deployment_manager_list[0].get(
251 "deploymentManagerId")) == deployment_manager_id1
254 def test_view_deployment_manager_one(mock_uow):
255 session, uow = mock_uow
257 deployment_manager_id1 = str(uuid.uuid4())
258 session.return_value.query.return_value.filter_by.return_value.first.\
259 return_value.serialize.return_value = None
262 deployment_manager_res = ocloud_view.deployment_manager_one(
263 deployment_manager_id1, uow)
264 assert deployment_manager_res is None
266 dms_endpoint = "http://o2:30205/o2dms/v1/uuid"
267 session.return_value.query.return_value.filter_by.return_value.first.\
268 return_value.serialize.return_value = {
269 "deploymentManagerId": deployment_manager_id1,
270 "deploymentManagementServiceEndpoint": dms_endpoint,
275 deployment_manager_res = ocloud_view.deployment_manager_one(
276 deployment_manager_id1, uow)
277 assert str(deployment_manager_res.get(
278 "deploymentManagerId")) == deployment_manager_id1
279 assert str(deployment_manager_res.get(
280 'deploymentManagementServiceEndpoint')) == dms_endpoint
281 assert deployment_manager_res.get('profile') is None
284 profileName = ocloud.DeploymentManagerProfileSOL018
285 cluster_endpoint = "https://test_k8s:6443"
286 session.return_value.query.return_value.filter_by.return_value.first.\
287 return_value.serialize.return_value['profile'] = {
288 "cluster_api_endpoint": cluster_endpoint
290 deployment_manager_res = ocloud_view.deployment_manager_one(
291 deployment_manager_id1, uow, profile=profileName)
292 assert str(deployment_manager_res.get(
293 'deploymentManagementServiceEndpoint')) == cluster_endpoint
294 assert str(deployment_manager_res.get(
295 "profileName")) == profileName
298 profileName = 'wrong_profile'
299 session.return_value.query.return_value.filter_by.return_value.first.\
300 return_value.serialize.return_value['profile'] = {
301 "cluster_api_endpoint": cluster_endpoint
303 deployment_manager_res = ocloud_view.deployment_manager_one(
304 deployment_manager_id1, uow, profile=profileName)
305 assert deployment_manager_res is None
308 def test_view_subscriptions(mock_uow):
309 session, uow = mock_uow
311 subscription_id1 = str(uuid.uuid4())
313 sub1.serialize.return_value = {
314 "subscriptionId": subscription_id1,
316 session.return_value.query.return_value = [sub1]
318 subscription_list = ocloud_view.subscriptions(uow)
319 assert str(subscription_list[0].get(
320 "subscriptionId")) == subscription_id1
323 def test_view_subscription_one(mock_uow):
324 session, uow = mock_uow
326 subscription_id1 = str(uuid.uuid4())
327 session.return_value.query.return_value.filter_by.return_value.first.\
328 return_value.serialize.return_value = None
331 subscription_res = ocloud_view.subscription_one(
332 subscription_id1, uow)
333 assert subscription_res is None
335 session.return_value.query.return_value.filter_by.return_value.first.\
336 return_value.serialize.return_value = {
337 "subscriptionId": subscription_id1,
340 subscription_res = ocloud_view.subscription_one(
341 subscription_id1, uow)
342 assert str(subscription_res.get(
343 "subscriptionId")) == subscription_id1
346 def test_flask_get_list(mock_flask_uow):
347 session, app = mock_flask_uow
348 session.query.return_value = []
349 apibase = config.get_o2ims_api_base()
351 with app.test_client() as client:
352 # Get list and return empty list
353 ##########################
354 resp = client.get(apibase+"/resourceTypes")
355 assert resp.get_data() == b'[]\n'
357 resp = client.get(apibase+"/resourcePools")
358 assert resp.get_data() == b'[]\n'
360 resource_pool_id1 = str(uuid.uuid4())
361 resp = client.get(apibase+"/resourcePools/" +
362 resource_pool_id1+"/resources")
363 assert resp.get_data() == b'[]\n'
365 resp = client.get(apibase+"/deploymentManagers")
366 assert resp.get_data() == b'[]\n'
368 resp = client.get(apibase+"/subscriptions")
369 assert resp.get_data() == b'[]\n'
372 def test_flask_get_one(mock_flask_uow):
373 session, app = mock_flask_uow
375 session.return_value.query.return_value.filter_by.return_value.\
376 first.return_value = None
377 apibase = config.get_o2ims_api_base()
379 with app.test_client() as client:
380 # Get one and return 404
381 ###########################
382 resp = client.get(apibase+"/")
383 assert resp.status_code == 404
385 resource_type_id1 = str(uuid.uuid4())
386 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
387 assert resp.status_code == 404
389 resource_pool_id1 = str(uuid.uuid4())
390 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
391 assert resp.status_code == 404
393 resource_id1 = str(uuid.uuid4())
394 resp = client.get(apibase+"/resourcePools/" +
395 resource_pool_id1+"/resources/"+resource_id1)
396 assert resp.status_code == 404
398 deployment_manager_id1 = str(uuid.uuid4())
399 resp = client.get(apibase+"/deploymentManagers/" +
400 deployment_manager_id1)
401 assert resp.status_code == 404
403 subscription_id1 = str(uuid.uuid4())
404 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
405 assert resp.status_code == 404
408 def test_flask_post(mock_flask_uow):
409 session, app = mock_flask_uow
410 apibase = config.get_o2ims_api_base()
412 with app.test_client() as client:
413 session.return_value.execute.return_value = []
415 sub_callback = 'http://subscription/callback/url'
416 resp = client.post(apibase+'/subscriptions', json={
417 'callback': sub_callback,
418 'consumerSubscriptionId': 'consumerSubId1',
421 assert resp.status_code == 201
422 assert 'subscriptionId' in resp.get_json()
425 def test_flask_delete(mock_flask_uow):
426 session, app = mock_flask_uow
427 apibase = config.get_o2ims_api_base()
429 with app.test_client() as client:
430 session.return_value.execute.return_value.first.return_value = {}
432 subscription_id1 = str(uuid.uuid4())
433 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
434 assert resp.status_code == 204
437 def test_flask_not_allowed(mock_flask_uow):
438 _, app = mock_flask_uow
439 apibase = config.get_o2ims_api_base()
441 with app.test_client() as client:
442 # Testing resource type not support method
443 ##########################
444 uri = apibase + "/resourceTypes"
445 resp = client.post(uri)
446 assert resp.status == '405 METHOD NOT ALLOWED'
447 resp = client.put(uri)
448 assert resp.status == '405 METHOD NOT ALLOWED'
449 resp = client.patch(uri)
450 assert resp.status == '405 METHOD NOT ALLOWED'
451 resp = client.delete(uri)
452 assert resp.status == '405 METHOD NOT ALLOWED'
454 resource_type_id1 = str(uuid.uuid4())
455 uri = apibase + "/resourceTypes/" + resource_type_id1
456 resp = client.post(uri)
457 assert resp.status == '405 METHOD NOT ALLOWED'
458 resp = client.put(uri)
459 assert resp.status == '405 METHOD NOT ALLOWED'
460 resp = client.patch(uri)
461 assert resp.status == '405 METHOD NOT ALLOWED'
462 resp = client.delete(uri)
463 assert resp.status == '405 METHOD NOT ALLOWED'
465 # Testing resource pool not support method
466 ##########################
467 uri = apibase + "/resourcePools"
468 resp = client.post(uri)
469 assert resp.status == '405 METHOD NOT ALLOWED'
470 resp = client.put(uri)
471 assert resp.status == '405 METHOD NOT ALLOWED'
472 resp = client.patch(uri)
473 assert resp.status == '405 METHOD NOT ALLOWED'
474 resp = client.delete(uri)
475 assert resp.status == '405 METHOD NOT ALLOWED'
477 resource_pool_id1 = str(uuid.uuid4())
478 uri = apibase + "/resourcePools/" + resource_pool_id1
479 resp = client.post(uri)
480 assert resp.status == '405 METHOD NOT ALLOWED'
481 resp = client.put(uri)
482 assert resp.status == '405 METHOD NOT ALLOWED'
483 resp = client.patch(uri)
484 assert resp.status == '405 METHOD NOT ALLOWED'
485 resp = client.delete(uri)
486 assert resp.status == '405 METHOD NOT ALLOWED'
488 # Testing resource not support method
489 ##########################
490 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
491 resp = client.post(uri)
492 assert resp.status == '405 METHOD NOT ALLOWED'
493 resp = client.put(uri)
494 assert resp.status == '405 METHOD NOT ALLOWED'
495 resp = client.patch(uri)
496 assert resp.status == '405 METHOD NOT ALLOWED'
497 resp = client.delete(uri)
498 assert resp.status == '405 METHOD NOT ALLOWED'
500 resource_id1 = str(uuid.uuid4())
501 uri = apibase + "/resourcePools/" + \
502 resource_pool_id1 + "/resources/" + resource_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 deployment managers not support method
513 ##########################
514 uri = apibase + "/deploymentManagers"
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 deployment_manager_id1 = str(uuid.uuid4())
525 uri = apibase + "/deploymentManagers/" + deployment_manager_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 subscriptions not support method
536 ##########################
537 uri = apibase + "/subscriptions"
538 resp = client.put(uri)
539 assert resp.status == '405 METHOD NOT ALLOWED'
540 resp = client.patch(uri)
541 assert resp.status == '405 METHOD NOT ALLOWED'
542 resp = client.delete(uri)
543 assert resp.status == '405 METHOD NOT ALLOWED'
545 subscription_id1 = str(uuid.uuid4())
546 uri = apibase + "/subscriptions/" + subscription_id1
547 resp = client.post(uri)
548 assert resp.status == '405 METHOD NOT ALLOWED'
549 resp = client.put(uri)
550 assert resp.status == '405 METHOD NOT ALLOWED'
551 resp = client.patch(uri)
552 assert resp.status == '405 METHOD NOT ALLOWED'