tests/integration/test_ocloud_route.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 o2common.config import config
  19 from o2ims.domain import ocloud
  20 from o2ims.domain import resource_type as rt
  21
  22
  23 pytestmark = pytest.mark.usefixtures("mappers")
  24
  25
  26 def setup_ocloud():
  27     ocloudid1 = str(uuid.uuid4())
  28     ocloud1 = ocloud.Ocloud(
  29         ocloudid1, "ocloud1", config.get_api_url(),
  30         "ocloud 1 for integration test", 1)
  31     return ocloud1
  32
  33
  34 def test_route_olcouds(sqlite_flask_uow):
  35     uow, app = sqlite_flask_uow
  36
  37     with uow:
  38         ocloud1 = setup_ocloud()
  39         ocloud1_UUID = ocloud1.oCloudId
  40         uow.oclouds.add(ocloud1)
  41         uow.commit()
  42
  43     with app.test_client() as c:
  44         apibase = config.get_o2ims_api_base()
  45         resp = c.get(apibase+"/")
  46         assert resp.status_code == 200
  47         assert ocloud1_UUID.encode() in resp.data
  48
  49
  50 def test_route_resource_types(sqlite_flask_uow):
  51     uow, app = sqlite_flask_uow
  52
  53     with uow:
  54         ocloud1_id = str(uuid.uuid4())
  55         resource_type_id1 = str(uuid.uuid4())
  56         resource_type1 = ocloud.ResourceType(
  57             resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
  58             ocloud1_id)
  59         uow.resource_types.add(resource_type1)
  60         uow.commit()
  61
  62     with app.test_client() as c:
  63         apibase = config.get_o2ims_api_base()
  64         resp = c.get(apibase+"/resourceTypes")
  65         assert resp.status_code == 200
  66         assert resource_type_id1.encode() in resp.data
  67
  68         resp = c.get(apibase+'/resourceTypes/'+resource_type_id1)
  69         assert resp.status_code == 200
  70         json_data = resp.get_json()
  71         assert resource_type_id1 in json_data['resourceTypeId']
  72
  73
  74 def test_route_resource_pools(sqlite_flask_uow):
  75     uow, app = sqlite_flask_uow
  76
  77     with uow:
  78         ocloud1_id = str(uuid.uuid4())
  79         resource_pool_id1 = str(uuid.uuid4())
  80         resource_pool1 = ocloud.ResourcePool(
  81             resource_pool_id1, "resourcepool1", config.get_api_url(),
  82             ocloud1_id)
  83         uow.resource_pools.add(resource_pool1)
  84         uow.commit()
  85
  86     with app.test_client() as c:
  87         apibase = config.get_o2ims_api_base()
  88         resp = c.get(apibase+"/resourcePools")
  89         assert resp.status_code == 200
  90         assert resource_pool_id1.encode() in resp.data
  91
  92         resp = c.get(apibase+'/resourcePools/'+resource_pool_id1)
  93         assert resp.status_code == 200
  94         json_data = resp.get_json()
  95         assert resource_pool_id1 in json_data['resourcePoolId']
  96
  97
  98 def test_route_resources(sqlite_flask_uow):
  99     uow, app = sqlite_flask_uow
 100
 101     with uow:
 102         resource_id1 = str(uuid.uuid4())
 103         resource_type_id1 = str(uuid.uuid4())
 104         resource_pool_id1 = str(uuid.uuid4())
 105         resource1 = ocloud.Resource(
 106             resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
 107         uow.resources.add(resource1)
 108         uow.commit()
 109
 110     with app.test_client() as c:
 111         apibase = config.get_o2ims_api_base()
 112         resp = c.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources")
 113         assert resp.status_code == 200
 114         assert resource_id1.encode() in resp.data
 115
 116         resp = c.get(apibase+"/resourcePools/"+resource_pool_id1 +
 117                      "/resources/" + resource_id1)
 118         assert resp.status_code == 200
 119         json_data = resp.get_json()
 120         assert resource_pool_id1 in json_data['resourcePoolId']
 121         assert resource_type_id1 in json_data['resourceTypeId']
 122         assert resource_id1 in json_data['resourceId']
 123
 124
 125 def test_route_deployment_managers(sqlite_flask_uow):
 126     uow, app = sqlite_flask_uow
 127
 128     with uow:
 129         ocloud_id1 = str(uuid.uuid4())
 130         deployment_manager_id1 = str(uuid.uuid4())
 131         deployment_manager1 = ocloud.DeploymentManager(
 132             deployment_manager_id1, "k8s1", ocloud_id1,
 133             config.get_api_url()+"/k8s1")
 134         uow.deployment_managers.add(deployment_manager1)
 135         uow.commit()
 136
 137     with app.test_client() as c:
 138         apibase = config.get_o2ims_api_base()
 139         resp = c.get(apibase+"/deploymentManagers")
 140         assert resp.status_code == 200
 141         assert deployment_manager_id1.encode() in resp.data
 142
 143         resp = c.get(apibase+'/deploymentManagers/'+deployment_manager_id1)
 144         assert resp.status_code == 200
 145         json_data = resp.get_json()
 146         assert deployment_manager_id1 in json_data['deploymentManagerId']
 147
 148
 149 def test_route_subscriptions(sqlite_flask_uow):
 150     _, app = sqlite_flask_uow
 151
 152     with app.test_client() as c:
 153         apibase = config.get_o2ims_api_base()
 154
 155         sub_callback = 'http://subscription/callback/url'
 156         resp = c.post(apibase+'/subscriptions', json={
 157             'callback': sub_callback,
 158             'consumerSubscriptionId': 'consumerSubId1',
 159             'filter': 'empty'
 160         })
 161         assert resp.status_code == 201
 162         json_data = resp.get_json()
 163         assert 'subscriptionId' in json_data
 164         subscriptionId1 = json_data['subscriptionId']
 165
 166         resp = c.get(apibase+'/subscriptions')
 167         assert resp.status_code == 200
 168         json_data = resp.get_json()
 169         assert 1 == len(json_data)
 170
 171         resp = c.get(apibase+'/subscriptions/'+subscriptionId1)
 172         assert resp.status_code == 200
 173         json_data = resp.get_json()
 174         assert sub_callback in json_data['callback']
 175
 176         resp = c.delete(apibase+'/subscriptions/'+subscriptionId1)
 177         assert resp.status_code == 204
 178
 179         resp = c.get(apibase+'/subscriptions')
 180         assert resp.status_code == 200
 181         json_data = resp.get_json()
 182         assert 0 == len(json_data)