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
  18 from o2ims.domain import ocloud, subscription_obj, configuration_obj
  19 from o2ims.domain import resource_type as rt
  20 from o2ims.views import ocloud_view
  21 from o2common.config import config
  22
  23
  24 def setup_ocloud():
  25     ocloudid1 = str(uuid.uuid4())
  26     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
  27                             config.get_api_url(), "ocloud for unit test", 1)
  28     return ocloud1
  29
  30
  31 def test_new_ocloud():
  32     ocloudid1 = str(uuid.uuid4())
  33     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1",
  34                             config.get_api_url(), "ocloud for unit test", 1)
  35     assert ocloudid1 is not None and ocloud1.oCloudId == ocloudid1
  36
  37
  38 # def test_add_ocloud_with_dms():
  39 #     ocloud1 = setup_ocloud()
  40 #     dmsid = str(uuid.uuid4())
  41 #     dms = ocloud.DeploymentManager(
  42 #         dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
  43 #     ocloud1.addDeploymentManager(dms)
  44 #     ocloud1.addDeploymentManager(dms)
  45 #     assert len(ocloud1.deploymentManagers) == 1
  46 #     # repo.update(ocloud1.oCloudId, {
  47 #     #             "deploymentManagers": ocloud1.deploymentManagers})
  48
  49
  50 def test_new_resource_type():
  51     ocloud1 = setup_ocloud()
  52     resource_type_id1 = str(uuid.uuid4())
  53     resource_type1 = ocloud.ResourceType(
  54         resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
  55         ocloud1.oCloudId)
  56     assert resource_type_id1 is not None and \
  57         resource_type1.resourceTypeId == resource_type_id1
  58
  59
  60 def test_new_resource_pool():
  61     ocloud1 = setup_ocloud()
  62     resource_pool_id1 = str(uuid.uuid4())
  63     resource_pool1 = ocloud.ResourcePool(
  64         resource_pool_id1, "resourcepool1", config.get_api_url(),
  65         ocloud1.oCloudId)
  66     assert resource_pool_id1 is not None and \
  67         resource_pool1.resourcePoolId == resource_pool_id1
  68
  69
  70 def test_new_resource():
  71     resource_id1 = str(uuid.uuid4())
  72     resource_type_id1 = str(uuid.uuid4())
  73     resource_pool_id1 = str(uuid.uuid4())
  74     resource1 = ocloud.Resource(
  75         resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
  76     assert resource_id1 is not None and resource1.resourceId == resource_id1
  77
  78
  79 def test_new_deployment_manager():
  80     ocloud_id1 = str(uuid.uuid4())
  81     deployment_manager_id1 = str(uuid.uuid4())
  82     deployment_manager1 = ocloud.DeploymentManager(
  83         deployment_manager_id1, "k8s1", ocloud_id1,
  84         config.get_api_url()+"/k8s1")
  85     assert deployment_manager_id1 is not None and deployment_manager1.\
  86         deploymentManagerId == deployment_manager_id1
  87
  88
  89 def test_new_subscription():
  90     subscription_id1 = str(uuid.uuid4())
  91     subscription1 = subscription_obj.Subscription(
  92         subscription_id1, "https://callback/uri/write/here")
  93     assert subscription_id1 is not None and\
  94         subscription1.subscriptionId == subscription_id1
  95
  96
  97 def test_new_configuration():
  98     configuration_id1 = str(uuid.uuid4())
  99     configuration1 = configuration_obj.Configuration(
 100         configuration_id1, "https://callback/uri/write/here",
 101         "SMO")
 102     assert configuration_id1 is not None and\
 103         configuration1.configurationId == configuration_id1
 104
 105
 106 def test_view_olcouds(mock_uow):
 107     session, uow = mock_uow
 108
 109     ocloud1_UUID = str(uuid.uuid4)
 110     ocloud1 = MagicMock()
 111     ocloud1.serialize.return_value = {
 112         'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
 113     session.return_value.query.return_value = [ocloud1]
 114
 115     ocloud_list = ocloud_view.oclouds(uow)
 116     # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
 117     assert len(ocloud_list) == 1
 118
 119
 120 def test_view_olcoud_one(mock_uow):
 121     session, uow = mock_uow
 122
 123     ocloud1_UUID = str(uuid.uuid4)
 124     session.return_value.query.return_value.filter_by.return_value.first.\
 125         return_value.serialize.return_value = None
 126
 127     # Query return None
 128     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 129     assert ocloud_res is None
 130
 131     session.return_value.query.return_value.filter_by.return_value.first.\
 132         return_value.serialize.return_value = {
 133             "oCloudId": ocloud1_UUID}
 134
 135     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 136     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
 137
 138
 139 def test_view_resource_types(mock_uow):
 140     session, uow = mock_uow
 141
 142     resource_type_id1 = str(uuid.uuid4())
 143     restype1 = MagicMock()
 144     restype1.serialize.return_value = {
 145         "resourceTypeId": resource_type_id1}
 146     session.return_value.query.return_value = [restype1]
 147
 148     resource_type_list = ocloud_view.resource_types(uow)
 149     assert str(resource_type_list[0].get(
 150         "resourceTypeId")) == resource_type_id1
 151
 152
 153 def test_view_resource_type_one(mock_uow):
 154     session, uow = mock_uow
 155
 156     resource_type_id1 = str(uuid.uuid4())
 157     session.return_value.query.return_value.filter_by.return_value.first.\
 158         return_value.serialize.return_value = None
 159
 160     # Query return None
 161     resource_type_res = ocloud_view.resource_type_one(
 162         resource_type_id1, uow)
 163     assert resource_type_res is None
 164
 165     session.return_value.query.return_value.filter_by.return_value.first.\
 166         return_value.serialize.return_value = {
 167             "resourceTypeId": resource_type_id1}
 168
 169     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
 170     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
 171
 172
 173 def test_view_resource_pools(mock_uow):
 174     session, uow = mock_uow
 175
 176     resource_pool_id1 = str(uuid.uuid4())
 177     respool1 = MagicMock()
 178     respool1.serialize.return_value = {
 179         "resourcePoolId": resource_pool_id1}
 180     session.return_value.query.return_value = [respool1]
 181
 182     resource_pool_list = ocloud_view.resource_pools(uow)
 183     assert str(resource_pool_list[0].get(
 184         "resourcePoolId")) == resource_pool_id1
 185
 186
 187 def test_view_resource_pool_one(mock_uow):
 188     session, uow = mock_uow
 189
 190     resource_pool_id1 = str(uuid.uuid4())
 191     session.return_value.query.return_value.filter_by.return_value.first.\
 192         return_value.serialize.return_value = None
 193
 194     # Query return None
 195     resource_pool_res = ocloud_view.resource_pool_one(
 196         resource_pool_id1, uow)
 197     assert resource_pool_res is None
 198
 199     session.return_value.query.return_value.filter_by.return_value.first.\
 200         return_value.serialize.return_value = {
 201             "resourcePoolId": resource_pool_id1
 202         }
 203
 204     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 205     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 206
 207
 208 def test_view_resources(mock_uow):
 209     session, uow = mock_uow
 210
 211     resource_id1 = str(uuid.uuid4())
 212     resource_pool_id1 = str(uuid.uuid4())
 213     res1 = MagicMock()
 214     res1.serialize.return_value = {
 215         "resourceId": resource_id1,
 216         "resourcePoolId": resource_pool_id1
 217     }
 218     session.return_value.query.return_value.filter_by.return_value = [res1]
 219
 220     resource_list = ocloud_view.resources(resource_pool_id1, uow)
 221     assert str(resource_list[0].get("resourceId")) == resource_id1
 222     assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
 223
 224
 225 def test_view_resource_one(mock_uow):
 226     session, uow = mock_uow
 227
 228     resource_id1 = str(uuid.uuid4())
 229     resource_pool_id1 = str(uuid.uuid4())
 230     session.return_value.query.return_value.filter_by.return_value.first.\
 231         return_value.serialize.return_value = None
 232
 233     # Query return None
 234     resource_res = ocloud_view.resource_one(resource_id1, uow)
 235     assert resource_res is None
 236
 237     session.return_value.query.return_value.filter_by.return_value.first.\
 238         return_value.serialize.return_value = {
 239             "resourceId": resource_id1,
 240             "resourcePoolId": resource_pool_id1
 241         }
 242
 243     resource_res = ocloud_view.resource_one(resource_id1, uow)
 244     assert str(resource_res.get("resourceId")) == resource_id1
 245
 246
 247 def test_view_deployment_managers(mock_uow):
 248     session, uow = mock_uow
 249
 250     deployment_manager_id1 = str(uuid.uuid4())
 251     dm1 = MagicMock()
 252     dm1.serialize.return_value = {
 253         "deploymentManagerId": deployment_manager_id1,
 254     }
 255     session.return_value.query.return_value = [dm1]
 256
 257     deployment_manager_list = ocloud_view.deployment_managers(uow)
 258     assert str(deployment_manager_list[0].get(
 259         "deploymentManagerId")) == deployment_manager_id1
 260
 261
 262 def test_view_deployment_manager_one(mock_uow):
 263     session, uow = mock_uow
 264
 265     deployment_manager_id1 = str(uuid.uuid4())
 266     session.return_value.query.return_value.filter_by.return_value.first.\
 267         return_value.serialize.return_value = None
 268
 269     # Query return None
 270     deployment_manager_res = ocloud_view.deployment_manager_one(
 271         deployment_manager_id1, uow)
 272     assert deployment_manager_res is None
 273
 274     session.return_value.query.return_value.filter_by.return_value.first.\
 275         return_value.serialize.return_value = {
 276             "deploymentManagerId": deployment_manager_id1,
 277         }
 278
 279     deployment_manager_res = ocloud_view.deployment_manager_one(
 280         deployment_manager_id1, uow)
 281     assert str(deployment_manager_res.get(
 282         "deploymentManagerId")) == deployment_manager_id1
 283
 284
 285 def test_view_subscriptions(mock_uow):
 286     session, uow = mock_uow
 287
 288     subscription_id1 = str(uuid.uuid4())
 289     sub1 = MagicMock()
 290     sub1.serialize.return_value = {
 291         "subscriptionId": subscription_id1,
 292     }
 293     session.return_value.query.return_value = [sub1]
 294
 295     subscription_list = ocloud_view.subscriptions(uow)
 296     assert str(subscription_list[0].get(
 297         "subscriptionId")) == subscription_id1
 298
 299
 300 def test_view_subscription_one(mock_uow):
 301     session, uow = mock_uow
 302
 303     subscription_id1 = str(uuid.uuid4())
 304     session.return_value.query.return_value.filter_by.return_value.first.\
 305         return_value.serialize.return_value = None
 306
 307     # Query return None
 308     subscription_res = ocloud_view.subscription_one(
 309         subscription_id1, uow)
 310     assert subscription_res is None
 311
 312     session.return_value.query.return_value.filter_by.return_value.first.\
 313         return_value.serialize.return_value = {
 314             "subscriptionId": subscription_id1,
 315         }
 316
 317     subscription_res = ocloud_view.subscription_one(
 318         subscription_id1, uow)
 319     assert str(subscription_res.get(
 320         "subscriptionId")) == subscription_id1
 321
 322
 323 def test_flask_get_list(mock_flask_uow):
 324     session, app = mock_flask_uow
 325     session.query.return_value = []
 326     apibase = config.get_o2ims_api_base()
 327
 328     with app.test_client() as client:
 329         # Get list and return empty list
 330         ##########################
 331         resp = client.get(apibase+"/resourceTypes")
 332         assert resp.get_data() == b'[]\n'
 333
 334         resp = client.get(apibase+"/resourcePools")
 335         assert resp.get_data() == b'[]\n'
 336
 337         resource_pool_id1 = str(uuid.uuid4())
 338         resp = client.get(apibase+"/resourcePools/" +
 339                           resource_pool_id1+"/resources")
 340         assert resp.get_data() == b'[]\n'
 341
 342         resp = client.get(apibase+"/deploymentManagers")
 343         assert resp.get_data() == b'[]\n'
 344
 345         resp = client.get(apibase+"/subscriptions")
 346         assert resp.get_data() == b'[]\n'
 347
 348
 349 def test_flask_get_one(mock_flask_uow):
 350     session, app = mock_flask_uow
 351
 352     session.return_value.query.return_value.filter_by.return_value.\
 353         first.return_value = None
 354     apibase = config.get_o2ims_api_base()
 355
 356     with app.test_client() as client:
 357         # Get one and return 404
 358         ###########################
 359         resp = client.get(apibase+"/")
 360         assert resp.status_code == 404
 361
 362         resource_type_id1 = str(uuid.uuid4())
 363         resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
 364         assert resp.status_code == 404
 365
 366         resource_pool_id1 = str(uuid.uuid4())
 367         resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
 368         assert resp.status_code == 404
 369
 370         resource_id1 = str(uuid.uuid4())
 371         resp = client.get(apibase+"/resourcePools/" +
 372                           resource_pool_id1+"/resources/"+resource_id1)
 373         assert resp.status_code == 404
 374
 375         deployment_manager_id1 = str(uuid.uuid4())
 376         resp = client.get(apibase+"/deploymentManagers/" +
 377                           deployment_manager_id1)
 378         assert resp.status_code == 404
 379
 380         subscription_id1 = str(uuid.uuid4())
 381         resp = client.get(apibase+"/subscriptions/"+subscription_id1)
 382         assert resp.status_code == 404
 383
 384
 385 def test_flask_post(mock_flask_uow):
 386     session, app = mock_flask_uow
 387     apibase = config.get_o2ims_api_base()
 388
 389     with app.test_client() as client:
 390         session.return_value.execute.return_value = []
 391
 392         sub_callback = 'http://subscription/callback/url'
 393         resp = client.post(apibase+'/subscriptions', json={
 394             'callback': sub_callback,
 395             'consumerSubscriptionId': 'consumerSubId1',
 396             'filter': 'empty'
 397         })
 398         assert resp.status_code == 201
 399         assert 'subscriptionId' in resp.get_json()
 400
 401
 402 def test_flask_delete(mock_flask_uow):
 403     session, app = mock_flask_uow
 404     apibase = config.get_o2ims_api_base()
 405
 406     with app.test_client() as client:
 407         session.return_value.execute.return_value.first.return_value = {}
 408
 409         subscription_id1 = str(uuid.uuid4())
 410         resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
 411         assert resp.status_code == 204
 412
 413
 414 def test_flask_not_allowed(mock_flask_uow):
 415     _, app = mock_flask_uow
 416     apibase = config.get_o2ims_api_base()
 417
 418     with app.test_client() as client:
 419         # Testing resource type not support method
 420         ##########################
 421         uri = apibase + "/resourceTypes"
 422         resp = client.post(uri)
 423         assert resp.status == '405 METHOD NOT ALLOWED'
 424         resp = client.put(uri)
 425         assert resp.status == '405 METHOD NOT ALLOWED'
 426         resp = client.patch(uri)
 427         assert resp.status == '405 METHOD NOT ALLOWED'
 428         resp = client.delete(uri)
 429         assert resp.status == '405 METHOD NOT ALLOWED'
 430
 431         resource_type_id1 = str(uuid.uuid4())
 432         uri = apibase + "/resourceTypes/" + resource_type_id1
 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         # Testing resource pool not support method
 443         ##########################
 444         uri = apibase + "/resourcePools"
 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_pool_id1 = str(uuid.uuid4())
 455         uri = apibase + "/resourcePools/" + resource_pool_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 not support method
 466         ##########################
 467         uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
 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_id1 = str(uuid.uuid4())
 478         uri = apibase + "/resourcePools/" + \
 479             resource_pool_id1 + "/resources/" + resource_id1
 480         resp = client.post(uri)
 481         assert resp.status == '405 METHOD NOT ALLOWED'
 482         resp = client.put(uri)
 483         assert resp.status == '405 METHOD NOT ALLOWED'
 484         resp = client.patch(uri)
 485         assert resp.status == '405 METHOD NOT ALLOWED'
 486         resp = client.delete(uri)
 487         assert resp.status == '405 METHOD NOT ALLOWED'
 488
 489         # Testing deployment managers not support method
 490         ##########################
 491         uri = apibase + "/deploymentManagers"
 492         resp = client.post(uri)
 493         assert resp.status == '405 METHOD NOT ALLOWED'
 494         resp = client.put(uri)
 495         assert resp.status == '405 METHOD NOT ALLOWED'
 496         resp = client.patch(uri)
 497         assert resp.status == '405 METHOD NOT ALLOWED'
 498         resp = client.delete(uri)
 499         assert resp.status == '405 METHOD NOT ALLOWED'
 500
 501         deployment_manager_id1 = str(uuid.uuid4())
 502         uri = apibase + "/deploymentManagers/" + deployment_manager_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 subscriptions not support method
 513         ##########################
 514         uri = apibase + "/subscriptions"
 515         resp = client.put(uri)
 516         assert resp.status == '405 METHOD NOT ALLOWED'
 517         resp = client.patch(uri)
 518         assert resp.status == '405 METHOD NOT ALLOWED'
 519         resp = client.delete(uri)
 520         assert resp.status == '405 METHOD NOT ALLOWED'
 521
 522         subscription_id1 = str(uuid.uuid4())
 523         uri = apibase + "/subscriptions/" + subscription_id1
 524         resp = client.post(uri)
 525         assert resp.status == '405 METHOD NOT ALLOWED'
 526         resp = client.put(uri)
 527         assert resp.status == '405 METHOD NOT ALLOWED'
 528         resp = client.patch(uri)
 529         assert resp.status == '405 METHOD NOT ALLOWED'