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 pytest
  16 from o2ims.adapter import ocloud_repository as repository
  17 from o2ims.domain import ocloud
  18 from o2ims import config
  19 import uuid
  20
  21 pytestmark = pytest.mark.usefixtures("mappers")
  22
  23
  24 def setup_ocloud():
  25     ocloudid1 = str(uuid.uuid4())
  26     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1", config.get_api_url(), "ocloud 1 for integration test", 1)
  27     return ocloud1
  28
  29 def setup_ocloud_and_save(sqlite_session_factory):
  30     session = sqlite_session_factory()
  31     repo = repository.OcloudSqlAlchemyRepository(session)
  32     ocloudid1 = str(uuid.uuid4())
  33     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1", config.get_api_url(), "ocloud for integration test", 1)
  34     repo.add(ocloud1)
  35     assert repo.get(ocloudid1) == ocloud1
  36     session.flush()
  37     return ocloud1
  38
  39 def test_add_ocloud(sqlite_session_factory):
  40     session = sqlite_session_factory()
  41     repo = repository.OcloudSqlAlchemyRepository(session)
  42     ocloudid1 = str(uuid.uuid4())
  43     ocloud1 = ocloud.Ocloud(ocloudid1, "ocloud1", config.get_api_url(), "ocloud for integration test", 1)
  44     repo.add(ocloud1)
  45     assert repo.get(ocloudid1) == ocloud1
  46
  47 def test_get_ocloud(sqlite_session_factory):
  48     ocloud1 = setup_ocloud_and_save(sqlite_session_factory)
  49     session = sqlite_session_factory()
  50     repo = repository.OcloudSqlAlchemyRepository(session)
  51     ocloud2 = repo.get(ocloud1.oCloudId)
  52     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
  53
  54 def test_add_ocloud_with_dms(sqlite_session_factory):
  55     session = sqlite_session_factory()
  56     repo = repository.OcloudSqlAlchemyRepository(session)
  57     ocloud1 = setup_ocloud()
  58     dmsid = str(uuid.uuid4())
  59     dms = ocloud.DeploymentManager(
  60         dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
  61     ocloud1.addDeploymentManager(dms)
  62     repo.add(ocloud1)
  63     session.flush()
  64     # seperate session to confirm ocloud is updated into repo
  65     session2 = sqlite_session_factory()
  66     repo2 = repository.OcloudSqlAlchemyRepository(session2)
  67     ocloud2 = repo2.get(ocloud1.oCloudId)
  68     assert ocloud2 is not None
  69     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
  70     assert len(ocloud2.deploymentManagers) == 1
  71
  72
  73 def test_update_ocloud_with_dms(sqlite_session_factory):
  74     session = sqlite_session_factory()
  75     repo = repository.OcloudSqlAlchemyRepository(session)
  76     ocloud1 = setup_ocloud()
  77     repo.add(ocloud1)
  78     session.flush()
  79     dmsid = str(uuid.uuid4())
  80     dms = ocloud.DeploymentManager(
  81         dmsid, "k8s1", ocloud1.oCloudId, config.get_api_url()+"/k8s1")
  82     ocloud1.addDeploymentManager(dms)
  83     repo.update(ocloud1)
  84     # repo.update(ocloud1.oCloudId, {"deploymentManagers": ocloud1.deploymentManagers})
  85     session.flush()
  86
  87     # seperate session to confirm ocloud is updated into repo
  88     session2 = sqlite_session_factory()
  89     repo2 = repository.OcloudSqlAlchemyRepository(session2)
  90     ocloud2 = repo2.get(ocloud1.oCloudId)
  91     assert ocloud2 is not None
  92     assert ocloud2 != ocloud1 and ocloud2.oCloudId == ocloud1.oCloudId
  93     assert len(ocloud2.deploymentManagers) == 1