1 # Copyright 2020 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.
18 from unittest
.mock
import Mock
, patch
21 from juju
.errors
import JujuAPIError
29 from n2vc
.libjuju
import Libjuju
30 from n2vc
.exceptions
import (
31 JujuControllerFailedConnecting
,
33 JujuApplicationNotFound
,
35 JujuApplicationExists
,
36 JujuInvalidK8sConfiguration
,
37 JujuLeaderUnitNotFound
,
40 from n2vc
.k8s_juju_conn
import generate_rbac_id
41 from n2vc
.tests
.unit
.utils
import AsyncMock
42 from n2vc
.vca
.connection
import Connection
43 from n2vc
.vca
.connection_data
import ConnectionData
46 cacert
= """-----BEGIN CERTIFICATE-----
48 -----END CERTIFICATE-----"""
51 @asynctest.mock
.patch("n2vc.libjuju.Controller")
52 class LibjujuTestCase(asynctest
.TestCase
):
53 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
56 mock_base64_to_cacert
=None,
58 self
.loop
= asyncio
.get_event_loop()
60 mock_base64_to_cacert
.return_value
= cacert
61 Connection
._load
_vca
_connection
_data
= Mock()
62 vca_connection
= Connection(AsyncMock())
63 vca_connection
._data
= ConnectionData(
65 "endpoints": ["1.2.3.4:17070"],
71 "lxd-credentials": "credentials",
72 "k8s-cloud": "k8s_cloud",
73 "k8s-credentials": "k8s_credentials",
75 "api-proxy": "api_proxy",
78 logging
.disable(logging
.CRITICAL
)
79 self
.libjuju
= Libjuju(vca_connection
, self
.loop
)
80 self
.loop
.run_until_complete(self
.libjuju
.disconnect())
83 @asynctest.mock
.patch("juju.controller.Controller.connect")
84 @asynctest.mock
.patch(
85 "juju.controller.Controller.api_endpoints",
86 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
88 class GetControllerTest(LibjujuTestCase
):
90 super(GetControllerTest
, self
).setUp()
92 def test_diff_endpoint(self
, mock_api_endpoints
, mock_connect
):
93 self
.libjuju
.endpoints
= []
94 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
95 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
97 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
100 mock_disconnect_controller
,
104 self
.libjuju
.endpoints
= []
106 mock_connect
.side_effect
= Exception()
108 with self
.assertRaises(JujuControllerFailedConnecting
):
109 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
110 self
.assertIsNone(controller
)
111 mock_disconnect_controller
.assert_called()
113 def test_same_endpoint_get_controller(self
, mock_api_endpoints
, mock_connect
):
114 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
115 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
116 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
119 class DisconnectTest(LibjujuTestCase
):
121 super(DisconnectTest
, self
).setUp()
123 @asynctest.mock
.patch("juju.model.Model.disconnect")
124 def test_disconnect_model(self
, mock_disconnect
):
125 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
126 mock_disconnect
.assert_called_once()
128 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
129 def test_disconnect_controller(self
, mock_disconnect
):
130 self
.loop
.run_until_complete(
131 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
133 mock_disconnect
.assert_called_once()
136 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
137 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
138 @asynctest.mock
.patch("juju.controller.Controller.add_model")
139 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
140 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
141 class AddModelTest(LibjujuTestCase
):
143 super(AddModelTest
, self
).setUp()
145 def test_existing_model(
147 mock_disconnect_model
,
148 mock_disconnect_controller
,
153 mock_model_exists
.return_value
= True
155 # This should not raise an exception
156 self
.loop
.run_until_complete(self
.libjuju
.add_model("existing_model", "cloud"))
158 mock_disconnect_controller
.assert_called()
160 # TODO Check two job executing at the same time and one returning without doing anything.
162 def test_non_existing_model(
164 mock_disconnect_model
,
165 mock_disconnect_controller
,
170 mock_model_exists
.return_value
= False
171 mock_get_controller
.return_value
= juju
.controller
.Controller()
173 self
.loop
.run_until_complete(
174 self
.libjuju
.add_model("nonexisting_model", Mock())
177 mock_add_model
.assert_called_once()
178 mock_disconnect_controller
.assert_called()
179 mock_disconnect_model
.assert_called()
182 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
183 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
184 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
185 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
186 @asynctest.mock
.patch(
187 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
189 @asynctest.mock
.patch("juju.model.Model.get_action_status")
190 @asynctest.mock
.patch("juju.model.Model.get_action_output")
191 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_actions")
192 class GetExecutedActionsTest(LibjujuTestCase
):
194 super(GetExecutedActionsTest
, self
).setUp()
199 mock_get_action_output
,
200 mock_get_action_status
,
202 mock_disconnect_controller
,
203 mock_disconnect_model
,
207 mock_get_model
.return_value
= None
208 with self
.assertRaises(JujuError
):
209 self
.loop
.run_until_complete(self
.libjuju
.get_executed_actions("model"))
211 mock_get_controller
.assert_called_once()
212 mock_disconnect_controller
.assert_called_once()
213 mock_get_model
.assert_called_once()
214 mock_disconnect_model
.assert_not_called()
219 mock_get_action_output
,
220 mock_get_action_status
,
222 mock_disconnect_controller
,
223 mock_disconnect_model
,
227 mock_get_model
.return_value
= juju
.model
.Model()
228 mock_applications
.return_value
= {"existing_app"}
229 mock_get_actions
.return_value
= {"action_name": "description"}
230 mock_get_action_status
.return_value
= {"id": "status"}
231 mock_get_action_output
.return_value
= {"output": "completed"}
233 executed_actions
= self
.loop
.run_until_complete(
234 self
.libjuju
.get_executed_actions("model")
239 "action": "action_name",
241 "output": "completed",
244 self
.assertListEqual(expected_result
, executed_actions
)
245 self
.assertIsInstance(executed_actions
, list)
247 mock_get_controller
.assert_called_once()
248 mock_get_model
.assert_called_once()
249 mock_disconnect_controller
.assert_called_once()
250 mock_disconnect_model
.assert_called_once()
253 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
254 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
255 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
256 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
257 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
258 class GetApplicationConfigsTest(LibjujuTestCase
):
260 super(GetApplicationConfigsTest
, self
).setUp()
264 mock_get_application
,
265 mock_disconnect_controller
,
266 mock_disconnect_model
,
270 mock_get_model
.return_value
= None
271 with self
.assertRaises(JujuError
):
272 self
.loop
.run_until_complete(
273 self
.libjuju
.get_application_configs("model", "app")
276 mock_get_controller
.assert_called_once()
277 mock_disconnect_controller
.assert_called_once()
278 mock_get_model
.assert_called_once()
279 mock_disconnect_model
.assert_not_called()
283 mock_get_application
,
284 mock_disconnect_controller
,
285 mock_disconnect_model
,
289 mock_get_application
.return_value
= FakeApplication()
290 application_configs
= self
.loop
.run_until_complete(
291 self
.libjuju
.get_application_configs("model", "app")
294 self
.assertEqual(application_configs
, ["app_config"])
296 mock_get_controller
.assert_called_once()
297 mock_get_model
.assert_called_once()
298 mock_disconnect_controller
.assert_called_once()
299 mock_disconnect_model
.assert_called_once()
302 @asynctest.mock
.patch("juju.controller.Controller.get_model")
303 class GetModelTest(LibjujuTestCase
):
305 super(GetModelTest
, self
).setUp()
311 mock_get_model
.return_value
= juju
.model
.Model()
312 model
= self
.loop
.run_until_complete(
313 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
315 self
.assertIsInstance(model
, juju
.model
.Model
)
318 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
319 @asynctest.mock
.patch("juju.controller.Controller.list_models")
320 class ModelExistsTest(LibjujuTestCase
):
322 super(ModelExistsTest
, self
).setUp()
324 async def test_existing_model(
329 mock_list_models
.return_value
= ["existing_model"]
331 await self
.libjuju
.model_exists(
332 "existing_model", juju
.controller
.Controller()
336 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
337 async def test_no_controller(
339 mock_disconnect_controller
,
343 mock_list_models
.return_value
= ["existing_model"]
344 mock_get_controller
.return_value
= juju
.controller
.Controller()
345 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
346 mock_disconnect_controller
.assert_called_once()
348 async def test_non_existing_model(
353 mock_list_models
.return_value
= ["existing_model"]
355 await self
.libjuju
.model_exists(
356 "not_existing_model", juju
.controller
.Controller()
361 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
362 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
363 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
364 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
365 @asynctest.mock
.patch("juju.model.Model.get_status")
366 class GetModelStatusTest(LibjujuTestCase
):
368 super(GetModelStatusTest
, self
).setUp()
373 mock_disconnect_controller
,
374 mock_disconnect_model
,
378 mock_get_model
.return_value
= juju
.model
.Model()
379 mock_get_status
.return_value
= {"status"}
381 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
383 mock_get_status
.assert_called_once()
384 mock_disconnect_controller
.assert_called_once()
385 mock_disconnect_model
.assert_called_once()
387 self
.assertEqual(status
, {"status"})
392 mock_disconnect_controller
,
393 mock_disconnect_model
,
397 mock_get_model
.return_value
= juju
.model
.Model()
398 mock_get_status
.side_effect
= Exception()
400 with self
.assertRaises(Exception):
401 status
= self
.loop
.run_until_complete(
402 self
.libjuju
.get_model_status("model")
405 mock_disconnect_controller
.assert_called_once()
406 mock_disconnect_model
.assert_called_once()
408 self
.assertIsNone(status
)
411 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
412 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
413 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
414 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
415 @asynctest.mock
.patch("juju.model.Model.get_machines")
416 @asynctest.mock
.patch("juju.model.Model.add_machine")
417 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
418 class CreateMachineTest(LibjujuTestCase
):
420 super(CreateMachineTest
, self
).setUp()
422 def test_existing_machine(
427 mock_disconnect_controller
,
428 mock_disconnect_model
,
432 mock_get_model
.return_value
= juju
.model
.Model()
433 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
434 machine
, bool_res
= self
.loop
.run_until_complete(
435 self
.libjuju
.create_machine("model", "existing_machine")
438 self
.assertIsInstance(machine
, FakeMachine
)
439 self
.assertFalse(bool_res
)
441 mock_disconnect_controller
.assert_called()
442 mock_disconnect_model
.assert_called()
444 def test_non_existing_machine(
449 mock_disconnect_controller
,
450 mock_disconnect_model
,
456 mock_get_model
.return_value
= juju
.model
.Model()
457 with self
.assertRaises(JujuMachineNotFound
):
458 machine
, bool_res
= self
.loop
.run_until_complete(
459 self
.libjuju
.create_machine("model", "non_existing_machine")
461 self
.assertIsNone(machine
)
462 self
.assertIsNone(bool_res
)
464 mock_disconnect_controller
.assert_called()
465 mock_disconnect_model
.assert_called()
472 mock_disconnect_controller
,
473 mock_disconnect_model
,
477 mock_get_model
.return_value
= juju
.model
.Model()
478 mock_add_machine
.return_value
= FakeMachine()
480 machine
, bool_res
= self
.loop
.run_until_complete(
481 self
.libjuju
.create_machine("model")
484 self
.assertIsInstance(machine
, FakeMachine
)
485 self
.assertTrue(bool_res
)
487 mock_wait_for
.assert_called_once()
488 mock_add_machine
.assert_called_once()
490 mock_disconnect_controller
.assert_called()
491 mock_disconnect_model
.assert_called()
494 # TODO test provision machine
497 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
498 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
499 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
500 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
501 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
502 @asynctest.mock
.patch("juju.model.Model.deploy")
503 class DeployTest(LibjujuTestCase
):
505 super(DeployTest
, self
).setUp()
511 mock_disconnect_controller
,
512 mock_disconnect_model
,
516 mock_get_model
.return_value
= juju
.model
.Model()
517 self
.loop
.run_until_complete(
518 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
520 mock_deploy
.assert_called_once()
521 mock_wait_for_model
.assert_called_once()
522 mock_disconnect_controller
.assert_called_once()
523 mock_disconnect_model
.assert_called_once()
525 def test_deploy_no_wait(
529 mock_disconnect_controller
,
530 mock_disconnect_model
,
534 mock_get_model
.return_value
= juju
.model
.Model()
535 self
.loop
.run_until_complete(
536 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
538 mock_deploy
.assert_called_once()
539 mock_wait_for_model
.assert_not_called()
540 mock_disconnect_controller
.assert_called_once()
541 mock_disconnect_model
.assert_called_once()
543 def test_deploy_exception(
547 mock_disconnect_controller
,
548 mock_disconnect_model
,
552 mock_deploy
.side_effect
= Exception()
553 mock_get_model
.return_value
= juju
.model
.Model()
554 with self
.assertRaises(Exception):
555 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
556 mock_deploy
.assert_called_once()
557 mock_wait_for_model
.assert_not_called()
558 mock_disconnect_controller
.assert_called_once()
559 mock_disconnect_model
.assert_called_once()
562 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
563 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
564 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
565 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
566 @asynctest.mock
.patch(
567 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
569 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
570 @asynctest.mock
.patch("juju.model.Model.deploy")
571 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
572 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
573 class DeployCharmTest(LibjujuTestCase
):
575 super(DeployCharmTest
, self
).setUp()
577 def test_existing_app(
584 mock_disconnect_controller
,
585 mock_disconnect_model
,
589 mock_get_model
.return_value
= juju
.model
.Model()
590 mock_applications
.return_value
= {"existing_app"}
593 with self
.assertRaises(JujuApplicationExists
):
594 application
= self
.loop
.run_until_complete(
595 self
.libjuju
.deploy_charm(
602 self
.assertIsNone(application
)
604 mock_disconnect_controller
.assert_called()
605 mock_disconnect_model
.assert_called()
607 def test_non_existing_machine(
614 mock_disconnect_controller
,
615 mock_disconnect_model
,
619 mock_get_model
.return_value
= juju
.model
.Model()
620 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
622 with self
.assertRaises(JujuMachineNotFound
):
623 application
= self
.loop
.run_until_complete(
624 self
.libjuju
.deploy_charm(
632 self
.assertIsNone(application
)
634 mock_disconnect_controller
.assert_called()
635 mock_disconnect_model
.assert_called()
644 mock_disconnect_controller
,
645 mock_disconnect_model
,
649 mock_get_model
.return_value
= juju
.model
.Model()
650 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
651 mock_create_machine
.return_value
= (FakeMachine(), "other")
652 mock_deploy
.return_value
= FakeApplication()
653 application
= self
.loop
.run_until_complete(
654 self
.libjuju
.deploy_charm(
663 self
.assertIsInstance(application
, FakeApplication
)
665 mock_deploy
.assert_called_once()
666 mock_wait_for
.assert_called_once()
668 mock_create_machine
.assert_called_once()
670 mock_disconnect_controller
.assert_called()
671 mock_disconnect_model
.assert_called()
680 mock_disconnect_controller
,
681 mock_disconnect_model
,
685 mock_get_model
.return_value
= juju
.model
.Model()
686 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
687 mock_deploy
.return_value
= FakeApplication()
688 application
= self
.loop
.run_until_complete(
689 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
692 self
.assertIsInstance(application
, FakeApplication
)
694 mock_deploy
.assert_called_once()
695 mock_wait_for
.assert_called_once()
697 mock_disconnect_controller
.assert_called()
698 mock_disconnect_model
.assert_called()
701 @asynctest.mock
.patch(
702 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
704 class GetApplicationTest(LibjujuTestCase
):
706 super(GetApplicationTest
, self
).setUp()
708 def test_existing_application(
712 mock_applications
.return_value
= {"existing_app": "exists"}
713 model
= juju
.model
.Model()
714 result
= self
.libjuju
._get
_application
(model
, "existing_app")
715 self
.assertEqual(result
, "exists")
717 def test_non_existing_application(
721 mock_applications
.return_value
= {"existing_app": "exists"}
722 model
= juju
.model
.Model()
723 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
724 self
.assertIsNone(result
)
727 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
728 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
729 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
730 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
731 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
732 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
733 @asynctest.mock
.patch("juju.model.Model.get_action_output")
734 @asynctest.mock
.patch("juju.model.Model.get_action_status")
735 class ExecuteActionTest(LibjujuTestCase
):
737 super(ExecuteActionTest
, self
).setUp()
739 def test_no_application(
741 mock_get_action_status
,
742 mock_get_action_output
,
744 mock__get_application
,
745 mock_disconnect_controller
,
746 mock_disconnect_model
,
750 mock__get_application
.return_value
= None
751 mock_get_model
.return_value
= juju
.model
.Model()
754 with self
.assertRaises(JujuApplicationNotFound
):
755 output
, status
= self
.loop
.run_until_complete(
756 self
.libjuju
.execute_action(
762 self
.assertIsNone(output
)
763 self
.assertIsNone(status
)
765 mock_disconnect_controller
.assert_called()
766 mock_disconnect_model
.assert_called()
770 mock_get_action_status
,
771 mock_get_action_output
,
773 mock__get_application
,
774 mock_disconnect_controller
,
775 mock_disconnect_model
,
780 mock_get_model
.return_value
= juju
.model
.Model()
781 mock__get_application
.return_value
= FakeApplication()
784 with self
.assertRaises(JujuActionNotFound
):
785 output
, status
= self
.loop
.run_until_complete(
786 self
.libjuju
.execute_action(
792 self
.assertIsNone(output
)
793 self
.assertIsNone(status
)
795 mock_disconnect_controller
.assert_called()
796 mock_disconnect_model
.assert_called()
798 @asynctest.mock
.patch("asyncio.sleep")
799 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
802 mock_is_leader_from_status
,
804 mock_get_action_status
,
805 mock_get_action_output
,
807 mock__get_application
,
808 mock_disconnect_controller
,
809 mock_disconnect_model
,
813 mock_get_model
.return_value
= juju
.model
.Model()
814 mock__get_application
.return_value
= FakeApplication()
815 mock_is_leader_from_status
.return_value
= False
818 with self
.assertRaises(JujuLeaderUnitNotFound
):
819 output
, status
= self
.loop
.run_until_complete(
820 self
.libjuju
.execute_action(
826 self
.assertIsNone(output
)
827 self
.assertIsNone(status
)
829 mock_disconnect_controller
.assert_called()
830 mock_disconnect_model
.assert_called()
832 def test_successful_exec(
834 mock_get_action_status
,
835 mock_get_action_output
,
837 mock__get_application
,
838 mock_disconnect_controller
,
839 mock_disconnect_model
,
843 mock_get_model
.return_value
= juju
.model
.Model()
844 mock__get_application
.return_value
= FakeApplication()
845 mock_get_action_output
.return_value
= "output"
846 mock_get_action_status
.return_value
= {"id": "status"}
847 output
, status
= self
.loop
.run_until_complete(
848 self
.libjuju
.execute_action("app", "model", "existing_action")
850 self
.assertEqual(output
, "output")
851 self
.assertEqual(status
, "status")
853 mock_wait_for
.assert_called_once()
855 mock_disconnect_controller
.assert_called()
856 mock_disconnect_model
.assert_called()
859 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
860 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
861 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
862 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
863 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
864 class GetActionTest(LibjujuTestCase
):
866 super(GetActionTest
, self
).setUp()
870 mock_get_application
,
871 mock_disconnect_controller
,
872 mock_disconnect_model
,
876 mock_get_application
.side_effect
= Exception()
878 with self
.assertRaises(Exception):
879 actions
= self
.loop
.run_until_complete(
880 self
.libjuju
.get_actions("app", "model")
883 self
.assertIsNone(actions
)
884 mock_disconnect_controller
.assert_called_once()
885 mock_disconnect_model
.assert_called_once()
889 mock_get_application
,
890 mock_disconnect_controller
,
891 mock_disconnect_model
,
895 mock_get_application
.return_value
= FakeApplication()
897 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
899 self
.assertEqual(actions
, ["existing_action"])
901 mock_get_controller
.assert_called_once()
902 mock_get_model
.assert_called_once()
903 mock_disconnect_controller
.assert_called_once()
904 mock_disconnect_model
.assert_called_once()
907 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
908 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
909 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
910 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
911 @asynctest.mock
.patch("juju.application.Application.get_metrics")
912 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
913 class GetMetricsTest(LibjujuTestCase
):
915 super(GetMetricsTest
, self
).setUp()
917 def test_get_metrics_success(
919 mock_get_application
,
921 mock_disconnect_controller
,
922 mock_disconnect_model
,
926 mock_get_application
.return_value
= FakeApplication()
927 mock_get_model
.return_value
= juju
.model
.Model()
929 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
931 mock_disconnect_controller
.assert_called_once()
932 mock_disconnect_model
.assert_called_once()
934 def test_get_metrics_exception(
936 mock_get_application
,
938 mock_disconnect_controller
,
939 mock_disconnect_model
,
943 mock_get_model
.return_value
= juju
.model
.Model()
944 mock_get_metrics
.side_effect
= Exception()
945 with self
.assertRaises(Exception):
946 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
948 mock_disconnect_controller
.assert_called_once()
949 mock_disconnect_model
.assert_called_once()
951 def test_missing_args_exception(
953 mock_get_application
,
955 mock_disconnect_controller
,
956 mock_disconnect_model
,
960 mock_get_model
.return_value
= juju
.model
.Model()
962 with self
.assertRaises(Exception):
963 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
965 mock_get_controller
.assert_not_called()
966 mock_get_model
.assert_not_called()
967 mock_disconnect_controller
.assert_not_called()
968 mock_disconnect_model
.assert_not_called()
971 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
972 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
973 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
974 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
975 @asynctest.mock
.patch("juju.model.Model.add_relation")
976 class AddRelationTest(LibjujuTestCase
):
978 super(AddRelationTest
, self
).setUp()
980 @asynctest.mock
.patch("logging.Logger.warning")
985 mock_disconnect_controller
,
986 mock_disconnect_model
,
990 # TODO in libjuju.py should this fail only with a log message?
991 result
= {"error": "not found", "response": "response", "request-id": 1}
993 mock_get_model
.return_value
= juju
.model
.Model()
994 mock_add_relation
.side_effect
= JujuAPIError(result
)
996 self
.loop
.run_until_complete(
997 self
.libjuju
.add_relation(
1004 mock_warning
.assert_called_with("Relation not found: not found")
1005 mock_disconnect_controller
.assert_called_once()
1006 mock_disconnect_model
.assert_called_once()
1008 @asynctest.mock
.patch("logging.Logger.warning")
1009 def test_already_exists(
1013 mock_disconnect_controller
,
1014 mock_disconnect_model
,
1016 mock_get_controller
,
1018 # TODO in libjuju.py should this fail silently?
1019 result
= {"error": "already exists", "response": "response", "request-id": 1}
1021 mock_get_model
.return_value
= juju
.model
.Model()
1022 mock_add_relation
.side_effect
= JujuAPIError(result
)
1024 self
.loop
.run_until_complete(
1025 self
.libjuju
.add_relation(
1032 mock_warning
.assert_called_with("Relation already exists: already exists")
1033 mock_disconnect_controller
.assert_called_once()
1034 mock_disconnect_model
.assert_called_once()
1039 mock_disconnect_controller
,
1040 mock_disconnect_model
,
1042 mock_get_controller
,
1044 mock_get_model
.return_value
= juju
.model
.Model()
1045 result
= {"error": "", "response": "response", "request-id": 1}
1046 mock_add_relation
.side_effect
= JujuAPIError(result
)
1048 with self
.assertRaises(JujuAPIError
):
1049 self
.loop
.run_until_complete(
1050 self
.libjuju
.add_relation(
1057 mock_disconnect_controller
.assert_called_once()
1058 mock_disconnect_model
.assert_called_once()
1063 mock_disconnect_controller
,
1064 mock_disconnect_model
,
1066 mock_get_controller
,
1068 mock_get_model
.return_value
= juju
.model
.Model()
1070 self
.loop
.run_until_complete(
1071 self
.libjuju
.add_relation(
1078 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1079 mock_disconnect_controller
.assert_called_once()
1080 mock_disconnect_model
.assert_called_once()
1085 mock_disconnect_controller
,
1086 mock_disconnect_model
,
1088 mock_get_controller
,
1090 mock_get_model
.return_value
= juju
.model
.Model()
1092 self
.loop
.run_until_complete(
1093 self
.libjuju
.add_relation(
1100 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1101 mock_disconnect_controller
.assert_called_once()
1102 mock_disconnect_model
.assert_called_once()
1105 # TODO destroy_model testcase
1108 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1109 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1112 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1113 class DestroyApplicationTest(LibjujuTestCase
):
1115 super(DestroyApplicationTest
, self
).setUp()
1119 mock_get_controller
,
1121 mock_disconnect_controller
,
1122 mock_get_application
,
1123 mock_disconnect_model
,
1125 mock_get_application
.return_value
= FakeApplication()
1126 mock_get_model
.return_value
= None
1127 self
.loop
.run_until_complete(
1128 self
.libjuju
.destroy_application(
1134 mock_get_application
.assert_called()
1135 mock_disconnect_controller
.assert_called_once()
1136 mock_disconnect_model
.assert_called_once()
1138 def test_no_application(
1140 mock_get_controller
,
1142 mock_disconnect_controller
,
1143 mock_get_application
,
1144 mock_disconnect_model
,
1146 mock_get_model
.return_value
= None
1147 mock_get_application
.return_value
= None
1149 self
.loop
.run_until_complete(
1150 self
.libjuju
.destroy_application(
1156 mock_get_application
.assert_called()
1160 mock_get_controller
,
1162 mock_disconnect_controller
,
1163 mock_get_application
,
1164 mock_disconnect_model
,
1166 mock_get_application
.return_value
= FakeApplication
1167 mock_get_model
.return_value
= None
1169 with self
.assertRaises(Exception):
1170 self
.loop
.run_until_complete(
1171 self
.libjuju
.destroy_application(
1177 mock_get_application
.assert_called_once()
1180 # @asynctest.mock.patch("juju.model.Model.get_machines")
1181 # @asynctest.mock.patch("logging.Logger.debug")
1182 # class DestroyMachineTest(LibjujuTestCase):
1184 # super(DestroyMachineTest, self).setUp()
1186 # def test_success_manual_machine(
1187 # self, mock_debug, mock_get_machines,
1189 # mock_get_machines.side_effect = [
1190 # {"machine": FakeManualMachine()},
1191 # {"machine": FakeManualMachine()},
1194 # self.loop.run_until_complete(
1195 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1198 # asynctest.call("Waiting for machine machine is destroyed"),
1199 # asynctest.call("Machine destroyed: machine"),
1201 # mock_debug.assert_has_calls(calls)
1203 # def test_no_machine(
1204 # self, mock_debug, mock_get_machines,
1206 # mock_get_machines.return_value = {}
1207 # self.loop.run_until_complete(
1208 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1210 # mock_debug.assert_called_with("Machine not found: machine")
1213 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1214 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1215 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1216 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1218 class ConfigureApplicationTest(LibjujuTestCase
):
1220 super(ConfigureApplicationTest
, self
).setUp()
1224 mock_get_application
,
1225 mock_disconnect_controller
,
1226 mock_disconnect_model
,
1228 mock_get_controller
,
1231 mock_get_application
.return_value
= FakeApplication()
1233 self
.loop
.run_until_complete(
1234 self
.libjuju
.configure_application(
1240 mock_get_application
.assert_called_once()
1241 mock_disconnect_controller
.assert_called_once()
1242 mock_disconnect_model
.assert_called_once()
1246 mock_get_application
,
1247 mock_disconnect_controller
,
1248 mock_disconnect_model
,
1250 mock_get_controller
,
1253 mock_get_application
.side_effect
= Exception()
1255 with self
.assertRaises(Exception):
1256 self
.loop
.run_until_complete(
1257 self
.libjuju
.configure_application(
1263 mock_disconnect_controller
.assert_called_once()
1264 mock_disconnect_model
.assert_called_once()
1266 def test_controller_exception(
1268 mock_get_application
,
1269 mock_disconnect_controller
,
1270 mock_disconnect_model
,
1272 mock_get_controller
,
1275 result
= {"error": "not found", "response": "response", "request-id": 1}
1277 mock_get_controller
.side_effect
= JujuAPIError(result
)
1279 with self
.assertRaises(JujuAPIError
):
1280 self
.loop
.run_until_complete(
1281 self
.libjuju
.configure_application(
1287 mock_get_model
.assert_not_called()
1288 mock_disconnect_controller
.assert_not_called()
1289 mock_disconnect_model
.assert_not_called()
1291 def test_get_model_exception(
1293 mock_get_application
,
1294 mock_disconnect_controller
,
1295 mock_disconnect_model
,
1297 mock_get_controller
,
1300 result
= {"error": "not found", "response": "response", "request-id": 1}
1301 mock_get_model
.side_effect
= JujuAPIError(result
)
1303 with self
.assertRaises(JujuAPIError
):
1304 self
.loop
.run_until_complete(
1305 self
.libjuju
.configure_application(
1311 mock_get_model
.assert_called_once()
1312 mock_disconnect_controller
.assert_called_once()
1313 mock_disconnect_model
.assert_not_called()
1316 # TODO _get_api_endpoints_db test case
1317 # TODO _update_api_endpoints_db test case
1318 # TODO healthcheck test case
1321 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1322 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1323 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1324 class ListModelsTest(LibjujuTestCase
):
1326 super(ListModelsTest
, self
).setUp()
1328 def test_containing(
1331 mock_disconnect_controller
,
1332 mock_get_controller
,
1334 mock_get_controller
.return_value
= juju
.controller
.Controller()
1335 mock_list_models
.return_value
= ["existingmodel"]
1336 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1338 mock_disconnect_controller
.assert_called_once()
1339 self
.assertEquals(models
, ["existingmodel"])
1341 def test_not_containing(
1344 mock_disconnect_controller
,
1345 mock_get_controller
,
1347 mock_get_controller
.return_value
= juju
.controller
.Controller()
1348 mock_list_models
.return_value
= ["existingmodel", "model"]
1349 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1351 mock_disconnect_controller
.assert_called_once()
1352 self
.assertEquals(models
, [])
1354 def test_no_contains_arg(
1357 mock_disconnect_controller
,
1358 mock_get_controller
,
1360 mock_get_controller
.return_value
= juju
.controller
.Controller()
1361 mock_list_models
.return_value
= ["existingmodel", "model"]
1362 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1364 mock_disconnect_controller
.assert_called_once()
1365 self
.assertEquals(models
, ["existingmodel", "model"])
1368 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1369 class ModelsExistTest(LibjujuTestCase
):
1371 super(ModelsExistTest
, self
).setUp()
1373 def test_model_names_none(self
, mock_list_models
):
1374 mock_list_models
.return_value
= []
1375 with self
.assertRaises(Exception):
1376 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1378 def test_model_names_empty(self
, mock_list_models
):
1379 mock_list_models
.return_value
= []
1380 with self
.assertRaises(Exception):
1381 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1382 self
.libjuju
.models_exist([])
1385 def test_model_names_not_existing(self
, mock_list_models
):
1386 mock_list_models
.return_value
= ["prometheus", "grafana"]
1387 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1388 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1390 self
.assertFalse(exist
)
1391 self
.assertEqual(non_existing_models
, ["prometheus2"])
1393 def test_model_names_exist(self
, mock_list_models
):
1394 mock_list_models
.return_value
= ["prometheus", "grafana"]
1395 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1396 self
.libjuju
.models_exist(["prometheus", "grafana"])
1398 self
.assertTrue(exist
)
1399 self
.assertEqual(non_existing_models
, [])
1402 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1403 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1404 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1405 class ListOffers(LibjujuTestCase
):
1407 super(ListOffers
, self
).setUp()
1409 def test_disconnect_controller(
1412 mock_disconnect_controller
,
1413 mock_get_controller
,
1415 mock_get_controller
.return_value
= juju
.controller
.Controller()
1416 mock_list_offers
.side_effect
= Exception()
1417 with self
.assertRaises(Exception):
1418 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1419 mock_disconnect_controller
.assert_called_once()
1421 def test_empty_list(
1424 mock_disconnect_controller
,
1425 mock_get_controller
,
1427 mock_get_controller
.return_value
= juju
.controller
.Controller()
1428 mock_list_offers
.return_value
= []
1429 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1430 self
.assertEqual(offers
, [])
1431 mock_disconnect_controller
.assert_called_once()
1433 def test_non_empty_list(
1436 mock_disconnect_controller
,
1437 mock_get_controller
,
1439 mock_get_controller
.return_value
= juju
.controller
.Controller()
1440 mock_list_offers
.return_value
= ["offer"]
1441 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1442 self
.assertEqual(offers
, ["offer"])
1443 mock_disconnect_controller
.assert_called_once()
1446 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1447 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1448 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1449 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1450 @asynctest.mock
.patch("juju.model.Model.consume")
1451 class ConsumeTest(LibjujuTestCase
):
1453 super(ConsumeTest
, self
).setUp()
1458 mock_disconnect_controller
,
1459 mock_disconnect_model
,
1461 mock_get_controller
,
1463 mock_get_controller
.return_value
= juju
.controller
.Controller()
1464 mock_get_model
.return_value
= juju
.model
.Model()
1466 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1467 mock_consume
.assert_called_once()
1468 mock_disconnect_model
.assert_called_once()
1469 mock_disconnect_controller
.assert_called_once()
1471 def test_parsing_error_exception(
1474 mock_disconnect_controller
,
1475 mock_disconnect_model
,
1477 mock_get_controller
,
1479 mock_get_controller
.return_value
= juju
.controller
.Controller()
1480 mock_get_model
.return_value
= juju
.model
.Model()
1481 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1483 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1484 self
.loop
.run_until_complete(
1485 self
.libjuju
.consume("offer_url", "model_name")
1487 mock_consume
.assert_called_once()
1488 mock_disconnect_model
.assert_called_once()
1489 mock_disconnect_controller
.assert_called_once()
1491 def test_juju_error_exception(
1494 mock_disconnect_controller
,
1495 mock_disconnect_model
,
1497 mock_get_controller
,
1499 mock_get_controller
.return_value
= juju
.controller
.Controller()
1500 mock_get_model
.return_value
= juju
.model
.Model()
1501 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1503 with self
.assertRaises(juju
.errors
.JujuError
):
1504 self
.loop
.run_until_complete(
1505 self
.libjuju
.consume("offer_url", "model_name")
1507 mock_consume
.assert_called_once()
1508 mock_disconnect_model
.assert_called_once()
1509 mock_disconnect_controller
.assert_called_once()
1511 def test_juju_api_error_exception(
1514 mock_disconnect_controller
,
1515 mock_disconnect_model
,
1517 mock_get_controller
,
1519 mock_get_controller
.return_value
= juju
.controller
.Controller()
1520 mock_get_model
.return_value
= juju
.model
.Model()
1521 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1522 {"error": "", "response": "", "request-id": ""}
1525 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1526 self
.loop
.run_until_complete(
1527 self
.libjuju
.consume("offer_url", "model_name")
1529 mock_consume
.assert_called_once()
1530 mock_disconnect_model
.assert_called_once()
1531 mock_disconnect_controller
.assert_called_once()
1534 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1535 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1536 class AddK8sTest(LibjujuTestCase
):
1538 super(AddK8sTest
, self
).setUp()
1540 rbac_id
= generate_rbac_id()
1542 client_cert_data
= "cert"
1543 configuration
= kubernetes
.client
.configuration
.Configuration()
1544 storage_class
= "storage_class"
1545 credential_name
= name
1547 self
._add
_k
8s
_args
= {
1551 "client_cert_data": client_cert_data
,
1552 "configuration": configuration
,
1553 "storage_class": storage_class
,
1554 "credential_name": credential_name
,
1557 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1558 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1559 mock_add_cloud
.assert_called_once()
1560 mock_get_k8s_cloud_credential
.assert_called_once()
1562 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1563 mock_add_cloud
.side_effect
= Exception()
1564 with self
.assertRaises(Exception):
1565 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1566 mock_add_cloud
.assert_called_once()
1567 mock_get_k8s_cloud_credential
.assert_called_once()
1569 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1570 self
._add
_k
8s
_args
["name"] = ""
1571 with self
.assertRaises(Exception):
1572 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1573 mock_add_cloud
.assert_not_called()
1575 def test_add_k8s_missing_storage_name(
1576 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1578 self
._add
_k
8s
_args
["storage_class"] = ""
1579 with self
.assertRaises(Exception):
1580 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1581 mock_add_cloud
.assert_not_called()
1583 def test_add_k8s_missing_configuration_keys(
1584 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1586 self
._add
_k
8s
_args
["configuration"] = None
1587 with self
.assertRaises(Exception):
1588 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1589 mock_add_cloud
.assert_not_called()
1592 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1593 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1594 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1595 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1596 class AddCloudTest(LibjujuTestCase
):
1598 super(AddCloudTest
, self
).setUp()
1599 self
.cloud
= juju
.client
.client
.Cloud()
1600 self
.credential
= juju
.client
.client
.CloudCredential()
1602 def test_add_cloud_with_credential(
1604 mock_add_credential
,
1606 mock_disconnect_controller
,
1607 mock_get_controller
,
1609 mock_get_controller
.return_value
= juju
.controller
.Controller()
1611 cloud
= self
.loop
.run_until_complete(
1612 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1614 self
.assertEqual(cloud
, self
.cloud
)
1615 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1616 mock_add_credential
.assert_called_once_with(
1617 "cloud", credential
=self
.credential
, cloud
="cloud"
1619 mock_disconnect_controller
.assert_called_once()
1621 def test_add_cloud_no_credential(
1623 mock_add_credential
,
1625 mock_disconnect_controller
,
1626 mock_get_controller
,
1628 mock_get_controller
.return_value
= juju
.controller
.Controller()
1630 cloud
= self
.loop
.run_until_complete(
1631 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1633 self
.assertEqual(cloud
, self
.cloud
)
1634 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1635 mock_add_credential
.assert_not_called()
1636 mock_disconnect_controller
.assert_called_once()
1638 def test_add_cloud_exception(
1640 mock_add_credential
,
1642 mock_disconnect_controller
,
1643 mock_get_controller
,
1645 mock_get_controller
.return_value
= juju
.controller
.Controller()
1646 mock_add_cloud
.side_effect
= Exception()
1647 with self
.assertRaises(Exception):
1648 self
.loop
.run_until_complete(
1649 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1652 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1653 mock_add_credential
.assert_not_called()
1654 mock_disconnect_controller
.assert_called_once()
1656 def test_add_credential_exception(
1658 mock_add_credential
,
1660 mock_disconnect_controller
,
1661 mock_get_controller
,
1663 mock_get_controller
.return_value
= juju
.controller
.Controller()
1664 mock_add_credential
.side_effect
= Exception()
1665 with self
.assertRaises(Exception):
1666 self
.loop
.run_until_complete(
1667 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1670 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1671 mock_add_credential
.assert_called_once_with(
1672 "cloud", credential
=self
.credential
, cloud
="cloud"
1674 mock_disconnect_controller
.assert_called_once()
1677 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1678 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1679 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1680 class RemoveCloudTest(LibjujuTestCase
):
1682 super(RemoveCloudTest
, self
).setUp()
1684 def test_remove_cloud(
1687 mock_disconnect_controller
,
1688 mock_get_controller
,
1690 mock_get_controller
.return_value
= juju
.controller
.Controller()
1692 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1693 mock_remove_cloud
.assert_called_once_with("cloud")
1694 mock_disconnect_controller
.assert_called_once()
1696 def test_remove_cloud_exception(
1699 mock_disconnect_controller
,
1700 mock_get_controller
,
1702 mock_get_controller
.return_value
= juju
.controller
.Controller()
1703 mock_remove_cloud
.side_effect
= Exception()
1705 with self
.assertRaises(Exception):
1706 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1707 mock_remove_cloud
.assert_called_once_with("cloud")
1708 mock_disconnect_controller
.assert_called_once()
1711 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1712 class GetK8sCloudCredentials(LibjujuTestCase
):
1714 super(GetK8sCloudCredentials
, self
).setUp()
1715 self
.cert_data
= "cert"
1716 self
.token
= "token"
1718 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1719 def test_not_supported(self
, mock_exception
, mock_configuration
):
1720 mock_configuration
.username
= ""
1721 mock_configuration
.password
= ""
1722 mock_configuration
.ssl_ca_cert
= None
1723 mock_configuration
.cert_file
= None
1724 mock_configuration
.key_file
= None
1725 exception_raised
= False
1727 self
.cert_data
= None
1729 _
= self
.libjuju
.get_k8s_cloud_credential(
1734 except JujuInvalidK8sConfiguration
as e
:
1735 exception_raised
= True
1738 "authentication method not supported",
1740 self
.assertTrue(exception_raised
)
1742 def test_user_pass(self
, mock_configuration
):
1743 mock_configuration
.username
= "admin"
1744 mock_configuration
.password
= "admin"
1745 mock_configuration
.ssl_ca_cert
= None
1746 mock_configuration
.cert_file
= None
1747 mock_configuration
.key_file
= None
1749 self
.cert_data
= None
1750 credential
= self
.libjuju
.get_k8s_cloud_credential(
1757 juju
.client
._definitions
.CloudCredential(
1758 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1762 def test_user_pass_with_cert(self
, mock_configuration
):
1763 mock_configuration
.username
= "admin"
1764 mock_configuration
.password
= "admin"
1765 mock_configuration
.ssl_ca_cert
= None
1766 mock_configuration
.cert_file
= None
1767 mock_configuration
.key_file
= None
1769 credential
= self
.libjuju
.get_k8s_cloud_credential(
1776 juju
.client
._definitions
.CloudCredential(
1778 "ClientCertificateData": self
.cert_data
,
1779 "username": "admin",
1780 "password": "admin",
1782 auth_type
="userpasswithcert",
1786 def test_user_no_pass(self
, mock_configuration
):
1787 mock_configuration
.username
= "admin"
1788 mock_configuration
.password
= ""
1789 mock_configuration
.ssl_ca_cert
= None
1790 mock_configuration
.cert_file
= None
1791 mock_configuration
.key_file
= None
1793 self
.cert_data
= None
1794 with patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1795 credential
= self
.libjuju
.get_k8s_cloud_credential(
1802 juju
.client
._definitions
.CloudCredential(
1803 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1806 mock_debug
.assert_called_once_with(
1807 "credential for user admin has empty password"
1810 def test_cert(self
, mock_configuration
):
1811 mock_configuration
.username
= ""
1812 mock_configuration
.password
= ""
1813 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1814 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1815 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1816 ssl_ca_cert_file
.write("cacert")
1817 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1818 mock_configuration
.cert_file
= None
1819 mock_configuration
.key_file
= None
1820 credential
= self
.libjuju
.get_k8s_cloud_credential(
1827 juju
.client
._definitions
.CloudCredential(
1828 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1829 auth_type
="certificate",
1833 # TODO: Fix this test when oauth authentication is supported
1834 # def test_oauth2(self, mock_configuration):
1835 # mock_configuration.username = ""
1836 # mock_configuration.password = ""
1837 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1838 # key = tempfile.NamedTemporaryFile()
1839 # with open(key.name, "w") as key_file:
1840 # key_file.write("key")
1841 # mock_configuration.ssl_ca_cert = None
1842 # mock_configuration.cert_file = None
1843 # mock_configuration.key_file = key.name
1844 # credential = self.libjuju.get_k8s_cloud_credential(
1845 # mock_configuration,
1851 # juju.client._definitions.CloudCredential(
1852 # attrs={"ClientKeyData": "key", "Token": "Token"},
1853 # auth_type="oauth2",
1857 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1858 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1859 # mock_configuration.username = ""
1860 # mock_configuration.password = ""
1861 # key = tempfile.NamedTemporaryFile()
1862 # with open(key.name, "w") as key_file:
1863 # key_file.write("key")
1864 # mock_configuration.ssl_ca_cert = None
1865 # mock_configuration.cert_file = None
1866 # mock_configuration.key_file = key.name
1867 # exception_raised = False
1869 # _ = self.libjuju.get_k8s_cloud_credential(
1870 # mock_configuration,
1874 # except JujuInvalidK8sConfiguration as e:
1875 # exception_raised = True
1878 # "missing token for auth type oauth2",
1880 # self.assertTrue(exception_raised)
1882 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1883 mock_configuration
.username
= "admin"
1884 mock_configuration
.password
= "pass"
1885 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1886 mock_configuration
.ssl_ca_cert
= None
1887 mock_configuration
.cert_file
= None
1888 mock_configuration
.key_file
= None
1889 exception_raised
= False
1891 _
= self
.libjuju
.get_k8s_cloud_credential(
1896 except JujuInvalidK8sConfiguration
as e
:
1897 exception_raised
= True
1900 "Cannot set both token and user/pass",
1902 self
.assertTrue(exception_raised
)
1905 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1906 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1907 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1908 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1909 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1910 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
1911 class ScaleApplicationTest(LibjujuTestCase
):
1913 super(ScaleApplicationTest
, self
).setUp()
1915 @asynctest.mock
.patch("asyncio.sleep")
1916 def test_scale_application(
1919 mock_wait_for_model
,
1920 mock_disconnect_controller
,
1921 mock_disconnect_model
,
1922 mock_get_application
,
1924 mock_get_controller
,
1926 mock_get_model
.return_value
= juju
.model
.Model()
1927 mock_get_application
.return_value
= FakeApplication()
1928 self
.loop
.run_until_complete(self
.libjuju
.scale_application("model", "app", 2))
1929 mock_wait_for_model
.assert_called_once()
1930 mock_disconnect_controller
.assert_called_once()
1931 mock_disconnect_model
.assert_called_once()
1933 def test_no_application(
1936 mock_disconnect_controller
,
1937 mock_disconnect_model
,
1938 mock_get_application
,
1940 mock_get_controller
,
1942 mock_get_application
.return_value
= None
1943 mock_get_model
.return_value
= juju
.model
.Model()
1944 with self
.assertRaises(JujuApplicationNotFound
):
1945 self
.loop
.run_until_complete(
1946 self
.libjuju
.scale_application("model", "app", 2)
1948 mock_disconnect_controller
.assert_called()
1949 mock_disconnect_model
.assert_called()
1954 mock_disconnect_controller
,
1955 mock_disconnect_model
,
1956 mock_get_application
,
1958 mock_get_controller
,
1960 mock_get_model
.return_value
= None
1961 mock_get_application
.return_value
= FakeApplication()
1962 with self
.assertRaises(Exception):
1963 self
.loop
.run_until_complete(
1964 self
.libjuju
.scale_application("model", "app", 2, total_timeout
=0)
1966 mock_disconnect_controller
.assert_called_once()
1969 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1970 class GetUnitNumberTest(LibjujuTestCase
):
1972 super(GetUnitNumberTest
, self
).setUp()
1974 def test_successful_get_unit_number(
1976 mock_get_applications
,
1978 mock_get_applications
.return_value
= FakeApplication()
1979 model
= juju
.model
.Model()
1980 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1981 self
.assertEqual(result
, 2)
1983 def test_non_existing_application(
1985 mock_get_applications
,
1987 mock_get_applications
.return_value
= None
1988 model
= juju
.model
.Model()
1989 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1990 self
.assertEqual(result
, None)
1993 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
1994 class GetMachineInfoTest(LibjujuTestCase
):
1996 super(GetMachineInfoTest
, self
).setUp()
1998 def test_successful(
2002 machine_id
= "existing_machine"
2003 model
= juju
.model
.Model()
2004 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2005 machine
, series
= self
.libjuju
._get
_machine
_info
(
2006 machine_id
=machine_id
,
2009 self
.assertIsNotNone(machine
, series
)
2015 machine_id
= "not_existing_machine"
2016 machine
= series
= None
2017 model
= juju
.model
.Model()
2018 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2019 with self
.assertRaises(JujuMachineNotFound
):
2020 machine
, series
= self
.libjuju
._get
_machine
_info
(
2021 machine_id
=machine_id
,
2024 self
.assertIsNone(machine
, series
)
2027 class GetUnitTest(LibjujuTestCase
):
2029 super(GetUnitTest
, self
).setUp()
2031 def test_successful(self
):
2032 result
= self
.libjuju
._get
_unit
(FakeApplication(), "existing_machine_id")
2033 self
.assertIsInstance(result
, FakeUnit
)
2035 def test_return_none(self
):
2036 result
= self
.libjuju
._get
_unit
(FakeApplication(), "not_existing_machine_id")
2037 self
.assertIsNone(result
)
2040 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2041 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2042 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2043 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2044 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2045 class CheckApplicationExists(LibjujuTestCase
):
2047 super(CheckApplicationExists
, self
).setUp()
2049 def test_successful(
2051 mock_get_application
,
2052 mock_disconnect_controller
,
2053 mock_disconnect_model
,
2055 mock_get_controller
,
2057 mock_get_model
.return_value
= juju
.model
.Model()
2058 mock_get_application
.return_value
= FakeApplication()
2059 result
= self
.loop
.run_until_complete(
2060 self
.libjuju
.check_application_exists(
2065 self
.assertEqual(result
, True)
2067 mock_get_application
.assert_called_once()
2068 mock_get_controller
.assert_called_once()
2069 mock_get_model
.assert_called_once()
2070 mock_disconnect_controller
.assert_called_once()
2071 mock_disconnect_model
.assert_called_once()
2073 def test_no_application(
2075 mock_get_application
,
2076 mock_disconnect_controller
,
2077 mock_disconnect_model
,
2079 mock_get_controller
,
2081 mock_get_model
.return_value
= juju
.model
.Model()
2082 mock_get_application
.return_value
= None
2083 result
= self
.loop
.run_until_complete(
2084 self
.libjuju
.check_application_exists(
2089 self
.assertEqual(result
, False)
2091 mock_get_application
.assert_called_once()
2092 mock_get_controller
.assert_called_once()
2093 mock_get_model
.assert_called_once()
2094 mock_disconnect_controller
.assert_called_once()
2095 mock_disconnect_model
.assert_called_once()
2098 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2099 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2100 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2101 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2102 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2103 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_machine_info")
2104 class AddUnitTest(LibjujuTestCase
):
2106 super(AddUnitTest
, self
).setUp()
2108 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
2109 @asynctest.mock
.patch("asyncio.sleep")
2110 def test_successful(
2114 mock_get_machine_info
,
2115 mock_get_application
,
2116 mock_disconnect_controller
,
2117 mock_disconnect_model
,
2119 mock_get_controller
,
2121 mock_get_model
.return_value
= juju
.model
.Model()
2122 mock_get_application
.return_value
= FakeApplication()
2123 mock_get_machine_info
.return_value
= FakeMachine(), "series"
2124 self
.loop
.run_until_complete(
2125 self
.libjuju
.add_unit(
2132 mock_wait_for
.assert_called_once()
2133 mock_get_application
.assert_called_once()
2134 mock_get_controller
.assert_called_once()
2135 mock_get_model
.assert_called_once()
2136 mock_disconnect_controller
.assert_called_once()
2137 mock_disconnect_model
.assert_called_once()
2141 mock_get_machine_info
,
2142 mock_get_application
,
2143 mock_disconnect_controller
,
2144 mock_disconnect_model
,
2146 mock_get_controller
,
2148 mock_get_model
.return_value
= juju
.model
.Model()
2149 mock_get_application
.return_value
= None
2150 with self
.assertRaises(JujuApplicationNotFound
):
2151 self
.loop
.run_until_complete(
2152 self
.libjuju
.add_unit(
2159 mock_get_application
.assert_called_once()
2160 mock_get_controller
.assert_called_once()
2161 mock_get_model
.assert_called_once()
2162 mock_disconnect_controller
.assert_called_once()
2163 mock_disconnect_model
.assert_called_once()
2165 def test_no_machine(
2167 mock_get_machine_info
,
2168 mock_get_application
,
2169 mock_disconnect_controller
,
2170 mock_disconnect_model
,
2172 mock_get_controller
,
2174 mock_get_model
.return_value
= juju
.model
.Model()
2175 mock_get_application
.return_value
= FakeApplication()
2176 mock_get_machine_info
.side_effect
= JujuMachineNotFound()
2177 with self
.assertRaises(JujuMachineNotFound
):
2178 self
.loop
.run_until_complete(
2179 self
.libjuju
.add_unit(
2186 mock_get_application
.assert_called_once()
2187 mock_get_controller
.assert_called_once()
2188 mock_get_model
.assert_called_once()
2189 mock_disconnect_controller
.assert_called_once()
2190 mock_disconnect_model
.assert_called_once()
2193 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2194 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2195 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2196 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2197 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2198 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_unit")
2199 class DestroyUnitTest(LibjujuTestCase
):
2201 super(DestroyUnitTest
, self
).setUp()
2203 @asynctest.mock
.patch("asyncio.sleep")
2204 def test_successful(
2208 mock_get_application
,
2209 mock_disconnect_controller
,
2210 mock_disconnect_model
,
2212 mock_get_controller
,
2214 mock_get_model
.return_value
= juju
.model
.Model()
2215 mock_get_application
.return_value
= FakeApplication()
2217 self
.loop
.run_until_complete(
2218 self
.libjuju
.destroy_unit("app", "model", "machine", 0)
2221 mock_get_unit
.assert_called()
2222 mock_get_application
.assert_called_once()
2223 mock_get_controller
.assert_called_once()
2224 mock_get_model
.assert_called_once()
2225 mock_disconnect_controller
.assert_called_once()
2226 mock_disconnect_model
.assert_called_once()
2231 mock_get_application
,
2232 mock_disconnect_controller
,
2233 mock_disconnect_model
,
2235 mock_get_controller
,
2237 mock_get_model
.return_value
= juju
.model
.Model()
2238 mock_get_application
.return_value
= None
2240 with self
.assertRaises(JujuApplicationNotFound
):
2241 self
.loop
.run_until_complete(
2242 self
.libjuju
.destroy_unit("app", "model", "machine")
2245 mock_get_application
.assert_called_once()
2246 mock_get_controller
.assert_called_once()
2247 mock_get_model
.assert_called_once()
2248 mock_disconnect_controller
.assert_called_once()
2249 mock_disconnect_model
.assert_called_once()
2254 mock_get_application
,
2255 mock_disconnect_controller
,
2256 mock_disconnect_model
,
2258 mock_get_controller
,
2260 mock_get_model
.return_value
= juju
.model
.Model()
2261 mock_get_application
.return_value
= FakeApplication()
2262 mock_get_unit
.return_value
= None
2264 with self
.assertRaises(JujuError
):
2265 self
.loop
.run_until_complete(
2266 self
.libjuju
.destroy_unit("app", "model", "machine")
2269 mock_get_unit
.assert_called_once()
2270 mock_get_application
.assert_called_once()
2271 mock_get_controller
.assert_called_once()
2272 mock_get_model
.assert_called_once()
2273 mock_disconnect_controller
.assert_called_once()
2274 mock_disconnect_model
.assert_called_once()