1 # Copyright (C) 2021 Wind River Systems, Inc.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from o2ims import config
19 from o2ims.views import ocloud_view
20 from o2ims.domain import ocloud
21 from o2ims.domain import resource_type as rt
24 pytestmark = pytest.mark.usefixtures("mappers")
28 ocloudid1 = str(uuid.uuid4())
29 ocloud1 = ocloud.Ocloud(
30 ocloudid1, "ocloud1", config.get_api_url(),
31 "ocloud 1 for integration test", 1)
35 def test_view_olcouds(sqlite_uow):
36 ocloud1 = setup_ocloud()
37 ocloud1_UUID = ocloud1.oCloudId
38 with sqlite_uow as uow:
39 uow.oclouds.add(ocloud1)
42 ocloud_list = ocloud_view.oclouds(uow)
43 assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
46 def test_view_olcoud_one(sqlite_uow):
47 ocloud1 = setup_ocloud()
48 ocloud1_UUID = ocloud1.oCloudId
51 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, sqlite_uow)
52 assert ocloud_res is None
54 with sqlite_uow as uow:
55 uow.oclouds.add(ocloud1)
56 # INSERT INTO ocloud (oCloudId, name) VALUES (ocloud1_UUID, 'ocloud1')
58 ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
59 assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
62 def test_view_resource_types(sqlite_uow):
63 ocloud1 = setup_ocloud()
64 resource_type_id1 = str(uuid.uuid4())
65 resource_type1 = ocloud.ResourceType(
66 resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
68 with sqlite_uow as uow:
69 # uow.session.execute()
70 uow.oclouds.add(resource_type1)
73 resource_type_list = ocloud_view.resource_types(uow)
74 assert str(resource_type_list[0].get(
75 "resourceTypeId")) == resource_type_id1
78 def test_view_resource_type_one(sqlite_uow):
79 ocloud1 = setup_ocloud()
80 resource_type_id1 = str(uuid.uuid4())
81 resource_type1 = ocloud.ResourceType(
82 resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
86 resource_type_res = ocloud_view.resource_type_one(
87 resource_type_id1, sqlite_uow)
88 assert resource_type_res is None
90 with sqlite_uow as uow:
91 uow.oclouds.add(resource_type1)
93 resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
94 assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
97 def test_view_resource_pools(sqlite_uow):
98 ocloud1 = setup_ocloud()
99 resource_pool_id1 = str(uuid.uuid4())
100 resource_pool1 = ocloud.ResourcePool(
101 resource_pool_id1, "resourcepool1", config.get_api_url(),
103 with sqlite_uow as uow:
104 uow.oclouds.add(resource_pool1)
107 resource_pool_list = ocloud_view.resource_pools(uow)
108 assert str(resource_pool_list[0].get(
109 "resourcePoolId")) == resource_pool_id1
112 def test_view_resource_pool_one(sqlite_uow):
113 ocloud1 = setup_ocloud()
114 resource_pool_id1 = str(uuid.uuid4())
115 resource_pool1 = ocloud.ResourcePool(
116 resource_pool_id1, "resourcepool1", config.get_api_url(),
120 resource_pool_res = ocloud_view.resource_pool_one(
121 resource_pool_id1, sqlite_uow)
122 assert resource_pool_res is None
124 with sqlite_uow as uow:
125 uow.oclouds.add(resource_pool1)
127 resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
128 assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
131 def test_view_resources(sqlite_uow):
132 resource_id1 = str(uuid.uuid4())
133 resource_type_id1 = str(uuid.uuid4())
134 resource_pool_id1 = str(uuid.uuid4())
135 resource1 = ocloud.Resource(
136 resource_id1, resource_type_id1, resource_pool_id1)
137 with sqlite_uow as uow:
138 uow.oclouds.add(resource1)
141 resource_list = ocloud_view.resources(resource_pool_id1, uow)
142 assert str(resource_list[0].get("resourceId")) == resource_id1
145 def test_view_resource_one(sqlite_uow):
146 resource_id1 = str(uuid.uuid4())
147 resource_type_id1 = str(uuid.uuid4())
148 resource_pool_id1 = str(uuid.uuid4())
149 resource1 = ocloud.Resource(
150 resource_id1, resource_type_id1, resource_pool_id1)
153 resource_res = ocloud_view.resource_one(resource_id1, sqlite_uow)
154 assert resource_res is None
156 with sqlite_uow as uow:
157 uow.oclouds.add(resource1)
160 resource_res = ocloud_view.resource_one(resource_id1, uow)
161 assert str(resource_res.get("resourceId")) == resource_id1
164 def test_view_deployment_managers(sqlite_uow):
165 ocloud_id1 = str(uuid.uuid4())
166 deployment_manager_id1 = str(uuid.uuid4())
167 deployment_manager1 = ocloud.DeploymentManager(
168 deployment_manager_id1, "k8s1", ocloud_id1,
169 config.get_api_url()+"/k8s1")
170 with sqlite_uow as uow:
171 uow.oclouds.add(deployment_manager1)
174 deployment_manager_list = ocloud_view.deployment_managers(uow)
175 assert str(deployment_manager_list[0].get(
176 "deploymentManagerId")) == deployment_manager_id1
179 def test_view_deployment_manager_one(sqlite_uow):
180 ocloud_id1 = str(uuid.uuid4())
181 deployment_manager_id1 = str(uuid.uuid4())
182 deployment_manager1 = ocloud.DeploymentManager(
183 deployment_manager_id1, "k8s1", ocloud_id1,
184 config.get_api_url()+"/k8s1")
187 deployment_manager_res = ocloud_view.deployment_manager_one(
188 deployment_manager_id1, sqlite_uow)
189 assert deployment_manager_res is None
191 with sqlite_uow as uow:
192 uow.oclouds.add(deployment_manager1)
195 deployment_manager_res = ocloud_view.deployment_manager_one(
196 deployment_manager_id1, sqlite_uow)
197 assert str(deployment_manager_res.get(
198 "deploymentManagerId")) == deployment_manager_id1
201 def test_view_subscriptions(mock_uow):
202 session, uow = mock_uow
204 subscription_id1 = str(uuid.uuid4())
205 session.return_value.execute.return_value = [{
206 "subscriptionId": subscription_id1,
209 deployment_manager_list = ocloud_view.subscriptions(uow)
210 assert str(deployment_manager_list[0].get(
211 "subscriptionId")) == subscription_id1
214 def test_view_subscription_one(mock_uow):
215 session, uow = mock_uow
217 subscription_id1 = str(uuid.uuid4())
218 session.return_value.execute.return_value.first.return_value = None
221 deployment_manager_res = ocloud_view.subscription_one(
222 subscription_id1, uow)
223 assert deployment_manager_res is None
225 session.return_value.execute.return_value.first.return_value = {
226 "deploymentManagerId": subscription_id1,
229 deployment_manager_res = ocloud_view.subscription_one(
230 subscription_id1, uow)
231 assert str(deployment_manager_res.get(
232 "deploymentManagerId")) == subscription_id1