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.
17 from o2ims.domain import ocloud
18 from o2ims.domain import resource_type as rt
19 from o2ims.views import ocloud_view
20 from o2common.config import config
24 ocloudid1 = str(uuid.uuid4())
25 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
26 config.get_api_url(), "ocloud for unit test", 1)
30 def test_new_ocloud():
31 ocloudid1 = str(uuid.uuid4())
32 ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
33 config.get_api_url(), "ocloud for unit test", 1)
34 assert ocloudid1 is not None and ocloud1.oCloudId == ocloudid1
37 # def test_add_ocloud_with_dms():
38 # ocloud1 = setup_ocloud()
39 # dmsid = str(uuid.uuid4())
40 # dms = ocloud.DeploymentManager(
41 # dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
42 # ocloud1.addDeploymentManager(dms)
43 # ocloud1.addDeploymentManager(dms)
44 # assert len(ocloud1.deploymentManagers) == 1
45 # # repo.update(ocloud1.oCloudId, {
46 # # "deploymentManagers": ocloud1.deploymentManagers})
49 def test_new_resource_type():
50 ocloud1 = setup_ocloud()
51 resource_type_id1 = str(uuid.uuid4())
52 resource_type1 = ocloud.ResourceType(
53 resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
55 assert resource_type_id1 is not None and \
56 resource_type1.resourceTypeId == resource_type_id1
59 def test_new_resource_pool():
60 ocloud1 = setup_ocloud()
61 resource_pool_id1 = str(uuid.uuid4())
62 resource_pool1 = ocloud.ResourcePool(
63 resource_pool_id1, "resourcepool1", config.get_api_url(),
65 assert resource_pool_id1 is not None and \
66 resource_pool1.resourcePoolId == resource_pool_id1
69 def test_new_resource():
70 resource_id1 = str(uuid.uuid4())
71 resource_type_id1 = str(uuid.uuid4())
72 resource_pool_id1 = str(uuid.uuid4())
73 resource1 = ocloud.Resource(
74 resource_id1, resource_type_id1, resource_pool_id1)
75 assert resource_id1 is not None and resource1.resourceId == resource_id1
78 def test_new_deployment_manager():
79 ocloud_id1 = str(uuid.uuid4())
80 deployment_manager_id1 = str(uuid.uuid4())
81 deployment_manager1 = ocloud.DeploymentManager(
82 deployment_manager_id1, "k8s1", ocloud_id1,
83 config.get_api_url()+"/k8s1")
84 assert deployment_manager_id1 is not None and deployment_manager1.\
85 deploymentManagerId == deployment_manager_id1
88 def test_new_subscription():
89 subscription_id1 = str(uuid.uuid4())
90 subscription1 = ocloud.Subscription(
91 subscription_id1, "https://callback/uri/write/here")
92 assert subscription_id1 is not None and\
93 subscription1.subscriptionId == subscription_id1
96 def test_view_olcouds(mock_uow):
97 session, uow = mock_uow
99 ocloud1_UUID = str(uuid.uuid4)
100 session.return_value.execute.return_value = [
101 {"oCloudId": ocloud1_UUID}]
103 ocloud_list = ocloud_view.oclouds(uow)
104 assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
107 def test_view_olcoud_one(mock_uow):
108 session, uow = mock_uow
110 ocloud1_UUID = str(uuid.uuid4)
111 session.return_value.execute.return_value.first.return_value = None
114 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
115 assert ocloud_res is None
117 session.return_value.execute.return_value.first.return_value = {
118 "oCloudId": ocloud1_UUID}
120 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
121 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
124 def test_view_resource_types(mock_uow):
125 session, uow = mock_uow
127 resource_type_id1 = str(uuid.uuid4())
128 session.return_value.execute.return_value = [
129 {"resourceTypeId": resource_type_id1}
132 resource_type_list = ocloud_view.resource_types(uow)
133 assert str(resource_type_list[0].get(
134 "resourceTypeId")) == resource_type_id1
137 def test_view_resource_type_one(mock_uow):
138 session, uow = mock_uow
140 resource_type_id1 = str(uuid.uuid4())
141 session.return_value.execute.return_value.first.return_value = None
144 resource_type_res = ocloud_view.resource_type_one(
145 resource_type_id1, uow)
146 assert resource_type_res is None
148 session.return_value.execute.return_value.first.return_value = {
149 "resourceTypeId": resource_type_id1}
151 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
152 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
155 def test_view_resource_pools(mock_uow):
156 session, uow = mock_uow
158 resource_pool_id1 = str(uuid.uuid4())
159 session.return_value.execute.return_value = [
160 {"resourcePoolId": resource_pool_id1}
163 resource_pool_list = ocloud_view.resource_pools(uow)
164 assert str(resource_pool_list[0].get(
165 "resourcePoolId")) == resource_pool_id1
168 def test_view_resource_pool_one(mock_uow):
169 session, uow = mock_uow
171 resource_pool_id1 = str(uuid.uuid4())
172 session.return_value.execute.return_value.first.return_value = None
175 resource_pool_res = ocloud_view.resource_pool_one(
176 resource_pool_id1, uow)
177 assert resource_pool_res is None
179 session.return_value.execute.return_value.first.return_value = {
180 "resourcePoolId": resource_pool_id1
183 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
184 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
187 def test_view_resources(mock_uow):
188 session, uow = mock_uow
190 resource_id1 = str(uuid.uuid4())
191 resource_pool_id1 = str(uuid.uuid4())
192 session.return_value.execute.return_value = [{
193 "resourceId": resource_id1,
194 "resourcePoolId": resource_pool_id1
197 resource_list = ocloud_view.resources(resource_pool_id1, uow)
198 assert str(resource_list[0].get("resourceId")) == resource_id1
199 assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
202 def test_view_resource_one(mock_uow):
203 session, uow = mock_uow
205 resource_id1 = str(uuid.uuid4())
206 resource_pool_id1 = str(uuid.uuid4())
207 session.return_value.execute.return_value.first.return_value = None
210 resource_res = ocloud_view.resource_one(resource_id1, uow)
211 assert resource_res is None
213 session.return_value.execute.return_value.first.return_value = {
214 "resourceId": resource_id1,
215 "resourcePoolId": resource_pool_id1
218 resource_res = ocloud_view.resource_one(resource_id1, uow)
219 assert str(resource_res.get("resourceId")) == resource_id1
222 def test_view_deployment_managers(mock_uow):
223 session, uow = mock_uow
225 deployment_manager_id1 = str(uuid.uuid4())
226 session.return_value.execute.return_value = [{
227 "deploymentManagerId": deployment_manager_id1,
230 deployment_manager_list = ocloud_view.deployment_managers(uow)
231 assert str(deployment_manager_list[0].get(
232 "deploymentManagerId")) == deployment_manager_id1
235 def test_view_deployment_manager_one(mock_uow):
236 session, uow = mock_uow
238 deployment_manager_id1 = str(uuid.uuid4())
239 session.return_value.execute.return_value.first.return_value = None
242 deployment_manager_res = ocloud_view.deployment_manager_one(
243 deployment_manager_id1, uow)
244 assert deployment_manager_res is None
246 session.return_value.execute.return_value.first.return_value = {
247 "deploymentManagerId": deployment_manager_id1,
250 deployment_manager_res = ocloud_view.deployment_manager_one(
251 deployment_manager_id1, uow)
252 assert str(deployment_manager_res.get(
253 "deploymentManagerId")) == deployment_manager_id1
256 def test_view_subscriptions(mock_uow):
257 session, uow = mock_uow
259 subscription_id1 = str(uuid.uuid4())
260 session.return_value.execute.return_value = [{
261 "subscriptionId": subscription_id1,
264 subscription_list = ocloud_view.subscriptions(uow)
265 assert str(subscription_list[0].get(
266 "subscriptionId")) == subscription_id1
269 def test_view_subscription_one(mock_uow):
270 session, uow = mock_uow
272 subscription_id1 = str(uuid.uuid4())
273 session.return_value.execute.return_value.first.return_value = None
276 subscription_res = ocloud_view.subscription_one(
277 subscription_id1, uow)
278 assert subscription_res is None
280 session.return_value.execute.return_value.first.return_value = {
281 "subscriptionId": subscription_id1,
284 subscription_res = ocloud_view.subscription_one(
285 subscription_id1, uow)
286 assert str(subscription_res.get(
287 "subscriptionId")) == subscription_id1
290 def test_flask_get_list(mock_flask_uow):
291 session, app = mock_flask_uow
292 session.return_value.execute.return_value = []
293 apibase = config.get_o2ims_api_base()
295 with app.test_client() as client:
296 # Get list and return empty list
297 ##########################
298 resp = client.get(apibase+"/resourceTypes")
299 assert resp.get_data() == b'[]\n'
301 resp = client.get(apibase+"/resourcePools")
302 assert resp.get_data() == b'[]\n'
304 resource_pool_id1 = str(uuid.uuid4())
305 resp = client.get(apibase+"/resourcePools/" +
306 resource_pool_id1+"/resources")
307 assert resp.get_data() == b'[]\n'
309 resp = client.get(apibase+"/deploymentManagers")
310 assert resp.get_data() == b'[]\n'
312 resp = client.get(apibase+"/subscriptions")
313 assert resp.get_data() == b'[]\n'
316 def test_flask_get_one(mock_flask_uow):
317 session, app = mock_flask_uow
318 session.return_value.execute.return_value.first.return_value = None
319 apibase = config.get_o2ims_api_base()
321 with app.test_client() as client:
322 # Get one and return 404
323 ###########################
324 resp = client.get(apibase+"/")
325 assert resp.status_code == 404
327 resource_type_id1 = str(uuid.uuid4())
328 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
329 assert resp.status_code == 404
331 resource_pool_id1 = str(uuid.uuid4())
332 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
333 assert resp.status_code == 404
335 resource_id1 = str(uuid.uuid4())
336 resp = client.get(apibase+"/resourcePools/" +
337 resource_pool_id1+"/resources/"+resource_id1)
338 assert resp.status_code == 404
340 deployment_manager_id1 = str(uuid.uuid4())
341 resp = client.get(apibase+"/deploymentManagers/" +
342 deployment_manager_id1)
343 assert resp.status_code == 404
345 subscription_id1 = str(uuid.uuid4())
346 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
347 assert resp.status_code == 404
350 def test_flask_post(mock_flask_uow):
351 session, app = mock_flask_uow
352 apibase = config.get_o2ims_api_base()
354 with app.test_client() as client:
355 session.return_value.execute.return_value = []
357 sub_callback = 'http://subscription/callback/url'
358 resp = client.post(apibase+'/subscriptions', json={
359 'callback': sub_callback,
360 'consumerSubscriptionId': 'consumerSubId1',
363 assert resp.status_code == 201
364 assert 'subscriptionId' in resp.get_json()
367 def test_flask_delete(mock_flask_uow):
368 session, app = mock_flask_uow
369 apibase = config.get_o2ims_api_base()
371 with app.test_client() as client:
372 session.return_value.execute.return_value.first.return_value = {}
374 subscription_id1 = str(uuid.uuid4())
375 resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
376 assert resp.status_code == 204
379 def test_flask_not_allowed(mock_flask_uow):
380 _, app = mock_flask_uow
381 apibase = config.get_o2ims_api_base()
383 with app.test_client() as client:
384 # Testing resource type not support method
385 ##########################
386 uri = apibase + "/resourceTypes"
387 resp = client.post(uri)
388 assert resp.status == '405 METHOD NOT ALLOWED'
389 resp = client.put(uri)
390 assert resp.status == '405 METHOD NOT ALLOWED'
391 resp = client.patch(uri)
392 assert resp.status == '405 METHOD NOT ALLOWED'
393 resp = client.delete(uri)
394 assert resp.status == '405 METHOD NOT ALLOWED'
396 resource_type_id1 = str(uuid.uuid4())
397 uri = apibase + "/resourceTypes/" + resource_type_id1
398 resp = client.post(uri)
399 assert resp.status == '405 METHOD NOT ALLOWED'
400 resp = client.put(uri)
401 assert resp.status == '405 METHOD NOT ALLOWED'
402 resp = client.patch(uri)
403 assert resp.status == '405 METHOD NOT ALLOWED'
404 resp = client.delete(uri)
405 assert resp.status == '405 METHOD NOT ALLOWED'
407 # Testing resource pool not support method
408 ##########################
409 uri = apibase + "/resourcePools"
410 resp = client.post(uri)
411 assert resp.status == '405 METHOD NOT ALLOWED'
412 resp = client.put(uri)
413 assert resp.status == '405 METHOD NOT ALLOWED'
414 resp = client.patch(uri)
415 assert resp.status == '405 METHOD NOT ALLOWED'
416 resp = client.delete(uri)
417 assert resp.status == '405 METHOD NOT ALLOWED'
419 resource_pool_id1 = str(uuid.uuid4())
420 uri = apibase + "/resourcePools/" + resource_pool_id1
421 resp = client.post(uri)
422 assert resp.status == '405 METHOD NOT ALLOWED'
423 resp = client.put(uri)
424 assert resp.status == '405 METHOD NOT ALLOWED'
425 resp = client.patch(uri)
426 assert resp.status == '405 METHOD NOT ALLOWED'
427 resp = client.delete(uri)
428 assert resp.status == '405 METHOD NOT ALLOWED'
430 # Testing resource not support method
431 ##########################
432 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
433 resp = client.post(uri)
434 assert resp.status == '405 METHOD NOT ALLOWED'
435 resp = client.put(uri)
436 assert resp.status == '405 METHOD NOT ALLOWED'
437 resp = client.patch(uri)
438 assert resp.status == '405 METHOD NOT ALLOWED'
439 resp = client.delete(uri)
440 assert resp.status == '405 METHOD NOT ALLOWED'
442 resource_id1 = str(uuid.uuid4())
443 uri = apibase + "/resourcePools/" + \
444 resource_pool_id1 + "/resources/" + resource_id1
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 # Testing deployment managers not support method
455 ##########################
456 uri = apibase + "/deploymentManagers"
457 resp = client.post(uri)
458 assert resp.status == '405 METHOD NOT ALLOWED'
459 resp = client.put(uri)
460 assert resp.status == '405 METHOD NOT ALLOWED'
461 resp = client.patch(uri)
462 assert resp.status == '405 METHOD NOT ALLOWED'
463 resp = client.delete(uri)
464 assert resp.status == '405 METHOD NOT ALLOWED'
466 deployment_manager_id1 = str(uuid.uuid4())
467 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
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 # Testing subscriptions not support method
478 ##########################
479 uri = apibase + "/subscriptions"
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 subscription_id1 = str(uuid.uuid4())
488 uri = apibase + "/subscriptions/" + subscription_id1
489 resp = client.post(uri)
490 assert resp.status == '405 METHOD NOT ALLOWED'
491 resp = client.put(uri)
492 assert resp.status == '405 METHOD NOT ALLOWED'
493 resp = client.patch(uri)
494 assert resp.status == '405 METHOD NOT ALLOWED'