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 o2ims 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 deployment_manager_list = ocloud_view.subscriptions(uow)
265 assert str(deployment_manager_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 deployment_manager_res = ocloud_view.subscription_one(
277 subscription_id1, uow)
278 assert deployment_manager_res is None
280 session.return_value.execute.return_value.first.return_value = {
281 "deploymentManagerId": subscription_id1,
284 deployment_manager_res = ocloud_view.subscription_one(
285 subscription_id1, uow)
286 assert str(deployment_manager_res.get(
287 "deploymentManagerId")) == subscription_id1
290 def test_flask_get_list(mock_flask_uow):
291 session, client = mock_flask_uow
292 session.return_value.execute.return_value = []
293 apibase = config.get_o2ims_api_base()
295 # Get list and return empty list
296 ##########################
297 resp = client.get(apibase+"/")
298 assert resp.get_data() == b'[]\n'
300 resp = client.get(apibase+"/resourceTypes")
301 assert resp.get_data() == b'[]\n'
303 resp = client.get(apibase+"/resourcePools")
304 assert resp.get_data() == b'[]\n'
306 resource_pool_id1 = str(uuid.uuid4())
307 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources")
308 assert resp.get_data() == b'[]\n'
310 resp = client.get(apibase+"/deploymentManagers")
311 assert resp.get_data() == b'[]\n'
314 def test_flask_get_one(mock_flask_uow):
315 session, client = mock_flask_uow
316 session.return_value.execute.return_value.first.return_value = None
317 apibase = config.get_o2ims_api_base()
319 # Get one and return 404
320 ###########################
321 resource_type_id1 = str(uuid.uuid4())
322 resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
323 assert resp.status_code == 404
325 resource_pool_id1 = str(uuid.uuid4())
326 resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
327 assert resp.status_code == 404
329 resource_id1 = str(uuid.uuid4())
330 resp = client.get(apibase+"/resourcePools/" +
331 resource_pool_id1+"/resources/"+resource_id1)
332 assert resp.status_code == 404
334 deployment_manager_id1 = str(uuid.uuid4())
335 resp = client.get(apibase+"/deploymentManagers/"+deployment_manager_id1)
336 assert resp.status_code == 404
338 subscription_id1 = str(uuid.uuid4())
339 resp = client.get(apibase+"/subscriptions/"+subscription_id1)
340 assert resp.status_code == 404
343 def test_flask_not_allowed(mock_flask_uow):
344 _, client = mock_flask_uow
345 apibase = config.get_o2ims_api_base()
347 # Testing resource type not support method
348 ##########################
349 uri = apibase + "/resourceTypes"
350 resp = client.post(uri)
351 assert resp.status == '405 METHOD NOT ALLOWED'
352 resp = client.put(uri)
353 assert resp.status == '405 METHOD NOT ALLOWED'
354 resp = client.patch(uri)
355 assert resp.status == '405 METHOD NOT ALLOWED'
356 resp = client.delete(uri)
357 assert resp.status == '405 METHOD NOT ALLOWED'
359 resource_type_id1 = str(uuid.uuid4())
360 uri = apibase + "/resourceTypes/" + resource_type_id1
361 resp = client.post(uri)
362 assert resp.status == '405 METHOD NOT ALLOWED'
363 resp = client.put(uri)
364 assert resp.status == '405 METHOD NOT ALLOWED'
365 resp = client.patch(uri)
366 assert resp.status == '405 METHOD NOT ALLOWED'
367 resp = client.delete(uri)
368 assert resp.status == '405 METHOD NOT ALLOWED'
370 # Testing resource pool not support method
371 ##########################
372 uri = apibase + "/resourcePools"
373 resp = client.post(uri)
374 assert resp.status == '405 METHOD NOT ALLOWED'
375 resp = client.put(uri)
376 assert resp.status == '405 METHOD NOT ALLOWED'
377 resp = client.patch(uri)
378 assert resp.status == '405 METHOD NOT ALLOWED'
379 resp = client.delete(uri)
380 assert resp.status == '405 METHOD NOT ALLOWED'
382 resource_pool_id1 = str(uuid.uuid4())
383 uri = apibase + "/resourcePools/" + resource_pool_id1
384 resp = client.post(uri)
385 assert resp.status == '405 METHOD NOT ALLOWED'
386 resp = client.put(uri)
387 assert resp.status == '405 METHOD NOT ALLOWED'
388 resp = client.patch(uri)
389 assert resp.status == '405 METHOD NOT ALLOWED'
390 resp = client.delete(uri)
391 assert resp.status == '405 METHOD NOT ALLOWED'
393 # Testing resource not support method
394 ##########################
395 uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
396 resp = client.post(uri)
397 assert resp.status == '405 METHOD NOT ALLOWED'
398 resp = client.put(uri)
399 assert resp.status == '405 METHOD NOT ALLOWED'
400 resp = client.patch(uri)
401 assert resp.status == '405 METHOD NOT ALLOWED'
402 resp = client.delete(uri)
403 assert resp.status == '405 METHOD NOT ALLOWED'
405 resource_id1 = str(uuid.uuid4())
406 uri = apibase + "/resourcePools/" + \
407 resource_pool_id1 + "/resources/" + resource_id1
408 resp = client.post(uri)
409 assert resp.status == '405 METHOD NOT ALLOWED'
410 resp = client.put(uri)
411 assert resp.status == '405 METHOD NOT ALLOWED'
412 resp = client.patch(uri)
413 assert resp.status == '405 METHOD NOT ALLOWED'
414 resp = client.delete(uri)
415 assert resp.status == '405 METHOD NOT ALLOWED'
417 # Testing deployment managers not support method
418 ##########################
419 uri = apibase + "/deploymentManagers"
420 resp = client.post(uri)
421 assert resp.status == '405 METHOD NOT ALLOWED'
422 resp = client.put(uri)
423 assert resp.status == '405 METHOD NOT ALLOWED'
424 resp = client.patch(uri)
425 assert resp.status == '405 METHOD NOT ALLOWED'
426 resp = client.delete(uri)
427 assert resp.status == '405 METHOD NOT ALLOWED'
429 deployment_manager_id1 = str(uuid.uuid4())
430 uri = apibase + "/deploymentManagers/" + deployment_manager_id1
431 resp = client.post(uri)
432 assert resp.status == '405 METHOD NOT ALLOWED'
433 resp = client.put(uri)
434 assert resp.status == '405 METHOD NOT ALLOWED'
435 resp = client.patch(uri)
436 assert resp.status == '405 METHOD NOT ALLOWED'
437 resp = client.delete(uri)
438 assert resp.status == '405 METHOD NOT ALLOWED'
440 # Testing subscriptions not support method
441 ##########################
442 uri = apibase + "/subscriptions"
443 resp = client.post(uri)
444 assert resp.status == '405 METHOD NOT ALLOWED'
445 resp = client.put(uri)
446 assert resp.status == '405 METHOD NOT ALLOWED'
447 resp = client.patch(uri)
448 assert resp.status == '405 METHOD NOT ALLOWED'
449 resp = client.delete(uri)
450 assert resp.status == '405 METHOD NOT ALLOWED'
452 subscription_id1 = str(uuid.uuid4())
453 uri = apibase + "/subscriptions/" + subscription_id1
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'