tests/integration/test_ocloud_view.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 import pytest
  17
  18 from o2ims import config
  19 from o2ims.views import ocloud_view
  20 from o2ims.domain import ocloud
  21 from o2ims.domain import resource_type as rt
  22
  23
  24 pytestmark = pytest.mark.usefixtures("mappers")
  25
  26
  27 def setup_ocloud():
  28     ocloudid1 = str(uuid.uuid4())
  29     ocloud1 = ocloud.Ocloud(
  30         ocloudid1, "ocloud1", config.get_api_url(),
  31         "ocloud 1 for integration test", 1)
  32     return ocloud1
  33
  34
  35 def test_view_olcouds(sqlite_uow):
  36     ocloud1 = setup_ocloud()
  37     ocloud1_UUID = ocloud1.oCloudId
  38     with sqlite_uow as uow:
  39         uow.oclouds.add(ocloud1)
  40         uow.commit()
  41
  42     ocloud_list = ocloud_view.oclouds(uow)
  43     assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
  44
  45
  46 def test_view_olcoud_one(sqlite_uow):
  47     ocloud1 = setup_ocloud()
  48     ocloud1_UUID = ocloud1.oCloudId
  49
  50     # Query return None
  51     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, sqlite_uow)
  52     assert ocloud_res is None
  53
  54     with sqlite_uow as uow:
  55         uow.oclouds.add(ocloud1)
  56         # INSERT INTO ocloud (oCloudId, name) VALUES (ocloud1_UUID, 'ocloud1')
  57         uow.commit()
  58     ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
  59     assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
  60
  61
  62 def test_view_resource_types(sqlite_uow):
  63     ocloud1 = setup_ocloud()
  64     resource_type_id1 = str(uuid.uuid4())
  65     resource_type1 = ocloud.ResourceType(
  66         resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
  67         ocloud1.oCloudId)
  68     with sqlite_uow as uow:
  69         uow.resource_types.add(resource_type1)
  70         uow.commit()
  71
  72     resource_type_list = ocloud_view.resource_types(uow)
  73     assert str(resource_type_list[0].get(
  74         "resourceTypeId")) == resource_type_id1
  75
  76
  77 def test_view_resource_type_one(sqlite_uow):
  78     ocloud1 = setup_ocloud()
  79     resource_type_id1 = str(uuid.uuid4())
  80     resource_type1 = ocloud.ResourceType(
  81         resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
  82         ocloud1.oCloudId)
  83
  84     # Query return None
  85     resource_type_res = ocloud_view.resource_type_one(
  86         resource_type_id1, sqlite_uow)
  87     assert resource_type_res is None
  88
  89     with sqlite_uow as uow:
  90         uow.resource_types.add(resource_type1)
  91         uow.commit()
  92     resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
  93     assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
  94
  95
  96 def test_view_resource_pools(sqlite_uow):
  97     ocloud1 = setup_ocloud()
  98     resource_pool_id1 = str(uuid.uuid4())
  99     resource_pool1 = ocloud.ResourcePool(
 100         resource_pool_id1, "resourcepool1", config.get_api_url(),
 101         ocloud1.oCloudId)
 102     with sqlite_uow as uow:
 103         uow.resource_pools.add(resource_pool1)
 104         uow.commit()
 105
 106     resource_pool_list = ocloud_view.resource_pools(uow)
 107     assert str(resource_pool_list[0].get(
 108         "resourcePoolId")) == resource_pool_id1
 109
 110
 111 def test_view_resource_pool_one(sqlite_uow):
 112     ocloud1 = setup_ocloud()
 113     resource_pool_id1 = str(uuid.uuid4())
 114     resource_pool1 = ocloud.ResourcePool(
 115         resource_pool_id1, "resourcepool1", config.get_api_url(),
 116         ocloud1.oCloudId)
 117
 118     # Query return None
 119     resource_pool_res = ocloud_view.resource_pool_one(
 120         resource_pool_id1, sqlite_uow)
 121     assert resource_pool_res is None
 122
 123     with sqlite_uow as uow:
 124         uow.resource_pools.add(resource_pool1)
 125         uow.commit()
 126     resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
 127     assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
 128
 129
 130 def test_view_resources(sqlite_uow):
 131     resource_id1 = str(uuid.uuid4())
 132     resource_type_id1 = str(uuid.uuid4())
 133     resource_pool_id1 = str(uuid.uuid4())
 134     resource1 = ocloud.Resource(
 135         resource_id1, resource_type_id1, resource_pool_id1)
 136     with sqlite_uow as uow:
 137         uow.resources.add(resource1)
 138         uow.commit()
 139
 140     resource_list = ocloud_view.resources(resource_pool_id1, uow)
 141     assert str(resource_list[0].get("resourceId")) == resource_id1
 142
 143
 144 def test_view_resource_one(sqlite_uow):
 145     resource_id1 = str(uuid.uuid4())
 146     resource_type_id1 = str(uuid.uuid4())
 147     resource_pool_id1 = str(uuid.uuid4())
 148     resource1 = ocloud.Resource(
 149         resource_id1, resource_type_id1, resource_pool_id1)
 150
 151     # Query return None
 152     resource_res = ocloud_view.resource_one(resource_id1, sqlite_uow)
 153     assert resource_res is None
 154
 155     with sqlite_uow as uow:
 156         uow.resources.add(resource1)
 157         uow.commit()
 158
 159     resource_res = ocloud_view.resource_one(resource_id1, uow)
 160     assert str(resource_res.get("resourceId")) == resource_id1
 161
 162
 163 def test_view_deployment_managers(sqlite_uow):
 164     ocloud_id1 = str(uuid.uuid4())
 165     deployment_manager_id1 = str(uuid.uuid4())
 166     deployment_manager1 = ocloud.DeploymentManager(
 167         deployment_manager_id1, "k8s1", ocloud_id1,
 168         config.get_api_url()+"/k8s1")
 169     with sqlite_uow as uow:
 170         uow.deployment_managers.add(deployment_manager1)
 171         uow.commit()
 172
 173     deployment_manager_list = ocloud_view.deployment_managers(uow)
 174     assert str(deployment_manager_list[0].get(
 175         "deploymentManagerId")) == deployment_manager_id1
 176
 177
 178 def test_view_deployment_manager_one(sqlite_uow):
 179     ocloud_id1 = str(uuid.uuid4())
 180     deployment_manager_id1 = str(uuid.uuid4())
 181     deployment_manager1 = ocloud.DeploymentManager(
 182         deployment_manager_id1, "k8s1", ocloud_id1,
 183         config.get_api_url()+"/k8s1")
 184
 185     # Query return None
 186     deployment_manager_res = ocloud_view.deployment_manager_one(
 187         deployment_manager_id1, sqlite_uow)
 188     assert deployment_manager_res is None
 189
 190     with sqlite_uow as uow:
 191         uow.deployment_managers.add(deployment_manager1)
 192         uow.commit()
 193
 194     deployment_manager_res = ocloud_view.deployment_manager_one(
 195         deployment_manager_id1, sqlite_uow)
 196     assert str(deployment_manager_res.get(
 197         "deploymentManagerId")) == deployment_manager_id1
 198
 199
 200 def test_view_subscriptions(sqlite_uow):
 201
 202     subscription_id1 = str(uuid.uuid4())
 203     subscription1 = ocloud.Subscription(
 204         subscription_id1, "https://callback/uri/write/here")
 205     with sqlite_uow as uow:
 206         uow.subscriptions.add(subscription1)
 207         uow.commit()
 208
 209     subscription_list = ocloud_view.subscriptions(uow)
 210     assert str(subscription_list[0].get(
 211         "subscriptionId")) == subscription_id1
 212
 213
 214 def test_view_subscription_one(sqlite_uow):
 215
 216     subscription_id1 = str(uuid.uuid4())
 217     subscription1 = ocloud.Subscription(
 218         subscription_id1, "https://callback/uri/write/here")
 219
 220     # Query return None
 221     subscription_res = ocloud_view.subscription_one(
 222         subscription_id1, sqlite_uow)
 223     assert subscription_res is None
 224
 225     with sqlite_uow as uow:
 226         uow.subscriptions.add(subscription1)
 227         uow.commit()
 228
 229     subscription_res = ocloud_view.subscription_one(
 230         subscription_id1, sqlite_uow)
 231     assert str(subscription_res.get(
 232         "subscriptionId")) == subscription_id1
 233
 234
 235 def test_view_subscription_delete(sqlite_uow):
 236
 237     subscription_id1 = str(uuid.uuid4())
 238     subscription1 = ocloud.Subscription(
 239         subscription_id1, "https://callback/uri/write/here")
 240
 241     with sqlite_uow as uow:
 242         uow.subscriptions.add(subscription1)
 243         uow.commit()
 244
 245     subscription_res = ocloud_view.subscription_one(
 246         subscription_id1, sqlite_uow)
 247     assert str(subscription_res.get(
 248         "subscriptionId")) == subscription_id1
 249
 250     with sqlite_uow as uow:
 251         uow.subscriptions.delete(subscription_id1)
 252         uow.commit()
 253
 254     subscription_res = ocloud_view.subscription_one(
 255         subscription_id1, sqlite_uow)
 256     assert subscription_res is None