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
  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
  21
  22
  23 def setup_ocloud():
  24     ocloudid1 = str(uuid.uuid4())
  25     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
  26                             config.get_api_url(), "ocloud for unit test", 1)
  27     return ocloud1
  28
  29
  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
  35
  36
  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})
  47
  48
  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,
  54         ocloud1.oCloudId)
  55     assert resource_type_id1 is not None and \
  56         resource_type1.resourceTypeId == resource_type_id1
  57
  58
  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(),
  64         ocloud1.oCloudId)
  65     assert resource_pool_id1 is not None and \
  66         resource_pool1.resourcePoolId == resource_pool_id1
  67
  68
  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
  76
  77
  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
  86
  87
  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
  94
  95
  96 def test_view_olcouds(mock_uow):
  97     session, uow = mock_uow
  98
  99     ocloud1_UUID = str(uuid.uuid4)
 100     session.return_value.execute.return_value = [
 101         {"oCloudId": ocloud1_UUID}]
 102
 103     ocloud_list = ocloud_view.oclouds(uow)
 104     assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
 105
 106
 107 def test_view_olcoud_one(mock_uow):
 108     session, uow = mock_uow
 109
 110     ocloud1_UUID = str(uuid.uuid4)
 111     session.return_value.execute.return_value.first.return_value = None
 112
 113     # Query return None
 114     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 115     assert ocloud_res is None
 116
 117     session.return_value.execute.return_value.first.return_value = {
 118         "oCloudId": ocloud1_UUID}
 119
 120     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 121     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
 122
 123
 124 def test_view_resource_types(mock_uow):
 125     session, uow = mock_uow
 126
 127     resource_type_id1 = str(uuid.uuid4())
 128     session.return_value.execute.return_value = [
 129         {"resourceTypeId": resource_type_id1}
 130     ]
 131
 132     resource_type_list = ocloud_view.resource_types(uow)
 133     assert str(resource_type_list[0].get(
 134         "resourceTypeId")) == resource_type_id1
 135
 136
 137 def test_view_resource_type_one(mock_uow):
 138     session, uow = mock_uow
 139
 140     resource_type_id1 = str(uuid.uuid4())
 141     session.return_value.execute.return_value.first.return_value = None
 142
 143     # Query return None
 144     resource_type_res = ocloud_view.resource_type_one(
 145         resource_type_id1, uow)
 146     assert resource_type_res is None
 147
 148     session.return_value.execute.return_value.first.return_value = {
 149         "resourceTypeId": resource_type_id1}
 150
 151     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
 152     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
 153
 154
 155 def test_view_resource_pools(mock_uow):
 156     session, uow = mock_uow
 157
 158     resource_pool_id1 = str(uuid.uuid4())
 159     session.return_value.execute.return_value = [
 160         {"resourcePoolId": resource_pool_id1}
 161     ]
 162
 163     resource_pool_list = ocloud_view.resource_pools(uow)
 164     assert str(resource_pool_list[0].get(
 165         "resourcePoolId")) == resource_pool_id1
 166
 167
 168 def test_view_resource_pool_one(mock_uow):
 169     session, uow = mock_uow
 170
 171     resource_pool_id1 = str(uuid.uuid4())
 172     session.return_value.execute.return_value.first.return_value = None
 173
 174     # Query return None
 175     resource_pool_res = ocloud_view.resource_pool_one(
 176         resource_pool_id1, uow)
 177     assert resource_pool_res is None
 178
 179     session.return_value.execute.return_value.first.return_value = {
 180         "resourcePoolId": resource_pool_id1
 181     }
 182
 183     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 184     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 185
 186
 187 def test_view_resources(mock_uow):
 188     session, uow = mock_uow
 189
 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
 195     }]
 196
 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
 200
 201
 202 def test_view_resource_one(mock_uow):
 203     session, uow = mock_uow
 204
 205     resource_id1 = str(uuid.uuid4())
 206     resource_pool_id1 = str(uuid.uuid4())
 207     session.return_value.execute.return_value.first.return_value = None
 208
 209     # Query return None
 210     resource_res = ocloud_view.resource_one(resource_id1, uow)
 211     assert resource_res is None
 212
 213     session.return_value.execute.return_value.first.return_value = {
 214         "resourceId": resource_id1,
 215         "resourcePoolId": resource_pool_id1
 216     }
 217
 218     resource_res = ocloud_view.resource_one(resource_id1, uow)
 219     assert str(resource_res.get("resourceId")) == resource_id1
 220
 221
 222 def test_view_deployment_managers(mock_uow):
 223     session, uow = mock_uow
 224
 225     deployment_manager_id1 = str(uuid.uuid4())
 226     session.return_value.execute.return_value = [{
 227         "deploymentManagerId": deployment_manager_id1,
 228     }]
 229
 230     deployment_manager_list = ocloud_view.deployment_managers(uow)
 231     assert str(deployment_manager_list[0].get(
 232         "deploymentManagerId")) == deployment_manager_id1
 233
 234
 235 def test_view_deployment_manager_one(mock_uow):
 236     session, uow = mock_uow
 237
 238     deployment_manager_id1 = str(uuid.uuid4())
 239     session.return_value.execute.return_value.first.return_value = None
 240
 241     # Query return None
 242     deployment_manager_res = ocloud_view.deployment_manager_one(
 243         deployment_manager_id1, uow)
 244     assert deployment_manager_res is None
 245
 246     session.return_value.execute.return_value.first.return_value = {
 247         "deploymentManagerId": deployment_manager_id1,
 248     }
 249
 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
 254
 255
 256 def test_view_subscriptions(mock_uow):
 257     session, uow = mock_uow
 258
 259     subscription_id1 = str(uuid.uuid4())
 260     session.return_value.execute.return_value = [{
 261         "subscriptionId": subscription_id1,
 262     }]
 263
 264     subscription_list = ocloud_view.subscriptions(uow)
 265     assert str(subscription_list[0].get(
 266         "subscriptionId")) == subscription_id1
 267
 268
 269 def test_view_subscription_one(mock_uow):
 270     session, uow = mock_uow
 271
 272     subscription_id1 = str(uuid.uuid4())
 273     session.return_value.execute.return_value.first.return_value = None
 274
 275     # Query return None
 276     subscription_res = ocloud_view.subscription_one(
 277         subscription_id1, uow)
 278     assert subscription_res is None
 279
 280     session.return_value.execute.return_value.first.return_value = {
 281         "subscriptionId": subscription_id1,
 282     }
 283
 284     subscription_res = ocloud_view.subscription_one(
 285         subscription_id1, uow)
 286     assert str(subscription_res.get(
 287         "subscriptionId")) == subscription_id1
 288
 289
 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()
 294
 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'
 300
 301         resp = client.get(apibase+"/resourcePools")
 302         assert resp.get_data() == b'[]\n'
 303
 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'
 308
 309         resp = client.get(apibase+"/deploymentManagers")
 310         assert resp.get_data() == b'[]\n'
 311
 312         resp = client.get(apibase+"/subscriptions")
 313         assert resp.get_data() == b'[]\n'
 314
 315
 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()
 320
 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
 326
 327         resource_type_id1 = str(uuid.uuid4())
 328         resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
 329         assert resp.status_code == 404
 330
 331         resource_pool_id1 = str(uuid.uuid4())
 332         resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
 333         assert resp.status_code == 404
 334
 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
 339
 340         deployment_manager_id1 = str(uuid.uuid4())
 341         resp = client.get(apibase+"/deploymentManagers/" +
 342                           deployment_manager_id1)
 343         assert resp.status_code == 404
 344
 345         subscription_id1 = str(uuid.uuid4())
 346         resp = client.get(apibase+"/subscriptions/"+subscription_id1)
 347         assert resp.status_code == 404
 348
 349
 350 def test_flask_post(mock_flask_uow):
 351     session, app = mock_flask_uow
 352     apibase = config.get_o2ims_api_base()
 353
 354     with app.test_client() as client:
 355         session.return_value.execute.return_value = []
 356
 357         sub_callback = 'http://subscription/callback/url'
 358         resp = client.post(apibase+'/subscriptions', json={
 359             'callback': sub_callback,
 360             'consumerSubscriptionId': 'consumerSubId1',
 361             'filter': 'empty'
 362         })
 363         assert resp.status_code == 201
 364         assert 'subscriptionId' in resp.get_json()
 365
 366
 367 def test_flask_delete(mock_flask_uow):
 368     session, app = mock_flask_uow
 369     apibase = config.get_o2ims_api_base()
 370
 371     with app.test_client() as client:
 372         session.return_value.execute.return_value.first.return_value = {}
 373
 374         subscription_id1 = str(uuid.uuid4())
 375         resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
 376         assert resp.status_code == 204
 377
 378
 379 def test_flask_not_allowed(mock_flask_uow):
 380     _, app = mock_flask_uow
 381     apibase = config.get_o2ims_api_base()
 382
 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'
 395
 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'
 406
 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'
 418
 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'
 429
 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'
 441
 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'
 453
 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'
 465
 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'
 476
 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'
 486
 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'