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 = 'sol0018'
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 def test_view_subscriptions(mock_uow):
308 session, uow = mock_uow
310 subscription_id1 = str(uuid.uuid4())
312 sub1.serialize.return_value = {
313 "subscriptionId": subscription_id1,
315 session.return_value.query.return_value = [sub1]
317 subscription_list = ocloud_view.subscriptions(uow)
318 assert str(subscription_list[0].get(
319 "subscriptionId")) == subscription_id1
322 def test_view_subscription_one(mock_uow):
323 session, uow = mock_uow
325 subscription_id1 = str(uuid.uuid4())
326 session.return_value.query.return_value.filter_by.return_value.first.\
327 return_value.serialize.return_value = None
330 subscription_res = ocloud_view.subscription_one(
331 subscription_id1, uow)
332 assert subscription_res is None
334 session.return_value.query.return_value.filter_by.return_value.first.\
335 return_value.serialize.return_value = {
336 "subscriptionId": subscription_id1,
339 subscription_res = ocloud_view.subscription_one(
340 subscription_id1, uow)
341 assert str(subscription_res.get(
342 "subscriptionId")) == subscription_id1
345 def test_flask_get_list(mock_flask_uow):
346 session, app = mock_flask_uow
347 session.query.return_value = []
348 apibase = config.get_o2ims_api_base()
350 with app.test_client() as client:
351 # Get list and return empty list
352 ##########################
353 resp = client.get(apibase+"/resourceTypes")
354 assert resp.get_data() == b'[]\n'
356 resp = client.get(apibase+"/resourcePools")
357 assert resp.get_data() == b'[]\n'
359 resource_pool_id1 = str(uuid.uuid4())
360 resp = client.get(apibase+"/resourcePools/" +
361 resource_pool_id1+"/resources")
362 assert resp.get_data() == b'[]\n'
364 resp = client.get(apibase+"/deploymentManagers")
365 assert resp.get_data() == b'[]\n'
367 resp = client.get(apibase+"/subscriptions")
368 assert resp.get_data() == b'[]\n'
371 def test_flask_get_one(mock_flask_uow):
372 session, app = mock_flask_uow
374 session.return_value.query.return_value.filter_by.return_value.\
375 first.return_value = None
376 apibase = config.get_o2ims_api_base()
378 with app.test_client() as client:
379 # Get one and return 404
380 ###########################
381 resp = client.get(apibase+"/")
382 assert resp.status_code == 404
384 resource_type_id1 = str(uuid.uuid4())
385 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
386 assert resp.status_code == 404
388 resource_pool_id1 = str(uuid.uuid4())
389 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
390 assert resp.status_code == 404
392 resource_id1 = str(uuid.uuid4())
393 resp = client.get(apibase+"/resourcePools/" +
394 resource_pool_id1+"/resources/"+resource_id1)
395 assert resp.status_code == 404
397 deployment_manager_id1 = str(uuid.uuid4())
398 resp = client.get(apibase+"/deploymentManagers/" +
399 deployment_manager_id1)
400 assert resp.status_code == 404
402 subscription_id1 = str(uuid.uuid4())
403 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
404 assert resp.status_code == 404
407 def test_flask_post(mock_flask_uow):
408 session, app = mock_flask_uow
409 apibase = config.get_o2ims_api_base()
411 with app.test_client() as client:
412 session.return_value.execute.return_value = []
414 sub_callback = 'http://subscription/callback/url'
415 resp = client.post(apibase+'/subscriptions', json={
416 'callback': sub_callback,
417 'consumerSubscriptionId': 'consumerSubId1',
420 assert resp.status_code == 201
421 assert 'subscriptionId' in resp.get_json()
424 def test_flask_delete(mock_flask_uow):
425 session, app = mock_flask_uow
426 apibase = config.get_o2ims_api_base()
428 with app.test_client() as client:
429 session.return_value.execute.return_value.first.return_value = {}
431 subscription_id1 = str(uuid.uuid4())
432 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
433 assert resp.status_code == 204
436 def test_flask_not_allowed(mock_flask_uow):
437 _, app = mock_flask_uow
438 apibase = config.get_o2ims_api_base()
440 with app.test_client() as client:
441 # Testing resource type not support method
442 ##########################
443 uri = apibase + "/resourceTypes"
444 resp = client.post(uri)
445 assert resp.status == '405 METHOD NOT ALLOWED'
446 resp = client.put(uri)
447 assert resp.status == '405 METHOD NOT ALLOWED'
448 resp = client.patch(uri)
449 assert resp.status == '405 METHOD NOT ALLOWED'
450 resp = client.delete(uri)
451 assert resp.status == '405 METHOD NOT ALLOWED'
453 resource_type_id1 = str(uuid.uuid4())
454 uri = apibase + "/resourceTypes/" + resource_type_id1
455 resp = client.post(uri)
456 assert resp.status == '405 METHOD NOT ALLOWED'
457 resp = client.put(uri)
458 assert resp.status == '405 METHOD NOT ALLOWED'
459 resp = client.patch(uri)
460 assert resp.status == '405 METHOD NOT ALLOWED'
461 resp = client.delete(uri)
462 assert resp.status == '405 METHOD NOT ALLOWED'
464 # Testing resource pool not support method
465 ##########################
466 uri = apibase + "/resourcePools"
467 resp = client.post(uri)
468 assert resp.status == '405 METHOD NOT ALLOWED'
469 resp = client.put(uri)
470 assert resp.status == '405 METHOD NOT ALLOWED'
471 resp = client.patch(uri)
472 assert resp.status == '405 METHOD NOT ALLOWED'
473 resp = client.delete(uri)
474 assert resp.status == '405 METHOD NOT ALLOWED'
476 resource_pool_id1 = str(uuid.uuid4())
477 uri = apibase + "/resourcePools/" + resource_pool_id1
478 resp = client.post(uri)
479 assert resp.status == '405 METHOD NOT ALLOWED'
480 resp = client.put(uri)
481 assert resp.status == '405 METHOD NOT ALLOWED'
482 resp = client.patch(uri)
483 assert resp.status == '405 METHOD NOT ALLOWED'
484 resp = client.delete(uri)
485 assert resp.status == '405 METHOD NOT ALLOWED'
487 # Testing resource not support method
488 ##########################
489 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
490 resp = client.post(uri)
491 assert resp.status == '405 METHOD NOT ALLOWED'
492 resp = client.put(uri)
493 assert resp.status == '405 METHOD NOT ALLOWED'
494 resp = client.patch(uri)
495 assert resp.status == '405 METHOD NOT ALLOWED'
496 resp = client.delete(uri)
497 assert resp.status == '405 METHOD NOT ALLOWED'
499 resource_id1 = str(uuid.uuid4())
500 uri = apibase + "/resourcePools/" + \
501 resource_pool_id1 + "/resources/" + resource_id1
502 resp = client.post(uri)
503 assert resp.status == '405 METHOD NOT ALLOWED'
504 resp = client.put(uri)
505 assert resp.status == '405 METHOD NOT ALLOWED'
506 resp = client.patch(uri)
507 assert resp.status == '405 METHOD NOT ALLOWED'
508 resp = client.delete(uri)
509 assert resp.status == '405 METHOD NOT ALLOWED'
511 # Testing deployment managers not support method
512 ##########################
513 uri = apibase + "/deploymentManagers"
514 resp = client.post(uri)
515 assert resp.status == '405 METHOD NOT ALLOWED'
516 resp = client.put(uri)
517 assert resp.status == '405 METHOD NOT ALLOWED'
518 resp = client.patch(uri)
519 assert resp.status == '405 METHOD NOT ALLOWED'
520 resp = client.delete(uri)
521 assert resp.status == '405 METHOD NOT ALLOWED'
523 deployment_manager_id1 = str(uuid.uuid4())
524 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
525 resp = client.post(uri)
526 assert resp.status == '405 METHOD NOT ALLOWED'
527 resp = client.put(uri)
528 assert resp.status == '405 METHOD NOT ALLOWED'
529 resp = client.patch(uri)
530 assert resp.status == '405 METHOD NOT ALLOWED'
531 resp = client.delete(uri)
532 assert resp.status == '405 METHOD NOT ALLOWED'
534 # Testing subscriptions not support method
535 ##########################
536 uri = apibase + "/subscriptions"
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 subscription_id1 = str(uuid.uuid4())
545 uri = apibase + "/subscriptions/" + subscription_id1
546 resp = client.post(uri)
547 assert resp.status == '405 METHOD NOT ALLOWED'
548 resp = client.put(uri)
549 assert resp.status == '405 METHOD NOT ALLOWED'
550 resp = client.patch(uri)
551 assert resp.status == '405 METHOD NOT ALLOWED'