1 #######################################################################################
2 # Copyright ETSI Contributors and Others.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
22 from n2vc
.temporal_libjuju
import Libjuju
, ConnectionInfo
23 from n2vc
.exceptions
import (
24 JujuControllerFailedConnecting
,
25 JujuModelAlreadyExists
,
28 cacert
= """-----BEGIN CERTIFICATE-----
30 -----END CERTIFICATE-----"""
33 @asynctest.mock
.patch("n2vc.temporal_libjuju.Controller")
34 class LibjujuTestCase(asynctest
.TestCase
):
37 mock_base64_to_cacert
=None,
39 self
.loop
= asyncio
.get_event_loop()
40 self
.connection_info
= ConnectionInfo(
41 "1.2.3.4:17070", "user", "secret", cacert
, "k8s_cloud", "k8s_credentials"
43 logging
.disable(logging
.CRITICAL
)
44 self
.libjuju
= Libjuju(self
.connection_info
)
47 @asynctest.mock
.patch("juju.controller.Controller.connect")
48 class GetControllerTest(LibjujuTestCase
):
50 super(GetControllerTest
, self
).setUp()
52 def test_get_controller(self
, mock_connect
):
53 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
54 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
55 mock_connect
.assert_called_with(
56 endpoint
=self
.connection_info
.endpoint
,
57 username
=self
.connection_info
.user
,
58 password
=self
.connection_info
.password
,
59 cacert
=self
.connection_info
.cacert
,
62 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
65 mock_disconnect_controller
,
68 mock_connect
.side_effect
= Exception()
70 with self
.assertRaises(JujuControllerFailedConnecting
):
71 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
72 self
.assertIsNone(controller
)
73 mock_disconnect_controller
.assert_called()
76 class DisconnectTest(LibjujuTestCase
):
78 super(DisconnectTest
, self
).setUp()
80 @asynctest.mock
.patch("juju.model.Model.disconnect")
81 def test_disconnect_model(self
, mock_disconnect
):
82 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
83 mock_disconnect
.assert_called_once()
85 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
86 def test_disconnect_controller(self
, mock_disconnect
):
87 self
.loop
.run_until_complete(
88 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
90 mock_disconnect
.assert_called_once()
93 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
94 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
95 @asynctest.mock
.patch("juju.controller.Controller.add_model")
96 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
97 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
98 class AddModelTest(LibjujuTestCase
):
100 super(AddModelTest
, self
).setUp()
102 def test_existing_model_raises_exception(
104 mock_disconnect_model
,
105 mock_disconnect_controller
,
110 mock_model_exists
.return_value
= True
111 mock_get_controller
.return_value
= juju
.controller
.Controller()
112 with self
.assertRaises(JujuModelAlreadyExists
):
113 self
.loop
.run_until_complete(self
.libjuju
.add_model("existing_model"))
114 mock_get_controller
.assert_called()
115 mock_add_model
.assert_not_called()
116 mock_disconnect_controller
.assert_called()
118 def test_non_existing_model(
120 mock_disconnect_model
,
121 mock_disconnect_controller
,
126 mock_model_exists
.return_value
= False
127 mock_get_controller
.return_value
= juju
.controller
.Controller()
128 new_model_name
= "nonexisting_model"
129 self
.loop
.run_until_complete(self
.libjuju
.add_model(new_model_name
))
130 mock_add_model
.assert_called_once_with(
132 cloud_name
=self
.connection_info
.cloud_name
,
133 credential_name
=self
.connection_info
.cloud_credentials
,
135 mock_disconnect_controller
.assert_called()
136 mock_disconnect_model
.assert_called()
139 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
140 @asynctest.mock
.patch("juju.controller.Controller.list_models")
141 class ModelExistsTest(LibjujuTestCase
):
143 super(ModelExistsTest
, self
).setUp()
145 async def test_existing_model(
150 model_name
= "existing_model"
151 mock_list_models
.return_value
= [model_name
]
153 await self
.libjuju
.model_exists(model_name
, juju
.controller
.Controller())
156 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
157 async def test_no_controller(
159 mock_disconnect_controller
,
163 model_name
= "existing_model"
164 mock_list_models
.return_value
= [model_name
]
165 mock_get_controller
.return_value
= juju
.controller
.Controller()
166 self
.assertTrue(await self
.libjuju
.model_exists(model_name
))
167 mock_disconnect_controller
.assert_called_once()
169 async def test_non_existing_model(
174 mock_list_models
.return_value
= []
176 await self
.libjuju
.model_exists(
177 "not_existing_model", juju
.controller
.Controller()
182 @asynctest.mock
.patch("juju.controller.Controller.get_model")
183 class GetModelTest(LibjujuTestCase
):
185 super(GetModelTest
, self
).setUp()
191 mock_get_model
.return_value
= juju
.model
.Model()
192 model
= self
.loop
.run_until_complete(
193 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
195 self
.assertIsInstance(model
, juju
.model
.Model
)
198 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
199 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
200 @asynctest.mock
.patch("juju.controller.Controller.list_models")
201 class ListModelsTest(LibjujuTestCase
):
203 super(ListModelsTest
, self
).setUp()
208 mock_disconnect_controller
,
211 expected_list
= ["existingmodel"]
212 mock_get_controller
.return_value
= juju
.controller
.Controller()
213 mock_list_models
.return_value
= expected_list
214 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
216 mock_disconnect_controller
.assert_called_once()
217 self
.assertEquals(models
, expected_list
)
220 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
221 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.get_model")
222 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
223 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
224 @asynctest.mock
.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
225 @asynctest.mock
.patch("juju.controller.Controller.destroy_model")
226 class DestroyModelTest(LibjujuTestCase
):
228 super(DestroyModelTest
, self
).setUp()
230 def test_model_is_destroyed(
233 mock_disconnect_controller
,
234 mock_disconnect_model
,
239 mock_get_controller
.return_value
= juju
.controller
.Controller()
240 mock_model_exists
.return_value
= True
241 mock_get_model
.return_value
= juju
.model
.Model()
242 model_name
= "model_to_destroy"
245 self
.loop
.run_until_complete(self
.libjuju
.destroy_model(model_name
, force
))
246 mock_destroy
.assert_called_with(
247 model_name
, destroy_storage
=True, force
=force
, max_wait
=60
249 mock_disconnect_controller
.assert_called()
250 mock_disconnect_model
.assert_called()
252 def test_not_existing_model(
255 mock_disconnect_controller
,
256 mock_disconnect_model
,
261 mock_get_controller
.return_value
= juju
.controller
.Controller()
262 mock_model_exists
.return_value
= False
263 model_name
= "model_to_destroy"
266 self
.loop
.run_until_complete(self
.libjuju
.destroy_model(model_name
, force
))
267 mock_destroy
.assert_not_called()
268 mock_get_model
.assert_not_called()
269 mock_disconnect_controller
.assert_called()
270 mock_disconnect_model
.assert_called()
272 def test_raise_exception(
275 mock_disconnect_controller
,
276 mock_disconnect_model
,
281 mock_get_controller
.return_value
= juju
.controller
.Controller()
282 mock_model_exists
.return_value
= True
283 mock_get_model
.side_effect
= Exception()
284 model_name
= "model_to_destroy"
287 with self
.assertRaises(Exception):
288 self
.loop
.run_until_complete(self
.libjuju
.destroy_model(model_name
, force
))
289 mock_destroy
.assert_not_called()
290 mock_disconnect_controller
.assert_called()
291 mock_disconnect_model
.assert_called()