tests/unit/test_ocloud.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import uuid
  16 from unittest.mock import MagicMock
  17 # from o2dms.domain import dms
  18
  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
  23
  24
  25 def setup_ocloud():
  26     ocloudid1 = str(uuid.uuid4())
  27     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
  28                             config.get_api_url(), "ocloud for unit test", 1)
  29     return ocloud1
  30
  31
  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
  37
  38
  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})
  49
  50
  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,
  56         ocloud1.oCloudId)
  57     assert resource_type_id1 is not None and \
  58         resource_type1.resourceTypeId == resource_type_id1
  59
  60
  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(),
  66         ocloud1.oCloudId)
  67     assert resource_pool_id1 is not None and \
  68         resource_pool1.resourcePoolId == resource_pool_id1
  69
  70
  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
  78
  79
  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
  88
  89
  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
  96
  97
  98 def test_view_olcouds(mock_uow):
  99     session, uow = mock_uow
 100
 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]
 106
 107     ocloud_list = ocloud_view.oclouds(uow)
 108     # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
 109     assert len(ocloud_list) == 1
 110
 111
 112 def test_view_olcoud_one(mock_uow):
 113     session, uow = mock_uow
 114
 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
 118
 119     # Query return None
 120     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 121     assert ocloud_res is None
 122
 123     session.return_value.query.return_value.filter_by.return_value.first.\
 124         return_value.serialize.return_value = {
 125             "oCloudId": ocloud1_UUID}
 126
 127     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 128     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
 129
 130
 131 def test_view_resource_types(mock_uow):
 132     session, uow = mock_uow
 133
 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]
 139
 140     resource_type_list = ocloud_view.resource_types(uow)
 141     assert str(resource_type_list[0].get(
 142         "resourceTypeId")) == resource_type_id1
 143
 144
 145 def test_view_resource_type_one(mock_uow):
 146     session, uow = mock_uow
 147
 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
 151
 152     # Query return None
 153     resource_type_res = ocloud_view.resource_type_one(
 154         resource_type_id1, uow)
 155     assert resource_type_res is None
 156
 157     session.return_value.query.return_value.filter_by.return_value.first.\
 158         return_value.serialize.return_value = {
 159             "resourceTypeId": resource_type_id1}
 160
 161     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
 162     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
 163
 164
 165 def test_view_resource_pools(mock_uow):
 166     session, uow = mock_uow
 167
 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]
 173
 174     resource_pool_list = ocloud_view.resource_pools(uow)
 175     assert str(resource_pool_list[0].get(
 176         "resourcePoolId")) == resource_pool_id1
 177
 178
 179 def test_view_resource_pool_one(mock_uow):
 180     session, uow = mock_uow
 181
 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
 185
 186     # Query return None
 187     resource_pool_res = ocloud_view.resource_pool_one(
 188         resource_pool_id1, uow)
 189     assert resource_pool_res is None
 190
 191     session.return_value.query.return_value.filter_by.return_value.first.\
 192         return_value.serialize.return_value = {
 193             "resourcePoolId": resource_pool_id1
 194         }
 195
 196     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 197     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 198
 199
 200 def test_view_resources(mock_uow):
 201     session, uow = mock_uow
 202
 203     resource_id1 = str(uuid.uuid4())
 204     resource_pool_id1 = str(uuid.uuid4())
 205     res1 = MagicMock()
 206     res1.serialize.return_value = {
 207         "resourceId": resource_id1,
 208         "resourcePoolId": resource_pool_id1
 209     }
 210     session.return_value.query.return_value.filter_by.return_value = [res1]
 211
 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
 215
 216
 217 def test_view_resource_one(mock_uow):
 218     session, uow = mock_uow
 219
 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
 224
 225     # Query return None
 226     resource_res = ocloud_view.resource_one(resource_id1, uow)
 227     assert resource_res is None
 228
 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
 233         }
 234
 235     resource_res = ocloud_view.resource_one(resource_id1, uow)
 236     assert str(resource_res.get("resourceId")) == resource_id1
 237
 238
 239 def test_view_deployment_managers(mock_uow):
 240     session, uow = mock_uow
 241
 242     deployment_manager_id1 = str(uuid.uuid4())
 243     dm1 = MagicMock()
 244     dm1.serialize.return_value = {
 245         "deploymentManagerId": deployment_manager_id1,
 246     }
 247     session.return_value.query.return_value = [dm1]
 248
 249     deployment_manager_list = ocloud_view.deployment_managers(uow)
 250     assert str(deployment_manager_list[0].get(
 251         "deploymentManagerId")) == deployment_manager_id1
 252
 253
 254 def test_view_deployment_manager_one(mock_uow):
 255     session, uow = mock_uow
 256
 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
 260
 261     # Query return None
 262     deployment_manager_res = ocloud_view.deployment_manager_one(
 263         deployment_manager_id1, uow)
 264     assert deployment_manager_res is None
 265
 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,
 271             "profile": {}
 272         }
 273
 274     # profile default
 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
 282
 283     # profile sol018
 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
 289         }
 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
 296
 297     # profile wrong name
 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
 302         }
 303     deployment_manager_res = ocloud_view.deployment_manager_one(
 304         deployment_manager_id1, uow, profile=profileName)
 305     assert deployment_manager_res is None
 306
 307
 308 def test_view_subscriptions(mock_uow):
 309     session, uow = mock_uow
 310
 311     subscription_id1 = str(uuid.uuid4())
 312     sub1 = MagicMock()
 313     sub1.serialize.return_value = {
 314         "subscriptionId": subscription_id1,
 315     }
 316     session.return_value.query.return_value = [sub1]
 317
 318     subscription_list = ocloud_view.subscriptions(uow)
 319     assert str(subscription_list[0].get(
 320         "subscriptionId")) == subscription_id1
 321
 322
 323 def test_view_subscription_one(mock_uow):
 324     session, uow = mock_uow
 325
 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
 329
 330     # Query return None
 331     subscription_res = ocloud_view.subscription_one(
 332         subscription_id1, uow)
 333     assert subscription_res is None
 334
 335     session.return_value.query.return_value.filter_by.return_value.first.\
 336         return_value.serialize.return_value = {
 337             "subscriptionId": subscription_id1,
 338         }
 339
 340     subscription_res = ocloud_view.subscription_one(
 341         subscription_id1, uow)
 342     assert str(subscription_res.get(
 343         "subscriptionId")) == subscription_id1
 344
 345
 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()
 350
 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'
 356
 357         resp = client.get(apibase+"/resourcePools")
 358         assert resp.get_data() == b'[]\n'
 359
 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'
 364
 365         resp = client.get(apibase+"/deploymentManagers")
 366         assert resp.get_data() == b'[]\n'
 367
 368         resp = client.get(apibase+"/subscriptions")
 369         assert resp.get_data() == b'[]\n'
 370
 371
 372 def test_flask_get_one(mock_flask_uow):
 373     session, app = mock_flask_uow
 374
 375     session.return_value.query.return_value.filter_by.return_value.\
 376         first.return_value = None
 377     apibase = config.get_o2ims_api_base()
 378
 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
 384
 385         resource_type_id1 = str(uuid.uuid4())
 386         resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
 387         assert resp.status_code == 404
 388
 389         resource_pool_id1 = str(uuid.uuid4())
 390         resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
 391         assert resp.status_code == 404
 392
 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
 397
 398         deployment_manager_id1 = str(uuid.uuid4())
 399         resp = client.get(apibase+"/deploymentManagers/" +
 400                           deployment_manager_id1)
 401         assert resp.status_code == 404
 402
 403         subscription_id1 = str(uuid.uuid4())
 404         resp = client.get(apibase+"/subscriptions/"+subscription_id1)
 405         assert resp.status_code == 404
 406
 407
 408 def test_flask_post(mock_flask_uow):
 409     session, app = mock_flask_uow
 410     apibase = config.get_o2ims_api_base()
 411
 412     with app.test_client() as client:
 413         session.return_value.execute.return_value = []
 414
 415         sub_callback = 'http://subscription/callback/url'
 416         resp = client.post(apibase+'/subscriptions', json={
 417             'callback': sub_callback,
 418             'consumerSubscriptionId': 'consumerSubId1',
 419             'filter': 'empty'
 420         })
 421         assert resp.status_code == 201
 422         assert 'subscriptionId' in resp.get_json()
 423
 424
 425 def test_flask_delete(mock_flask_uow):
 426     session, app = mock_flask_uow
 427     apibase = config.get_o2ims_api_base()
 428
 429     with app.test_client() as client:
 430         session.return_value.execute.return_value.first.return_value = {}
 431
 432         subscription_id1 = str(uuid.uuid4())
 433         resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
 434         assert resp.status_code == 204
 435
 436
 437 def test_flask_not_allowed(mock_flask_uow):
 438     _, app = mock_flask_uow
 439     apibase = config.get_o2ims_api_base()
 440
 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'
 453
 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'
 464
 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'
 476
 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'
 487
 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'
 499
 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'
 511
 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'
 523
 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'
 534
 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'
 544
 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'