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 o2ims 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_view_olcouds(mock_uow):
  79     session, uow = mock_uow
  80
  81     ocloud1_UUID = str(uuid.uuid4)
  82     session.return_value.execute.return_value = [
  83         {"oCloudId": ocloud1_UUID}]
  84
  85     ocloud_list = ocloud_view.oclouds(uow)
  86     assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
  87
  88
  89 def test_view_olcoud_one(mock_uow):
  90     session, uow = mock_uow
  91
  92     ocloud1_UUID = str(uuid.uuid4)
  93     session.return_value.execute.return_value.first.return_value = None
  94
  95     # Query return None
  96     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
  97     assert ocloud_res is None
  98
  99     session.return_value.execute.return_value.first.return_value = {
 100         "oCloudId": ocloud1_UUID}
 101
 102     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
 103     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
 104
 105
 106 def test_view_resource_types(mock_uow):
 107     session, uow = mock_uow
 108
 109     resource_type_id1 = str(uuid.uuid4())
 110     session.return_value.execute.return_value = [
 111         {"resourceTypeId": resource_type_id1}
 112     ]
 113
 114     resource_type_list = ocloud_view.resource_types(uow)
 115     assert str(resource_type_list[0].get(
 116         "resourceTypeId")) == resource_type_id1
 117
 118
 119 def test_view_resource_type_one(mock_uow):
 120     session, uow = mock_uow
 121
 122     resource_type_id1 = str(uuid.uuid4())
 123     session.return_value.execute.return_value.first.return_value = None
 124
 125     # Query return None
 126     resource_type_res = ocloud_view.resource_type_one(
 127         resource_type_id1, uow)
 128     assert resource_type_res is None
 129
 130     session.return_value.execute.return_value.first.return_value = {
 131         "resourceTypeId": resource_type_id1}
 132
 133     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
 134     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
 135
 136
 137 def test_view_resource_pools(mock_uow):
 138     session, uow = mock_uow
 139
 140     resource_pool_id1 = str(uuid.uuid4())
 141     session.return_value.execute.return_value = [
 142         {"resourcePoolId": resource_pool_id1}
 143     ]
 144
 145     resource_pool_list = ocloud_view.resource_pools(uow)
 146     assert str(resource_pool_list[0].get(
 147         "resourcePoolId")) == resource_pool_id1
 148
 149
 150 def test_view_resource_pool_one(mock_uow):
 151     session, uow = mock_uow
 152
 153     resource_pool_id1 = str(uuid.uuid4())
 154     session.return_value.execute.return_value.first.return_value = None
 155
 156     # Query return None
 157     resource_pool_res = ocloud_view.resource_pool_one(
 158         resource_pool_id1, uow)
 159     assert resource_pool_res is None
 160
 161     session.return_value.execute.return_value.first.return_value = {
 162         "resourcePoolId": resource_pool_id1
 163     }
 164
 165     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 166     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 167
 168
 169 def test_view_resources(mock_uow):
 170     session, uow = mock_uow
 171
 172     resource_id1 = str(uuid.uuid4())
 173     resource_pool_id1 = str(uuid.uuid4())
 174     session.return_value.execute.return_value = [{
 175         "resourceId": resource_id1,
 176         "resourcePoolId": resource_pool_id1
 177     }]
 178
 179     resource_list = ocloud_view.resources(resource_pool_id1, uow)
 180     assert str(resource_list[0].get("resourceId")) == resource_id1
 181     assert str(resource_list[0].get("resourcePoolId")) == resource_pool_id1
 182
 183
 184 def test_view_resource_one(mock_uow):
 185     session, uow = mock_uow
 186
 187     resource_id1 = str(uuid.uuid4())
 188     resource_pool_id1 = str(uuid.uuid4())
 189     session.return_value.execute.return_value.first.return_value = None
 190
 191     # Query return None
 192     resource_res = ocloud_view.resource_one(resource_id1, uow)
 193     assert resource_res is None
 194
 195     session.return_value.execute.return_value.first.return_value = {
 196         "resourceId": resource_id1,
 197         "resourcePoolId": resource_pool_id1
 198     }
 199
 200     resource_res = ocloud_view.resource_one(resource_id1, uow)
 201     assert str(resource_res.get("resourceId")) == resource_id1
 202
 203
 204 def test_flask_get_list(mock_flask_uow):
 205     session, client = mock_flask_uow
 206     session.return_value.execute.return_value = []
 207     apibase = config.get_o2ims_api_base()
 208
 209     # Get list and return empty list
 210     ##########################
 211     resp = client.get(apibase)
 212     assert resp.get_data() == b'[]\n'
 213
 214     resp = client.get(apibase+"/resourceTypes")
 215     assert resp.get_data() == b'[]\n'
 216
 217     resp = client.get(apibase+"/resourcePools")
 218     assert resp.get_data() == b'[]\n'
 219
 220     resource_pool_id1 = str(uuid.uuid4())
 221     resp = client.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources")
 222     assert resp.get_data() == b'[]\n'
 223
 224     resp = client.get(apibase+"/deploymentManagers")
 225     assert resp.get_data() == b'[]\n'
 226
 227
 228 def test_flask_get_one(mock_flask_uow):
 229     session, client = mock_flask_uow
 230     session.return_value.execute.return_value.first.return_value = None
 231     apibase = config.get_o2ims_api_base()
 232
 233     # Get one and return nothing
 234     ###########################
 235     resource_type_id1 = str(uuid.uuid4())
 236     resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
 237     assert resp.get_data() == b''
 238
 239     resource_pool_id1 = str(uuid.uuid4())
 240     resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
 241     assert resp.get_data() == b''
 242
 243     resource_id1 = str(uuid.uuid4())
 244     resp = client.get(apibase+"/resourcePools/" +
 245                       resource_pool_id1+"/resources/"+resource_id1)
 246     assert resp.get_data() == b''
 247
 248     deployment_manager_id1 = str(uuid.uuid4())
 249     resp = client.get(apibase+"/deploymentManagers/"+deployment_manager_id1)
 250     assert resp.get_data() == b''
 251
 252
 253 def test_flask_not_allowed(mock_flask_uow):
 254     _, client = mock_flask_uow
 255     apibase = config.get_o2ims_api_base()
 256
 257     # Testing resource type not support method
 258     ##########################
 259     uri = apibase + "/resourceTypes"
 260     resp = client.post(uri)
 261     assert resp.status == '405 METHOD NOT ALLOWED'
 262     resp = client.put(uri)
 263     assert resp.status == '405 METHOD NOT ALLOWED'
 264     resp = client.patch(uri)
 265     assert resp.status == '405 METHOD NOT ALLOWED'
 266     resp = client.delete(uri)
 267     assert resp.status == '405 METHOD NOT ALLOWED'
 268
 269     resource_type_id1 = str(uuid.uuid4())
 270     uri = apibase + "/resourceTypes/" + resource_type_id1
 271     resp = client.post(uri)
 272     assert resp.status == '405 METHOD NOT ALLOWED'
 273     resp = client.put(uri)
 274     assert resp.status == '405 METHOD NOT ALLOWED'
 275     resp = client.patch(uri)
 276     assert resp.status == '405 METHOD NOT ALLOWED'
 277     resp = client.delete(uri)
 278     assert resp.status == '405 METHOD NOT ALLOWED'
 279
 280     # Testing resource pool not support method
 281     ##########################
 282     uri = apibase + "/resourcePools"
 283     resp = client.post(uri)
 284     assert resp.status == '405 METHOD NOT ALLOWED'
 285     resp = client.put(uri)
 286     assert resp.status == '405 METHOD NOT ALLOWED'
 287     resp = client.patch(uri)
 288     assert resp.status == '405 METHOD NOT ALLOWED'
 289     resp = client.delete(uri)
 290     assert resp.status == '405 METHOD NOT ALLOWED'
 291
 292     resource_pool_id1 = str(uuid.uuid4())
 293     uri = apibase + "/resourcePools/" + resource_pool_id1
 294     resp = client.post(uri)
 295     assert resp.status == '405 METHOD NOT ALLOWED'
 296     resp = client.put(uri)
 297     assert resp.status == '405 METHOD NOT ALLOWED'
 298     resp = client.patch(uri)
 299     assert resp.status == '405 METHOD NOT ALLOWED'
 300     resp = client.delete(uri)
 301     assert resp.status == '405 METHOD NOT ALLOWED'
 302
 303     # Testing resource not support method
 304     ##########################
 305     uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
 306     resp = client.post(uri)
 307     assert resp.status == '405 METHOD NOT ALLOWED'
 308     resp = client.put(uri)
 309     assert resp.status == '405 METHOD NOT ALLOWED'
 310     resp = client.patch(uri)
 311     assert resp.status == '405 METHOD NOT ALLOWED'
 312     resp = client.delete(uri)
 313     assert resp.status == '405 METHOD NOT ALLOWED'
 314
 315     resource_id1 = str(uuid.uuid4())
 316     uri = apibase + "/resourcePools/" + \
 317         resource_pool_id1 + "/resources/" + resource_id1
 318     resp = client.post(uri)
 319     assert resp.status == '405 METHOD NOT ALLOWED'
 320     resp = client.put(uri)
 321     assert resp.status == '405 METHOD NOT ALLOWED'
 322     resp = client.patch(uri)
 323     assert resp.status == '405 METHOD NOT ALLOWED'
 324     resp = client.delete(uri)
 325     assert resp.status == '405 METHOD NOT ALLOWED'
 326
 327     # Testing deployment managers not support method
 328     ##########################
 329     uri = apibase + "/deploymentManagers"
 330     resp = client.post(uri)
 331     assert resp.status == '405 METHOD NOT ALLOWED'
 332     resp = client.put(uri)
 333     assert resp.status == '405 METHOD NOT ALLOWED'
 334     resp = client.patch(uri)
 335     assert resp.status == '405 METHOD NOT ALLOWED'
 336     resp = client.delete(uri)
 337     assert resp.status == '405 METHOD NOT ALLOWED'
 338
 339     deployment_manager_id1 = str(uuid.uuid4())
 340     uri = apibase + "/deploymentManagers/" + deployment_manager_id1
 341     resp = client.post(uri)
 342     assert resp.status == '405 METHOD NOT ALLOWED'
 343     resp = client.put(uri)
 344     assert resp.status == '405 METHOD NOT ALLOWED'
 345     resp = client.patch(uri)
 346     assert resp.status == '405 METHOD NOT ALLOWED'
 347     resp = client.delete(uri)
 348     assert resp.status == '405 METHOD NOT ALLOWED'