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
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}
137 session.return_value.query.return_value = [restype1]
139 resource_type_list = ocloud_view.resource_types(uow)
140 assert str(resource_type_list[0].get(
141 "resourceTypeId")) == resource_type_id1
144 def test_view_resource_type_one(mock_uow):
145 session, uow = mock_uow
147 resource_type_id1 = str(uuid.uuid4())
148 session.return_value.query.return_value.filter_by.return_value.first.\
149 return_value.serialize.return_value = None
152 resource_type_res = ocloud_view.resource_type_one(
153 resource_type_id1, uow)
154 assert resource_type_res is None
156 session.return_value.query.return_value.filter_by.return_value.first.\
157 return_value.serialize.return_value = {
158 "resourceTypeId": resource_type_id1}
160 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
161 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
164 def test_view_resource_pools(mock_uow):
165 session, uow = mock_uow
167 resource_pool_id1 = str(uuid.uuid4())
168 respool1 = MagicMock()
169 respool1.serialize.return_value = {
170 "resourcePoolId": resource_pool_id1}
171 session.return_value.query.return_value = [respool1]
173 resource_pool_list = ocloud_view.resource_pools(uow)
174 assert str(resource_pool_list[0].get(
175 "resourcePoolId")) == resource_pool_id1
178 def test_view_resource_pool_one(mock_uow):
179 session, uow = mock_uow
181 resource_pool_id1 = str(uuid.uuid4())
182 session.return_value.query.return_value.filter_by.return_value.first.\
183 return_value.serialize.return_value = None
186 resource_pool_res = ocloud_view.resource_pool_one(
187 resource_pool_id1, uow)
188 assert resource_pool_res is None
190 session.return_value.query.return_value.filter_by.return_value.first.\
191 return_value.serialize.return_value = {
192 "resourcePoolId": resource_pool_id1
195 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
196 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
199 def test_view_resources(mock_uow):
200 session, uow = mock_uow
202 resource_id1 = str(uuid.uuid4())
203 resource_pool_id1 = str(uuid.uuid4())
205 res1.serialize.return_value = {
206 "resourceId": resource_id1,
207 "resourcePoolId": resource_pool_id1
209 session.return_value.query.return_value.filter_by.return_value = [res1]
211 resource_list = ocloud_view.resources(resource_pool_id1, uow)
212 assert str(resource_list[0].get("resourceId")) == resource_id1
213 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
216 def test_view_resource_one(mock_uow):
217 session, uow = mock_uow
219 resource_id1 = str(uuid.uuid4())
220 resource_pool_id1 = str(uuid.uuid4())
221 session.return_value.query.return_value.filter_by.return_value.first.\
222 return_value.serialize.return_value = None
225 resource_res = ocloud_view.resource_one(resource_id1, uow)
226 assert resource_res is None
228 session.return_value.query.return_value.filter_by.return_value.first.\
229 return_value.serialize.return_value = {
230 "resourceId": resource_id1,
231 "resourcePoolId": resource_pool_id1
234 resource_res = ocloud_view.resource_one(resource_id1, uow)
235 assert str(resource_res.get("resourceId")) == resource_id1
238 def test_view_deployment_managers(mock_uow):
239 session, uow = mock_uow
241 deployment_manager_id1 = str(uuid.uuid4())
243 dm1.serialize.return_value = {
244 "deploymentManagerId": deployment_manager_id1,
246 session.return_value.query.return_value = [dm1]
248 deployment_manager_list = ocloud_view.deployment_managers(uow)
249 assert str(deployment_manager_list[0].get(
250 "deploymentManagerId")) == deployment_manager_id1
253 def test_view_deployment_manager_one(mock_uow):
254 session, uow = mock_uow
256 deployment_manager_id1 = str(uuid.uuid4())
257 session.return_value.query.return_value.filter_by.return_value.first.\
258 return_value.serialize.return_value = None
261 deployment_manager_res = ocloud_view.deployment_manager_one(
262 deployment_manager_id1, uow)
263 assert deployment_manager_res is None
265 session.return_value.query.return_value.filter_by.return_value.first.\
266 return_value.serialize.return_value = {
267 "deploymentManagerId": deployment_manager_id1,
270 deployment_manager_res = ocloud_view.deployment_manager_one(
271 deployment_manager_id1, uow)
272 assert str(deployment_manager_res.get(
273 "deploymentManagerId")) == deployment_manager_id1
276 def test_view_subscriptions(mock_uow):
277 session, uow = mock_uow
279 subscription_id1 = str(uuid.uuid4())
281 sub1.serialize.return_value = {
282 "subscriptionId": subscription_id1,
284 session.return_value.query.return_value = [sub1]
286 subscription_list = ocloud_view.subscriptions(uow)
287 assert str(subscription_list[0].get(
288 "subscriptionId")) == subscription_id1
291 def test_view_subscription_one(mock_uow):
292 session, uow = mock_uow
294 subscription_id1 = str(uuid.uuid4())
295 session.return_value.query.return_value.filter_by.return_value.first.\
296 return_value.serialize.return_value = None
299 subscription_res = ocloud_view.subscription_one(
300 subscription_id1, uow)
301 assert subscription_res is None
303 session.return_value.query.return_value.filter_by.return_value.first.\
304 return_value.serialize.return_value = {
305 "subscriptionId": subscription_id1,
308 subscription_res = ocloud_view.subscription_one(
309 subscription_id1, uow)
310 assert str(subscription_res.get(
311 "subscriptionId")) == subscription_id1
314 def test_flask_get_list(mock_flask_uow):
315 session, app = mock_flask_uow
316 session.query.return_value = []
317 apibase = config.get_o2ims_api_base()
319 with app.test_client() as client:
320 # Get list and return empty list
321 ##########################
322 resp = client.get(apibase+"/resourceTypes")
323 assert resp.get_data() == b'[]\n'
325 resp = client.get(apibase+"/resourcePools")
326 assert resp.get_data() == b'[]\n'
328 resource_pool_id1 = str(uuid.uuid4())
329 resp = client.get(apibase+"/resourcePools/" +
330 resource_pool_id1+"/resources")
331 assert resp.get_data() == b'[]\n'
333 resp = client.get(apibase+"/deploymentManagers")
334 assert resp.get_data() == b'[]\n'
336 resp = client.get(apibase+"/subscriptions")
337 assert resp.get_data() == b'[]\n'
340 def test_flask_get_one(mock_flask_uow):
341 session, app = mock_flask_uow
343 session.return_value.query.return_value.filter_by.return_value.\
344 first.return_value = None
345 apibase = config.get_o2ims_api_base()
347 with app.test_client() as client:
348 # Get one and return 404
349 ###########################
350 resp = client.get(apibase+"/")
351 assert resp.status_code == 404
353 resource_type_id1 = str(uuid.uuid4())
354 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
355 assert resp.status_code == 404
357 resource_pool_id1 = str(uuid.uuid4())
358 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
359 assert resp.status_code == 404
361 resource_id1 = str(uuid.uuid4())
362 resp = client.get(apibase+"/resourcePools/" +
363 resource_pool_id1+"/resources/"+resource_id1)
364 assert resp.status_code == 404
366 deployment_manager_id1 = str(uuid.uuid4())
367 resp = client.get(apibase+"/deploymentManagers/" +
368 deployment_manager_id1)
369 assert resp.status_code == 404
371 subscription_id1 = str(uuid.uuid4())
372 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
373 assert resp.status_code == 404
376 def test_flask_post(mock_flask_uow):
377 session, app = mock_flask_uow
378 apibase = config.get_o2ims_api_base()
380 with app.test_client() as client:
381 session.return_value.execute.return_value = []
383 sub_callback = 'http://subscription/callback/url'
384 resp = client.post(apibase+'/subscriptions', json={
385 'callback': sub_callback,
386 'consumerSubscriptionId': 'consumerSubId1',
389 assert resp.status_code == 201
390 assert 'subscriptionId' in resp.get_json()
393 def test_flask_delete(mock_flask_uow):
394 session, app = mock_flask_uow
395 apibase = config.get_o2ims_api_base()
397 with app.test_client() as client:
398 session.return_value.execute.return_value.first.return_value = {}
400 subscription_id1 = str(uuid.uuid4())
401 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
402 assert resp.status_code == 204
405 def test_flask_not_allowed(mock_flask_uow):
406 _, app = mock_flask_uow
407 apibase = config.get_o2ims_api_base()
409 with app.test_client() as client:
410 # Testing resource type not support method
411 ##########################
412 uri = apibase + "/resourceTypes"
413 resp = client.post(uri)
414 assert resp.status == '405 METHOD NOT ALLOWED'
415 resp = client.put(uri)
416 assert resp.status == '405 METHOD NOT ALLOWED'
417 resp = client.patch(uri)
418 assert resp.status == '405 METHOD NOT ALLOWED'
419 resp = client.delete(uri)
420 assert resp.status == '405 METHOD NOT ALLOWED'
422 resource_type_id1 = str(uuid.uuid4())
423 uri = apibase + "/resourceTypes/" + resource_type_id1
424 resp = client.post(uri)
425 assert resp.status == '405 METHOD NOT ALLOWED'
426 resp = client.put(uri)
427 assert resp.status == '405 METHOD NOT ALLOWED'
428 resp = client.patch(uri)
429 assert resp.status == '405 METHOD NOT ALLOWED'
430 resp = client.delete(uri)
431 assert resp.status == '405 METHOD NOT ALLOWED'
433 # Testing resource pool not support method
434 ##########################
435 uri = apibase + "/resourcePools"
436 resp = client.post(uri)
437 assert resp.status == '405 METHOD NOT ALLOWED'
438 resp = client.put(uri)
439 assert resp.status == '405 METHOD NOT ALLOWED'
440 resp = client.patch(uri)
441 assert resp.status == '405 METHOD NOT ALLOWED'
442 resp = client.delete(uri)
443 assert resp.status == '405 METHOD NOT ALLOWED'
445 resource_pool_id1 = str(uuid.uuid4())
446 uri = apibase + "/resourcePools/" + resource_pool_id1
447 resp = client.post(uri)
448 assert resp.status == '405 METHOD NOT ALLOWED'
449 resp = client.put(uri)
450 assert resp.status == '405 METHOD NOT ALLOWED'
451 resp = client.patch(uri)
452 assert resp.status == '405 METHOD NOT ALLOWED'
453 resp = client.delete(uri)
454 assert resp.status == '405 METHOD NOT ALLOWED'
456 # Testing resource not support method
457 ##########################
458 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
459 resp = client.post(uri)
460 assert resp.status == '405 METHOD NOT ALLOWED'
461 resp = client.put(uri)
462 assert resp.status == '405 METHOD NOT ALLOWED'
463 resp = client.patch(uri)
464 assert resp.status == '405 METHOD NOT ALLOWED'
465 resp = client.delete(uri)
466 assert resp.status == '405 METHOD NOT ALLOWED'
468 resource_id1 = str(uuid.uuid4())
469 uri = apibase + "/resourcePools/" + \
470 resource_pool_id1 + "/resources/" + resource_id1
471 resp = client.post(uri)
472 assert resp.status == '405 METHOD NOT ALLOWED'
473 resp = client.put(uri)
474 assert resp.status == '405 METHOD NOT ALLOWED'
475 resp = client.patch(uri)
476 assert resp.status == '405 METHOD NOT ALLOWED'
477 resp = client.delete(uri)
478 assert resp.status == '405 METHOD NOT ALLOWED'
480 # Testing deployment managers not support method
481 ##########################
482 uri = apibase + "/deploymentManagers"
483 resp = client.post(uri)
484 assert resp.status == '405 METHOD NOT ALLOWED'
485 resp = client.put(uri)
486 assert resp.status == '405 METHOD NOT ALLOWED'
487 resp = client.patch(uri)
488 assert resp.status == '405 METHOD NOT ALLOWED'
489 resp = client.delete(uri)
490 assert resp.status == '405 METHOD NOT ALLOWED'
492 deployment_manager_id1 = str(uuid.uuid4())
493 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
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 # Testing subscriptions not support method
504 ##########################
505 uri = apibase + "/subscriptions"
506 resp = client.put(uri)
507 assert resp.status == '405 METHOD NOT ALLOWED'
508 resp = client.patch(uri)
509 assert resp.status == '405 METHOD NOT ALLOWED'
510 resp = client.delete(uri)
511 assert resp.status == '405 METHOD NOT ALLOWED'
513 subscription_id1 = str(uuid.uuid4())
514 uri = apibase + "/subscriptions/" + subscription_id1
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'