tests/integration/test_ocloud_repository.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.domain import resource_type as rt
  19 from o2ims.adapter import ocloud_repository as repository
  20 from o2ims.domain import ocloud, subscription_obj
  21 from o2common.config import config
  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 setup_ocloud_and_save(sqlite_session_factory):
  35     session = sqlite_session_factory()
  36     repo = repository.OcloudSqlAlchemyRepository(session)
  37     ocloudid1 = str(uuid.uuid4())
  38     ocloud1 = ocloud.Ocloud(
  39         ocloudid1, "ocloud1", config.get_api_url(),
  40         "ocloud for integration test", 1)
  41     repo.add(ocloud1)
  42     assert repo.get(ocloudid1) == ocloud1
  43     session.flush()
  44     return ocloud1
  45
  46
  47 def test_add_ocloud(sqlite_session_factory):
  48     session = sqlite_session_factory()
  49     repo = repository.OcloudSqlAlchemyRepository(session)
  50     ocloudid1 = str(uuid.uuid4())
  51     ocloud1 = ocloud.Ocloud(
  52         ocloudid1, "ocloud1", config.get_api_url(),
  53         "ocloud for integration test", 1)
  54     repo.add(ocloud1)
  55     assert repo.get(ocloudid1) == ocloud1
  56
  57
  58 def test_get_ocloud(sqlite_session_factory):
  59     ocloud1 = setup_ocloud_and_save(sqlite_session_factory)
  60     session = sqlite_session_factory()
  61     repo = repository.OcloudSqlAlchemyRepository(session)
  62     ocloud2 = repo.get(ocloud1.oCloudId)
  63     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
  64
  65
  66 # def test_add_ocloud_with_dms(sqlite_session_factory):
  67 #     session = sqlite_session_factory()
  68 #     repo = repository.OcloudSqlAlchemyRepository(session)
  69 #     ocloud1 = setup_ocloud()
  70 #     dmsid = str(uuid.uuid4())
  71 #     dms = ocloud.DeploymentManager(
  72 #         dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
  73 #     ocloud1.addDeploymentManager(dms)
  74 #     repo.add(ocloud1)
  75 #     session.flush()
  76 #     # seperate session to confirm ocloud is updated into repo
  77 #     session2 = sqlite_session_factory()
  78 #     repo2 = repository.OcloudSqlAlchemyRepository(session2)
  79 #     ocloud2 = repo2.get(ocloud1.oCloudId)
  80 #     assert ocloud2 is not None
  81 #     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
  82 #     assert len(ocloud2.deploymentManagers) == 1
  83
  84
  85 # def test_update_ocloud_with_dms(sqlite_session_factory):
  86 #     session = sqlite_session_factory()
  87 #     repo = repository.OcloudSqlAlchemyRepository(session)
  88 #     ocloud1 = setup_ocloud()
  89 #     repo.add(ocloud1)
  90 #     session.flush()
  91 #     dmsid = str(uuid.uuid4())
  92 #     dms = ocloud.DeploymentManager(
  93 #         dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
  94 #     ocloud1.addDeploymentManager(dms)
  95 #     repo.update(ocloud1)
  96 #     # repo.update(ocloud1.oCloudId, {"deploymentManagers":
  97 #     # ocloud1.deploymentManagers})
  98 #     session.flush()
  99
 100 #     # seperate session to confirm ocloud is updated into repo
 101 #     session2 = sqlite_session_factory()
 102 #     repo2 = repository.OcloudSqlAlchemyRepository(session2)
 103 #     ocloud2 = repo2.get(ocloud1.oCloudId)
 104 #     assert ocloud2 is not None
 105 #     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
 106 #     assert len(ocloud2.deploymentManagers) == 1
 107
 108
 109 def test_add_resource_type(sqlite_session_factory):
 110     session = sqlite_session_factory()
 111     repo = repository.ResouceTypeSqlAlchemyRepository(session)
 112     ocloud1_id = str(uuid.uuid4())
 113     resource_type_id1 = str(uuid.uuid4())
 114     resource_type1 = ocloud.ResourceType(
 115         resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
 116         ocloud1_id)
 117     repo.add(resource_type1)
 118     assert repo.get(resource_type_id1) == resource_type1
 119
 120
 121 def test_add_resource_pool(sqlite_session_factory):
 122     session = sqlite_session_factory()
 123     repo = repository.ResourcePoolSqlAlchemyRepository(session)
 124     ocloud1_id = str(uuid.uuid4())
 125     resource_pool_id1 = str(uuid.uuid4())
 126     resource_pool1 = ocloud.ResourcePool(
 127         resource_pool_id1, "resourcepool1", config.get_api_url(),
 128         ocloud1_id)
 129     repo.add(resource_pool1)
 130     assert repo.get(resource_pool_id1) == resource_pool1
 131
 132
 133 def test_add_resource(sqlite_session_factory):
 134     session = sqlite_session_factory()
 135     repo = repository.ResourceSqlAlchemyRepository(session)
 136     resource_id1 = str(uuid.uuid4())
 137     resource_type_id1 = str(uuid.uuid4())
 138     resource_pool_id1 = str(uuid.uuid4())
 139     resource1 = ocloud.Resource(
 140         resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
 141     repo.add(resource1)
 142     assert repo.get(resource_id1) == resource1
 143
 144
 145 def test_add_deployment_manager(sqlite_session_factory):
 146     session = sqlite_session_factory()
 147     repo = repository.DeploymentManagerSqlAlchemyRepository(session)
 148     ocloud_id1 = str(uuid.uuid4())
 149     deployment_manager_id1 = str(uuid.uuid4())
 150     deployment_manager1 = ocloud.DeploymentManager(
 151         deployment_manager_id1, "k8s1", ocloud_id1,
 152         config.get_api_url()+"/k8s1")
 153     repo.add(deployment_manager1)
 154     assert repo.get(deployment_manager_id1) == deployment_manager1
 155
 156
 157 def test_add_subscription(sqlite_session_factory):
 158     session = sqlite_session_factory()
 159     repo = repository.SubscriptionSqlAlchemyRepository(session)
 160     subscription_id1 = str(uuid.uuid4())
 161     subscription1 = subscription_obj.Subscription(
 162         subscription_id1, "https://callback/uri/write/here")
 163     repo.add(subscription1)
 164     assert repo.get(subscription_id1) == subscription1