1 # Copyright 2021 Canonical Ltd.
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.
16 from unittest
import TestCase
17 from unittest
.mock
import Mock
, patch
, MagicMock
20 from osm_common
import msgbase
21 from osm_common
.dbbase
import DbException
22 from osm_lcm
.vim_sdn
import K8sClusterLcm
, VcaLcm
25 class AsyncMock(MagicMock
):
26 async def __call__(self
, *args
, **kwargs
):
27 return super(AsyncMock
, self
).__call
__(*args
, **kwargs
)
30 class TestVcaLcm(TestCase
):
31 @patch("osm_lcm.lcm_utils.Database")
32 @patch("osm_lcm.lcm_utils.Filesystem")
33 def setUp(self
, mock_filesystem
, mock_database
):
34 self
.loop
= asyncio
.get_event_loop()
35 self
.msg
= Mock(msgbase
.MsgBase())
36 self
.lcm_tasks
= Mock()
37 self
.config
= {"database": {"driver": "mongo"}}
38 self
.vca_lcm
= VcaLcm(self
.msg
, self
.lcm_tasks
, self
.config
)
39 self
.vca_lcm
.db
= Mock()
40 self
.vca_lcm
.fs
= Mock()
42 def test_vca_lcm_create(self
):
43 vca_content
= {"op_id": "order-id", "_id": "id"}
48 "schema_version": "1.11",
51 self
.lcm_tasks
.lock_HA
.return_value
= True
52 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
53 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
54 self
.vca_lcm
.update_db_2
= Mock()
56 self
.loop
.run_until_complete(self
.vca_lcm
.create(vca_content
, order_id
))
58 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "create", "order-id")
59 self
.vca_lcm
.db
.encrypt_decrypt_fields
.assert_called_with(
63 schema_version
="1.11",
66 self
.vca_lcm
.update_db_2
.assert_called_with(
70 "_admin.operationalState": "ENABLED",
71 "_admin.detailed-status": "Connectivity: ok",
74 self
.lcm_tasks
.unlock_HA
.assert_called_with(
78 operationState
="COMPLETED",
79 detailed_status
="VCA validated",
81 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
83 def test_vca_lcm_create_exception(self
):
84 vca_content
= {"op_id": "order-id", "_id": "id"}
89 "schema_version": "1.11",
92 self
.lcm_tasks
.lock_HA
.return_value
= True
93 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
94 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
95 self
.vca_lcm
.n2vc
.validate_vca
.side_effect
= Exception("failed")
96 self
.vca_lcm
.update_db_2
= Mock()
97 self
.vca_lcm
.update_db_2
.side_effect
= DbException("failed")
98 self
.loop
.run_until_complete(self
.vca_lcm
.create(vca_content
, order_id
))
100 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "create", "order-id")
101 self
.vca_lcm
.db
.encrypt_decrypt_fields
.assert_called_with(
104 ["secret", "cacert"],
105 schema_version
="1.11",
108 self
.vca_lcm
.update_db_2
.assert_called_with(
112 "_admin.operationalState": "ERROR",
113 "_admin.detailed-status": "Failed with exception: failed",
116 self
.lcm_tasks
.unlock_HA
.assert_not_called()
117 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
119 def test_vca_lcm_edit_success_no_config(self
):
123 "description": "test-description",
129 "schema_version": "1.11",
131 order_id
= "order-id"
132 self
.lcm_tasks
.lock_HA
.return_value
= True
133 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
134 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
135 self
.vca_lcm
.update_db_2
= Mock()
136 self
.loop
.run_until_complete(self
.vca_lcm
.edit(vca_content
, order_id
))
137 self
.vca_lcm
.n2vc
.validate_vca
.assert_not_called()
138 self
.lcm_tasks
.unlock_HA
.assert_called_with(
142 operationState
="COMPLETED",
143 detailed_status
="Edited",
145 self
.vca_lcm
.update_db_2
.assert_called_with(
150 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
152 def test_vca_lcm_edit_success_config(self
):
153 vca_content
= {"op_id": "order-id", "_id": "id", "cacert": "editcacert"}
158 "schema_version": "1.11",
160 order_id
= "order-id"
161 self
.lcm_tasks
.lock_HA
.return_value
= True
162 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
163 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
164 self
.vca_lcm
.update_db_2
= Mock()
165 self
.loop
.run_until_complete(self
.vca_lcm
.edit(vca_content
, order_id
))
166 self
.vca_lcm
.n2vc
.validate_vca
.assert_called()
167 self
.lcm_tasks
.unlock_HA
.assert_called_with(
171 operationState
="COMPLETED",
172 detailed_status
="Edited",
174 self
.vca_lcm
.update_db_2
.assert_called_with(
178 "_admin.operationalState": "ENABLED",
179 "_admin.detailed-status": "Connectivity: ok",
182 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
184 def test_vca_lcm_edit_exception_no_config(self
):
188 "description": "new-description",
194 "schema_version": "1.11",
196 order_id
= "order-id"
197 self
.lcm_tasks
.lock_HA
.return_value
= True
198 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
199 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
200 # validate_vca should not be called in this case
201 self
.vca_lcm
.n2vc
.validate_vca
.side_effect
= Exception("failed")
202 self
.vca_lcm
.update_db_2
= Mock()
203 self
.loop
.run_until_complete(self
.vca_lcm
.edit(vca_content
, order_id
))
204 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "edit", "order-id")
205 self
.lcm_tasks
.unlock_HA
.assert_called_with(
209 operationState
="COMPLETED",
210 detailed_status
="Edited",
212 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
214 def test_vca_lcm_edit_exception_config(self
):
215 vca_content
= {"op_id": "order-id", "_id": "id", "user": "new-user"}
220 "schema_version": "1.11",
222 order_id
= "order-id"
223 self
.lcm_tasks
.lock_HA
.return_value
= True
224 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
225 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
226 # validate_vca should be called in this case
227 self
.vca_lcm
.n2vc
.validate_vca
.side_effect
= Exception("failed")
228 self
.vca_lcm
.update_db_2
= Mock()
229 self
.loop
.run_until_complete(self
.vca_lcm
.edit(vca_content
, order_id
))
230 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "edit", "order-id")
231 self
.lcm_tasks
.unlock_HA
.assert_called_with(
235 operationState
="FAILED",
236 detailed_status
="Failed with exception: failed",
238 self
.vca_lcm
.update_db_2
.assert_called_with(
242 "_admin.operationalState": "ERROR",
243 "_admin.detailed-status": "Failed with exception: failed",
246 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
248 def test_vca_lcm_edit_db_exception(self
):
252 "description": "new-description",
258 "schema_version": "1.11",
260 order_id
= "order-id"
261 self
.lcm_tasks
.lock_HA
.return_value
= True
262 self
.vca_lcm
.db
.get_one
.return_value
= db_vca
263 self
.vca_lcm
.n2vc
.validate_vca
= AsyncMock()
264 self
.vca_lcm
.update_db_2
= Mock()
265 self
.vca_lcm
.update_db_2
.side_effect
= DbException("failed")
266 self
.loop
.run_until_complete(self
.vca_lcm
.edit(vca_content
, order_id
))
267 self
.vca_lcm
.n2vc
.validate_vca
.assert_not_called()
268 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "edit", "order-id")
269 self
.vca_lcm
.update_db_2
.assert_called_with(
274 self
.lcm_tasks
.unlock_HA
.assert_not_called()
275 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
277 def test_vca_lcm_delete(self
):
278 vca_content
= {"op_id": "order-id", "_id": "id"}
279 order_id
= "order-id"
280 self
.lcm_tasks
.lock_HA
.return_value
= True
281 self
.vca_lcm
.update_db_2
= Mock()
283 self
.loop
.run_until_complete(self
.vca_lcm
.delete(vca_content
, order_id
))
285 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "delete", "order-id")
286 self
.vca_lcm
.db
.del_one
.assert_called_with("vca", {"_id": "id"})
287 self
.vca_lcm
.update_db_2
.assert_called_with("vca", "id", None)
288 self
.lcm_tasks
.unlock_HA
.assert_called_with(
292 operationState
="COMPLETED",
293 detailed_status
="deleted",
295 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
297 def test_vca_lcm_delete_exception(self
):
298 vca_content
= {"op_id": "order-id", "_id": "id"}
299 order_id
= "order-id"
300 self
.lcm_tasks
.lock_HA
.return_value
= True
301 self
.vca_lcm
.update_db_2
= Mock()
302 self
.vca_lcm
.db
.del_one
.side_effect
= Exception("failed deleting")
303 self
.vca_lcm
.update_db_2
.side_effect
= DbException("failed")
305 self
.loop
.run_until_complete(self
.vca_lcm
.delete(vca_content
, order_id
))
307 self
.lcm_tasks
.lock_HA
.assert_called_with("vca", "delete", "order-id")
308 self
.vca_lcm
.db
.del_one
.assert_called_with("vca", {"_id": "id"})
309 self
.vca_lcm
.update_db_2
.assert_called_with(
313 "_admin.operationalState": "ERROR",
314 "_admin.detailed-status": "Failed with exception: failed deleting",
317 self
.lcm_tasks
.unlock_HA
.not_called()
318 self
.lcm_tasks
.remove
.assert_called_with("vca", "id", "order-id")
321 class TestK8SClusterLcm(TestCase
):
322 @patch("osm_lcm.vim_sdn.K8sHelmConnector")
323 @patch("osm_lcm.vim_sdn.K8sHelm3Connector")
324 @patch("osm_lcm.vim_sdn.K8sJujuConnector")
325 @patch("osm_lcm.lcm_utils.Database")
326 @patch("osm_lcm.lcm_utils.Filesystem")
335 self
.loop
= asyncio
.get_event_loop()
336 self
.msg
= Mock(msgbase
.MsgBase())
337 self
.lcm_tasks
= Mock()
338 self
.config
= {"database": {"driver": "mongo"}}
341 "helmpath": "/usr/local/bin/helm",
342 "helm3path": "/usr/local/bin/helm3",
343 "kubectlpath": "/usr/bin/kubectl",
346 self
.k8scluster_lcm
= K8sClusterLcm(self
.msg
, self
.lcm_tasks
, self
.vca_config
)
347 self
.k8scluster_lcm
.db
= Mock()
348 self
.k8scluster_lcm
.fs
= Mock()
350 def test_k8scluster_edit(self
):
351 k8scluster_content
= {"op_id": "op-id", "_id": "id"}
352 order_id
= "order-id"
353 self
.lcm_tasks
.lock_HA
.return_value
= True
354 self
.loop
.run_until_complete(
355 self
.k8scluster_lcm
.edit(k8scluster_content
, order_id
)
357 self
.lcm_tasks
.unlock_HA
.assert_called_with(
361 operationState
="COMPLETED",
362 detailed_status
="Not implemented",
364 self
.lcm_tasks
.remove
.assert_called_with("k8scluster", "id", order_id
)
366 def test_k8scluster_edit_lock_false(self
):
367 k8scluster_content
= {"op_id": "op-id", "_id": "id"}
368 order_id
= "order-id"
369 self
.lcm_tasks
.lock_HA
.return_value
= False
370 self
.loop
.run_until_complete(
371 self
.k8scluster_lcm
.edit(k8scluster_content
, order_id
)
373 self
.lcm_tasks
.unlock_HA
.assert_not_called()
374 self
.lcm_tasks
.remove
.assert_not_called()
376 def test_k8scluster_edit_no_opid(self
):
377 k8scluster_content
= {"_id": "id"}
378 order_id
= "order-id"
379 self
.lcm_tasks
.lock_HA
.return_value
= True
380 self
.loop
.run_until_complete(
381 self
.k8scluster_lcm
.edit(k8scluster_content
, order_id
)
383 self
.lcm_tasks
.unlock_HA
.assert_called_with(
387 operationState
="COMPLETED",
388 detailed_status
="Not implemented",
390 self
.lcm_tasks
.remove
.assert_called_with("k8scluster", "id", order_id
)
392 def test_k8scluster_edit_no_orderid(self
):
393 k8scluster_content
= {"op_id": "op-id", "_id": "id"}
395 self
.lcm_tasks
.lock_HA
.return_value
= True
396 self
.loop
.run_until_complete(
397 self
.k8scluster_lcm
.edit(k8scluster_content
, order_id
)
399 self
.lcm_tasks
.unlock_HA
.assert_called_with(
403 operationState
="COMPLETED",
404 detailed_status
="Not implemented",
406 self
.lcm_tasks
.remove
.assert_called_with("k8scluster", "id", order_id
)