Update opMulti operations of the filter
[pti/o2.git] / 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