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
  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_view_olcouds(mock_uow):
  98     session, uow = mock_uow
  99
 100     ocloud1_UUID = str(uuid.uuid4)
 101     ocloud1 = MagicMock()
 102     ocloud1.serialize.return_value = {
 103         'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
 104     session.return_value.query.return_value = [ocloud1]
 105
 106     ocloud_list = ocloud_view.oclouds(uow)
 107     # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
 108     assert len(ocloud_list) == 1
 109
 110
 111 def test_view_olcoud_one(mock_uow):
 112     session, uow = mock_uow
 113
 114     ocloud1_UUID = str(uuid.uuid4)
 115     session.return_value.query.return_value.filter_by.return_value.first.\
 116         return_value.serialize.return_value = None
 117
 118     # Query return None
 119     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 120     assert ocloud_res is None
 121
 122     session.return_value.query.return_value.filter_by.return_value.first.\
 123         return_value.serialize.return_value = {
 124             "oCloudId": ocloud1_UUID}
 125
 126     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 127     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
 128
 129
 130 def test_view_resource_types(mock_uow):
 131     session, uow = mock_uow
 132
 133     resource_type_id1 = str(uuid.uuid4())
 134     restype1 = MagicMock()
 135     restype1.serialize.return_value = {
 136         "resourceTypeId": resource_type_id1}
 137     session.return_value.query.return_value = [restype1]
 138
 139     resource_type_list = ocloud_view.resource_types(uow)
 140     assert str(resource_type_list[0].get(
 141         "resourceTypeId")) == resource_type_id1
 142
 143
 144 def test_view_resource_type_one(mock_uow):
 145     session, uow = mock_uow
 146
 147     resource_type_id1 = str(uuid.uuid4())
 148     session.return_value.query.return_value.filter_by.return_value.first.\
 149         return_value.serialize.return_value = None
 150
 151     # Query return None
 152     resource_type_res = ocloud_view.resource_type_one(
 153         resource_type_id1, uow)
 154     assert resource_type_res is None
 155
 156     session.return_value.query.return_value.filter_by.return_value.first.\
 157         return_value.serialize.return_value = {
 158             "resourceTypeId": resource_type_id1}
 159
 160     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
 161     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
 162
 163
 164 def test_view_resource_pools(mock_uow):
 165     session, uow = mock_uow
 166
 167     resource_pool_id1 = str(uuid.uuid4())
 168     respool1 = MagicMock()
 169     respool1.serialize.return_value = {
 170         "resourcePoolId": resource_pool_id1}
 171     session.return_value.query.return_value = [respool1]
 172
 173     resource_pool_list = ocloud_view.resource_pools(uow)
 174     assert str(resource_pool_list[0].get(
 175         "resourcePoolId")) == resource_pool_id1
 176
 177
 178 def test_view_resource_pool_one(mock_uow):
 179     session, uow = mock_uow
 180
 181     resource_pool_id1 = str(uuid.uuid4())
 182     session.return_value.query.return_value.filter_by.return_value.first.\
 183         return_value.serialize.return_value = None
 184
 185     # Query return None
 186     resource_pool_res = ocloud_view.resource_pool_one(
 187         resource_pool_id1, uow)
 188     assert resource_pool_res is None
 189
 190     session.return_value.query.return_value.filter_by.return_value.first.\
 191         return_value.serialize.return_value = {
 192             "resourcePoolId": resource_pool_id1
 193         }
 194
 195     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 196     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 197
 198
 199 def test_view_resources(mock_uow):
 200     session, uow = mock_uow
 201
 202     resource_id1 = str(uuid.uuid4())
 203     resource_pool_id1 = str(uuid.uuid4())
 204     res1 = MagicMock()
 205     res1.serialize.return_value = {
 206         "resourceId": resource_id1,
 207         "resourcePoolId": resource_pool_id1
 208     }
 209     session.return_value.query.return_value.filter_by.return_value = [res1]
 210
 211     resource_list = ocloud_view.resources(resource_pool_id1, uow)
 212     assert str(resource_list[0].get("resourceId")) == resource_id1
 213     assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
 214
 215
 216 def test_view_resource_one(mock_uow):
 217     session, uow = mock_uow
 218
 219     resource_id1 = str(uuid.uuid4())
 220     resource_pool_id1 = str(uuid.uuid4())
 221     session.return_value.query.return_value.filter_by.return_value.first.\
 222         return_value.serialize.return_value = None
 223
 224     # Query return None
 225     resource_res = ocloud_view.resource_one(resource_id1, uow)
 226     assert resource_res is None
 227
 228     session.return_value.query.return_value.filter_by.return_value.first.\
 229         return_value.serialize.return_value = {
 230             "resourceId": resource_id1,
 231             "resourcePoolId": resource_pool_id1
 232         }
 233
 234     resource_res = ocloud_view.resource_one(resource_id1, uow)
 235     assert str(resource_res.get("resourceId")) == resource_id1
 236
 237
 238 def test_view_deployment_managers(mock_uow):
 239     session, uow = mock_uow
 240
 241     deployment_manager_id1 = str(uuid.uuid4())
 242     dm1 = MagicMock()
 243     dm1.serialize.return_value = {
 244         "deploymentManagerId": deployment_manager_id1,
 245     }
 246     session.return_value.query.return_value = [dm1]
 247
 248     deployment_manager_list = ocloud_view.deployment_managers(uow)
 249     assert str(deployment_manager_list[0].get(
 250         "deploymentManagerId")) == deployment_manager_id1
 251
 252
 253 def test_view_deployment_manager_one(mock_uow):
 254     session, uow = mock_uow
 255
 256     deployment_manager_id1 = str(uuid.uuid4())
 257     session.return_value.query.return_value.filter_by.return_value.first.\
 258         return_value.serialize.return_value = None
 259
 260     # Query return None
 261     deployment_manager_res = ocloud_view.deployment_manager_one(
 262         deployment_manager_id1, uow)
 263     assert deployment_manager_res is None
 264
 265     session.return_value.query.return_value.filter_by.return_value.first.\
 266         return_value.serialize.return_value = {
 267             "deploymentManagerId": deployment_manager_id1,
 268         }
 269
 270     deployment_manager_res = ocloud_view.deployment_manager_one(
 271         deployment_manager_id1, uow)
 272     assert str(deployment_manager_res.get(
 273         "deploymentManagerId")) == deployment_manager_id1
 274
 275
 276 def test_view_subscriptions(mock_uow):
 277     session, uow = mock_uow
 278
 279     subscription_id1 = str(uuid.uuid4())
 280     sub1 = MagicMock()
 281     sub1.serialize.return_value = {
 282         "subscriptionId": subscription_id1,
 283     }
 284     session.return_value.query.return_value = [sub1]
 285
 286     subscription_list = ocloud_view.subscriptions(uow)
 287     assert str(subscription_list[0].get(
 288         "subscriptionId")) == subscription_id1
 289
 290
 291 def test_view_subscription_one(mock_uow):
 292     session, uow = mock_uow
 293
 294     subscription_id1 = str(uuid.uuid4())
 295     session.return_value.query.return_value.filter_by.return_value.first.\
 296         return_value.serialize.return_value = None
 297
 298     # Query return None
 299     subscription_res = ocloud_view.subscription_one(
 300         subscription_id1, uow)
 301     assert subscription_res is None
 302
 303     session.return_value.query.return_value.filter_by.return_value.first.\
 304         return_value.serialize.return_value = {
 305             "subscriptionId": subscription_id1,
 306         }
 307
 308     subscription_res = ocloud_view.subscription_one(
 309         subscription_id1, uow)
 310     assert str(subscription_res.get(
 311         "subscriptionId")) == subscription_id1
 312
 313
 314 def test_flask_get_list(mock_flask_uow):
 315     session, app = mock_flask_uow
 316     session.query.return_value = []
 317     apibase = config.get_o2ims_api_base()
 318
 319     with app.test_client() as client:
 320         # Get list and return empty list
 321         ##########################
 322         resp = client.get(apibase+"/resourceTypes")
 323         assert resp.get_data() == b'[]\n'
 324
 325         resp = client.get(apibase+"/resourcePools")
 326         assert resp.get_data() == b'[]\n'
 327
 328         resource_pool_id1 = str(uuid.uuid4())
 329         resp = client.get(apibase+"/resourcePools/" +
 330                           resource_pool_id1+"/resources")
 331         assert resp.get_data() == b'[]\n'
 332
 333         resp = client.get(apibase+"/deploymentManagers")
 334         assert resp.get_data() == b'[]\n'
 335
 336         resp = client.get(apibase+"/subscriptions")
 337         assert resp.get_data() == b'[]\n'
 338
 339
 340 def test_flask_get_one(mock_flask_uow):
 341     session, app = mock_flask_uow
 342
 343     session.return_value.query.return_value.filter_by.return_value.\
 344         first.return_value = None
 345     apibase = config.get_o2ims_api_base()
 346
 347     with app.test_client() as client:
 348         # Get one and return 404
 349         ###########################
 350         resp = client.get(apibase+"/")
 351         assert resp.status_code == 404
 352
 353         resource_type_id1 = str(uuid.uuid4())
 354         resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
 355         assert resp.status_code == 404
 356
 357         resource_pool_id1 = str(uuid.uuid4())
 358         resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
 359         assert resp.status_code == 404
 360
 361         resource_id1 = str(uuid.uuid4())
 362         resp = client.get(apibase+"/resourcePools/" +
 363                           resource_pool_id1+"/resources/"+resource_id1)
 364         assert resp.status_code == 404
 365
 366         deployment_manager_id1 = str(uuid.uuid4())
 367         resp = client.get(apibase+"/deploymentManagers/" +
 368                           deployment_manager_id1)
 369         assert resp.status_code == 404
 370
 371         subscription_id1 = str(uuid.uuid4())
 372         resp = client.get(apibase+"/subscriptions/"+subscription_id1)
 373         assert resp.status_code == 404
 374
 375
 376 def test_flask_post(mock_flask_uow):
 377     session, app = mock_flask_uow
 378     apibase = config.get_o2ims_api_base()
 379
 380     with app.test_client() as client:
 381         session.return_value.execute.return_value = []
 382
 383         sub_callback = 'http://subscription/callback/url'
 384         resp = client.post(apibase+'/subscriptions', json={
 385             'callback': sub_callback,
 386             'consumerSubscriptionId': 'consumerSubId1',
 387             'filter': 'empty'
 388         })
 389         assert resp.status_code == 201
 390         assert 'subscriptionId' in resp.get_json()
 391
 392
 393 def test_flask_delete(mock_flask_uow):
 394     session, app = mock_flask_uow
 395     apibase = config.get_o2ims_api_base()
 396
 397     with app.test_client() as client:
 398         session.return_value.execute.return_value.first.return_value = {}
 399
 400         subscription_id1 = str(uuid.uuid4())
 401         resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
 402         assert resp.status_code == 204
 403
 404
 405 def test_flask_not_allowed(mock_flask_uow):
 406     _, app = mock_flask_uow
 407     apibase = config.get_o2ims_api_base()
 408
 409     with app.test_client() as client:
 410         # Testing resource type not support method
 411         ##########################
 412         uri = apibase + "/resourceTypes"
 413         resp = client.post(uri)
 414         assert resp.status == '405 METHOD NOT ALLOWED'
 415         resp = client.put(uri)
 416         assert resp.status == '405 METHOD NOT ALLOWED'
 417         resp = client.patch(uri)
 418         assert resp.status == '405 METHOD NOT ALLOWED'
 419         resp = client.delete(uri)
 420         assert resp.status == '405 METHOD NOT ALLOWED'
 421
 422         resource_type_id1 = str(uuid.uuid4())
 423         uri = apibase + "/resourceTypes/" + resource_type_id1
 424         resp = client.post(uri)
 425         assert resp.status == '405 METHOD NOT ALLOWED'
 426         resp = client.put(uri)
 427         assert resp.status == '405 METHOD NOT ALLOWED'
 428         resp = client.patch(uri)
 429         assert resp.status == '405 METHOD NOT ALLOWED'
 430         resp = client.delete(uri)
 431         assert resp.status == '405 METHOD NOT ALLOWED'
 432
 433         # Testing resource pool not support method
 434         ##########################
 435         uri = apibase + "/resourcePools"
 436         resp = client.post(uri)
 437         assert resp.status == '405 METHOD NOT ALLOWED'
 438         resp = client.put(uri)
 439         assert resp.status == '405 METHOD NOT ALLOWED'
 440         resp = client.patch(uri)
 441         assert resp.status == '405 METHOD NOT ALLOWED'
 442         resp = client.delete(uri)
 443         assert resp.status == '405 METHOD NOT ALLOWED'
 444
 445         resource_pool_id1 = str(uuid.uuid4())
 446         uri = apibase + "/resourcePools/" + resource_pool_id1
 447         resp = client.post(uri)
 448         assert resp.status == '405 METHOD NOT ALLOWED'
 449         resp = client.put(uri)
 450         assert resp.status == '405 METHOD NOT ALLOWED'
 451         resp = client.patch(uri)
 452         assert resp.status == '405 METHOD NOT ALLOWED'
 453         resp = client.delete(uri)
 454         assert resp.status == '405 METHOD NOT ALLOWED'
 455
 456         # Testing resource not support method
 457         ##########################
 458         uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
 459         resp = client.post(uri)
 460         assert resp.status == '405 METHOD NOT ALLOWED'
 461         resp = client.put(uri)
 462         assert resp.status == '405 METHOD NOT ALLOWED'
 463         resp = client.patch(uri)
 464         assert resp.status == '405 METHOD NOT ALLOWED'
 465         resp = client.delete(uri)
 466         assert resp.status == '405 METHOD NOT ALLOWED'
 467
 468         resource_id1 = str(uuid.uuid4())
 469         uri = apibase + "/resourcePools/" + \
 470             resource_pool_id1 + "/resources/" + resource_id1
 471         resp = client.post(uri)
 472         assert resp.status == '405 METHOD NOT ALLOWED'
 473         resp = client.put(uri)
 474         assert resp.status == '405 METHOD NOT ALLOWED'
 475         resp = client.patch(uri)
 476         assert resp.status == '405 METHOD NOT ALLOWED'
 477         resp = client.delete(uri)
 478         assert resp.status == '405 METHOD NOT ALLOWED'
 479
 480         # Testing deployment managers not support method
 481         ##########################
 482         uri = apibase + "/deploymentManagers"
 483         resp = client.post(uri)
 484         assert resp.status == '405 METHOD NOT ALLOWED'
 485         resp = client.put(uri)
 486         assert resp.status == '405 METHOD NOT ALLOWED'
 487         resp = client.patch(uri)
 488         assert resp.status == '405 METHOD NOT ALLOWED'
 489         resp = client.delete(uri)
 490         assert resp.status == '405 METHOD NOT ALLOWED'
 491
 492         deployment_manager_id1 = str(uuid.uuid4())
 493         uri = apibase + "/deploymentManagers/" + deployment_manager_id1
 494         resp = client.post(uri)
 495         assert resp.status == '405 METHOD NOT ALLOWED'
 496         resp = client.put(uri)
 497         assert resp.status == '405 METHOD NOT ALLOWED'
 498         resp = client.patch(uri)
 499         assert resp.status == '405 METHOD NOT ALLOWED'
 500         resp = client.delete(uri)
 501         assert resp.status == '405 METHOD NOT ALLOWED'
 502
 503         # Testing subscriptions not support method
 504         ##########################
 505         uri = apibase + "/subscriptions"
 506         resp = client.put(uri)
 507         assert resp.status == '405 METHOD NOT ALLOWED'
 508         resp = client.patch(uri)
 509         assert resp.status == '405 METHOD NOT ALLOWED'
 510         resp = client.delete(uri)
 511         assert resp.status == '405 METHOD NOT ALLOWED'
 512
 513         subscription_id1 = str(uuid.uuid4())
 514         uri = apibase + "/subscriptions/" + subscription_id1
 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'