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
  21 from o2ims 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)
 141     repo.add(resource1)
 142     assert repo.get(resource_id1) == resource1