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
,
779 mock_get_model
.return_value
= juju
.model
.Model()
780 mock__get_application
.return_value
= FakeApplication()
783 with self
.assertRaises(JujuActionNotFound
):
784 output
, status
= self
.loop
.run_until_complete(
785 self
.libjuju
.execute_action(
791 self
.assertIsNone(output
)
792 self
.assertIsNone(status
)
794 mock_disconnect_controller
.assert_called()
795 mock_disconnect_model
.assert_called()
797 @asynctest.mock
.patch("asyncio.sleep")
798 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
801 mock_is_leader_from_status
,
803 mock_get_action_status
,
804 mock_get_action_output
,
806 mock__get_application
,
807 mock_disconnect_controller
,
808 mock_disconnect_model
,
812 mock_get_model
.return_value
= juju
.model
.Model()
813 mock__get_application
.return_value
= FakeApplication()
814 mock_is_leader_from_status
.return_value
= False
817 with self
.assertRaises(JujuLeaderUnitNotFound
):
818 output
, status
= self
.loop
.run_until_complete(
819 self
.libjuju
.execute_action(
825 self
.assertIsNone(output
)
826 self
.assertIsNone(status
)
828 mock_disconnect_controller
.assert_called()
829 mock_disconnect_model
.assert_called()
831 def test_successful_exec(
833 mock_get_action_status
,
834 mock_get_action_output
,
836 mock__get_application
,
837 mock_disconnect_controller
,
838 mock_disconnect_model
,
842 mock_get_model
.return_value
= juju
.model
.Model()
843 mock__get_application
.return_value
= FakeApplication()
844 mock_get_action_output
.return_value
= "output"
845 mock_get_action_status
.return_value
= {"id": "status"}
846 output
, status
= self
.loop
.run_until_complete(
847 self
.libjuju
.execute_action("app", "model", "existing_action")
849 self
.assertEqual(output
, "output")
850 self
.assertEqual(status
, "status")
852 mock_wait_for
.assert_called_once()
854 mock_disconnect_controller
.assert_called()
855 mock_disconnect_model
.assert_called()
858 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
859 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
860 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
861 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
862 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
863 class GetActionTest(LibjujuTestCase
):
865 super(GetActionTest
, self
).setUp()
869 mock_get_application
,
870 mock_disconnect_controller
,
871 mock_disconnect_model
,
875 mock_get_application
.side_effect
= Exception()
877 with self
.assertRaises(Exception):
878 actions
= self
.loop
.run_until_complete(
879 self
.libjuju
.get_actions("app", "model")
882 self
.assertIsNone(actions
)
883 mock_disconnect_controller
.assert_called_once()
884 mock_disconnect_model
.assert_called_once()
888 mock_get_application
,
889 mock_disconnect_controller
,
890 mock_disconnect_model
,
894 mock_get_application
.return_value
= FakeApplication()
896 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
898 self
.assertEqual(actions
, ["existing_action"])
900 mock_get_controller
.assert_called_once()
901 mock_get_model
.assert_called_once()
902 mock_disconnect_controller
.assert_called_once()
903 mock_disconnect_model
.assert_called_once()
906 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
907 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
908 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
909 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
910 @asynctest.mock
.patch("juju.application.Application.get_metrics")
911 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
912 class GetMetricsTest(LibjujuTestCase
):
914 super(GetMetricsTest
, self
).setUp()
916 def test_get_metrics_success(
918 mock_get_application
,
920 mock_disconnect_controller
,
921 mock_disconnect_model
,
925 mock_get_application
.return_value
= FakeApplication()
926 mock_get_model
.return_value
= juju
.model
.Model()
928 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
930 mock_disconnect_controller
.assert_called_once()
931 mock_disconnect_model
.assert_called_once()
933 def test_get_metrics_exception(
935 mock_get_application
,
937 mock_disconnect_controller
,
938 mock_disconnect_model
,
942 mock_get_model
.return_value
= juju
.model
.Model()
943 mock_get_metrics
.side_effect
= Exception()
944 with self
.assertRaises(Exception):
945 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
947 mock_disconnect_controller
.assert_called_once()
948 mock_disconnect_model
.assert_called_once()
950 def test_missing_args_exception(
952 mock_get_application
,
954 mock_disconnect_controller
,
955 mock_disconnect_model
,
959 mock_get_model
.return_value
= juju
.model
.Model()
961 with self
.assertRaises(Exception):
962 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
964 mock_get_controller
.assert_not_called()
965 mock_get_model
.assert_not_called()
966 mock_disconnect_controller
.assert_not_called()
967 mock_disconnect_model
.assert_not_called()
970 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
971 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
972 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
973 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
974 @asynctest.mock
.patch("juju.model.Model.add_relation")
975 class AddRelationTest(LibjujuTestCase
):
977 super(AddRelationTest
, self
).setUp()
979 @asynctest.mock
.patch("logging.Logger.warning")
984 mock_disconnect_controller
,
985 mock_disconnect_model
,
989 # TODO in libjuju.py should this fail only with a log message?
990 result
= {"error": "not found", "response": "response", "request-id": 1}
992 mock_get_model
.return_value
= juju
.model
.Model()
993 mock_add_relation
.side_effect
= JujuAPIError(result
)
995 self
.loop
.run_until_complete(
996 self
.libjuju
.add_relation(
1003 mock_warning
.assert_called_with("Relation not found: not found")
1004 mock_disconnect_controller
.assert_called_once()
1005 mock_disconnect_model
.assert_called_once()
1007 @asynctest.mock
.patch("logging.Logger.warning")
1008 def test_already_exists(
1012 mock_disconnect_controller
,
1013 mock_disconnect_model
,
1015 mock_get_controller
,
1017 # TODO in libjuju.py should this fail silently?
1018 result
= {"error": "already exists", "response": "response", "request-id": 1}
1020 mock_get_model
.return_value
= juju
.model
.Model()
1021 mock_add_relation
.side_effect
= JujuAPIError(result
)
1023 self
.loop
.run_until_complete(
1024 self
.libjuju
.add_relation(
1031 mock_warning
.assert_called_with("Relation already exists: already exists")
1032 mock_disconnect_controller
.assert_called_once()
1033 mock_disconnect_model
.assert_called_once()
1038 mock_disconnect_controller
,
1039 mock_disconnect_model
,
1041 mock_get_controller
,
1043 mock_get_model
.return_value
= juju
.model
.Model()
1044 result
= {"error": "", "response": "response", "request-id": 1}
1045 mock_add_relation
.side_effect
= JujuAPIError(result
)
1047 with self
.assertRaises(JujuAPIError
):
1048 self
.loop
.run_until_complete(
1049 self
.libjuju
.add_relation(
1056 mock_disconnect_controller
.assert_called_once()
1057 mock_disconnect_model
.assert_called_once()
1062 mock_disconnect_controller
,
1063 mock_disconnect_model
,
1065 mock_get_controller
,
1067 mock_get_model
.return_value
= juju
.model
.Model()
1069 self
.loop
.run_until_complete(
1070 self
.libjuju
.add_relation(
1077 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1078 mock_disconnect_controller
.assert_called_once()
1079 mock_disconnect_model
.assert_called_once()
1084 mock_disconnect_controller
,
1085 mock_disconnect_model
,
1087 mock_get_controller
,
1089 mock_get_model
.return_value
= juju
.model
.Model()
1091 self
.loop
.run_until_complete(
1092 self
.libjuju
.add_relation(
1099 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1100 mock_disconnect_controller
.assert_called_once()
1101 mock_disconnect_model
.assert_called_once()
1104 # TODO destroy_model testcase
1107 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1108 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1109 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1112 class DestroyApplicationTest(LibjujuTestCase
):
1114 super(DestroyApplicationTest
, self
).setUp()
1118 mock_get_controller
,
1120 mock_disconnect_controller
,
1121 mock_get_application
,
1122 mock_disconnect_model
,
1124 mock_get_application
.return_value
= FakeApplication()
1125 mock_get_model
.return_value
= None
1126 self
.loop
.run_until_complete(
1127 self
.libjuju
.destroy_application(
1133 mock_get_application
.assert_called()
1134 mock_disconnect_controller
.assert_called_once()
1135 mock_disconnect_model
.assert_called_once()
1137 def test_no_application(
1139 mock_get_controller
,
1141 mock_disconnect_controller
,
1142 mock_get_application
,
1143 mock_disconnect_model
,
1145 mock_get_model
.return_value
= None
1146 mock_get_application
.return_value
= None
1148 self
.loop
.run_until_complete(
1149 self
.libjuju
.destroy_application(
1155 mock_get_application
.assert_called()
1159 mock_get_controller
,
1161 mock_disconnect_controller
,
1162 mock_get_application
,
1163 mock_disconnect_model
,
1165 mock_get_application
.return_value
= FakeApplication
1166 mock_get_model
.return_value
= None
1168 with self
.assertRaises(Exception):
1169 self
.loop
.run_until_complete(
1170 self
.libjuju
.destroy_application(
1176 mock_get_application
.assert_called_once()
1179 # @asynctest.mock.patch("juju.model.Model.get_machines")
1180 # @asynctest.mock.patch("logging.Logger.debug")
1181 # class DestroyMachineTest(LibjujuTestCase):
1183 # super(DestroyMachineTest, self).setUp()
1185 # def test_success_manual_machine(
1186 # self, mock_debug, mock_get_machines,
1188 # mock_get_machines.side_effect = [
1189 # {"machine": FakeManualMachine()},
1190 # {"machine": FakeManualMachine()},
1193 # self.loop.run_until_complete(
1194 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1197 # asynctest.call("Waiting for machine machine is destroyed"),
1198 # asynctest.call("Machine destroyed: machine"),
1200 # mock_debug.assert_has_calls(calls)
1202 # def test_no_machine(
1203 # self, mock_debug, mock_get_machines,
1205 # mock_get_machines.return_value = {}
1206 # self.loop.run_until_complete(
1207 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1209 # mock_debug.assert_called_with("Machine not found: machine")
1212 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1213 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1214 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1215 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1216 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1217 class ConfigureApplicationTest(LibjujuTestCase
):
1219 super(ConfigureApplicationTest
, self
).setUp()
1223 mock_get_application
,
1224 mock_disconnect_controller
,
1225 mock_disconnect_model
,
1227 mock_get_controller
,
1229 mock_get_application
.return_value
= FakeApplication()
1231 self
.loop
.run_until_complete(
1232 self
.libjuju
.configure_application(
1238 mock_get_application
.assert_called_once()
1239 mock_disconnect_controller
.assert_called_once()
1240 mock_disconnect_model
.assert_called_once()
1244 mock_get_application
,
1245 mock_disconnect_controller
,
1246 mock_disconnect_model
,
1248 mock_get_controller
,
1250 mock_get_application
.side_effect
= Exception()
1252 with self
.assertRaises(Exception):
1253 self
.loop
.run_until_complete(
1254 self
.libjuju
.configure_application(
1260 mock_disconnect_controller
.assert_called_once()
1261 mock_disconnect_model
.assert_called_once()
1263 def test_controller_exception(
1265 mock_get_application
,
1266 mock_disconnect_controller
,
1267 mock_disconnect_model
,
1269 mock_get_controller
,
1271 result
= {"error": "not found", "response": "response", "request-id": 1}
1273 mock_get_controller
.side_effect
= JujuAPIError(result
)
1275 with self
.assertRaises(JujuAPIError
):
1276 self
.loop
.run_until_complete(
1277 self
.libjuju
.configure_application(
1283 mock_get_model
.assert_not_called()
1284 mock_disconnect_controller
.assert_not_called()
1285 mock_disconnect_model
.assert_not_called()
1287 def test_get_model_exception(
1289 mock_get_application
,
1290 mock_disconnect_controller
,
1291 mock_disconnect_model
,
1293 mock_get_controller
,
1295 result
= {"error": "not found", "response": "response", "request-id": 1}
1296 mock_get_model
.side_effect
= JujuAPIError(result
)
1298 with self
.assertRaises(JujuAPIError
):
1299 self
.loop
.run_until_complete(
1300 self
.libjuju
.configure_application(
1306 mock_get_model
.assert_called_once()
1307 mock_disconnect_controller
.assert_called_once()
1308 mock_disconnect_model
.assert_not_called()
1311 # TODO _get_api_endpoints_db test case
1312 # TODO _update_api_endpoints_db test case
1313 # TODO healthcheck test case
1316 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1317 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1318 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1319 class ListModelsTest(LibjujuTestCase
):
1321 super(ListModelsTest
, self
).setUp()
1323 def test_containing(
1326 mock_disconnect_controller
,
1327 mock_get_controller
,
1329 mock_get_controller
.return_value
= juju
.controller
.Controller()
1330 mock_list_models
.return_value
= ["existingmodel"]
1331 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1333 mock_disconnect_controller
.assert_called_once()
1334 self
.assertEquals(models
, ["existingmodel"])
1336 def test_not_containing(
1339 mock_disconnect_controller
,
1340 mock_get_controller
,
1342 mock_get_controller
.return_value
= juju
.controller
.Controller()
1343 mock_list_models
.return_value
= ["existingmodel", "model"]
1344 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1346 mock_disconnect_controller
.assert_called_once()
1347 self
.assertEquals(models
, [])
1349 def test_no_contains_arg(
1352 mock_disconnect_controller
,
1353 mock_get_controller
,
1355 mock_get_controller
.return_value
= juju
.controller
.Controller()
1356 mock_list_models
.return_value
= ["existingmodel", "model"]
1357 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1359 mock_disconnect_controller
.assert_called_once()
1360 self
.assertEquals(models
, ["existingmodel", "model"])
1363 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1364 class ModelsExistTest(LibjujuTestCase
):
1366 super(ModelsExistTest
, self
).setUp()
1368 def test_model_names_none(self
, mock_list_models
):
1369 mock_list_models
.return_value
= []
1370 with self
.assertRaises(Exception):
1371 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1373 def test_model_names_empty(self
, mock_list_models
):
1374 mock_list_models
.return_value
= []
1375 with self
.assertRaises(Exception):
1376 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1377 self
.libjuju
.models_exist([])
1380 def test_model_names_not_existing(self
, mock_list_models
):
1381 mock_list_models
.return_value
= ["prometheus", "grafana"]
1382 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1383 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1385 self
.assertFalse(exist
)
1386 self
.assertEqual(non_existing_models
, ["prometheus2"])
1388 def test_model_names_exist(self
, mock_list_models
):
1389 mock_list_models
.return_value
= ["prometheus", "grafana"]
1390 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1391 self
.libjuju
.models_exist(["prometheus", "grafana"])
1393 self
.assertTrue(exist
)
1394 self
.assertEqual(non_existing_models
, [])
1397 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1398 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1399 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1400 class ListOffers(LibjujuTestCase
):
1402 super(ListOffers
, self
).setUp()
1404 def test_disconnect_controller(
1407 mock_disconnect_controller
,
1408 mock_get_controller
,
1410 mock_get_controller
.return_value
= juju
.controller
.Controller()
1411 mock_list_offers
.side_effect
= Exception()
1412 with self
.assertRaises(Exception):
1413 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1414 mock_disconnect_controller
.assert_called_once()
1416 def test_empty_list(
1419 mock_disconnect_controller
,
1420 mock_get_controller
,
1422 mock_get_controller
.return_value
= juju
.controller
.Controller()
1423 mock_list_offers
.return_value
= []
1424 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1425 self
.assertEqual(offers
, [])
1426 mock_disconnect_controller
.assert_called_once()
1428 def test_non_empty_list(
1431 mock_disconnect_controller
,
1432 mock_get_controller
,
1434 mock_get_controller
.return_value
= juju
.controller
.Controller()
1435 mock_list_offers
.return_value
= ["offer"]
1436 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1437 self
.assertEqual(offers
, ["offer"])
1438 mock_disconnect_controller
.assert_called_once()
1441 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1442 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1443 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1444 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1445 @asynctest.mock
.patch("juju.model.Model.consume")
1446 class ConsumeTest(LibjujuTestCase
):
1448 super(ConsumeTest
, self
).setUp()
1453 mock_disconnect_controller
,
1454 mock_disconnect_model
,
1456 mock_get_controller
,
1458 mock_get_controller
.return_value
= juju
.controller
.Controller()
1459 mock_get_model
.return_value
= juju
.model
.Model()
1461 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1462 mock_consume
.assert_called_once()
1463 mock_disconnect_model
.assert_called_once()
1464 mock_disconnect_controller
.assert_called_once()
1466 def test_parsing_error_exception(
1469 mock_disconnect_controller
,
1470 mock_disconnect_model
,
1472 mock_get_controller
,
1474 mock_get_controller
.return_value
= juju
.controller
.Controller()
1475 mock_get_model
.return_value
= juju
.model
.Model()
1476 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1478 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1479 self
.loop
.run_until_complete(
1480 self
.libjuju
.consume("offer_url", "model_name")
1482 mock_consume
.assert_called_once()
1483 mock_disconnect_model
.assert_called_once()
1484 mock_disconnect_controller
.assert_called_once()
1486 def test_juju_error_exception(
1489 mock_disconnect_controller
,
1490 mock_disconnect_model
,
1492 mock_get_controller
,
1494 mock_get_controller
.return_value
= juju
.controller
.Controller()
1495 mock_get_model
.return_value
= juju
.model
.Model()
1496 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1498 with self
.assertRaises(juju
.errors
.JujuError
):
1499 self
.loop
.run_until_complete(
1500 self
.libjuju
.consume("offer_url", "model_name")
1502 mock_consume
.assert_called_once()
1503 mock_disconnect_model
.assert_called_once()
1504 mock_disconnect_controller
.assert_called_once()
1506 def test_juju_api_error_exception(
1509 mock_disconnect_controller
,
1510 mock_disconnect_model
,
1512 mock_get_controller
,
1514 mock_get_controller
.return_value
= juju
.controller
.Controller()
1515 mock_get_model
.return_value
= juju
.model
.Model()
1516 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1517 {"error": "", "response": "", "request-id": ""}
1520 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1521 self
.loop
.run_until_complete(
1522 self
.libjuju
.consume("offer_url", "model_name")
1524 mock_consume
.assert_called_once()
1525 mock_disconnect_model
.assert_called_once()
1526 mock_disconnect_controller
.assert_called_once()
1529 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1530 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1531 class AddK8sTest(LibjujuTestCase
):
1533 super(AddK8sTest
, self
).setUp()
1535 rbac_id
= generate_rbac_id()
1537 client_cert_data
= "cert"
1538 configuration
= kubernetes
.client
.configuration
.Configuration()
1539 storage_class
= "storage_class"
1540 credential_name
= name
1542 self
._add
_k
8s
_args
= {
1546 "client_cert_data": client_cert_data
,
1547 "configuration": configuration
,
1548 "storage_class": storage_class
,
1549 "credential_name": credential_name
,
1552 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1553 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1554 mock_add_cloud
.assert_called_once()
1555 mock_get_k8s_cloud_credential
.assert_called_once()
1557 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1558 mock_add_cloud
.side_effect
= Exception()
1559 with self
.assertRaises(Exception):
1560 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1561 mock_add_cloud
.assert_called_once()
1562 mock_get_k8s_cloud_credential
.assert_called_once()
1564 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1565 self
._add
_k
8s
_args
["name"] = ""
1566 with self
.assertRaises(Exception):
1567 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1568 mock_add_cloud
.assert_not_called()
1570 def test_add_k8s_missing_storage_name(
1571 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1573 self
._add
_k
8s
_args
["storage_class"] = ""
1574 with self
.assertRaises(Exception):
1575 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1576 mock_add_cloud
.assert_not_called()
1578 def test_add_k8s_missing_configuration_keys(
1579 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1581 self
._add
_k
8s
_args
["configuration"] = None
1582 with self
.assertRaises(Exception):
1583 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1584 mock_add_cloud
.assert_not_called()
1587 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1588 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1589 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1590 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1591 class AddCloudTest(LibjujuTestCase
):
1593 super(AddCloudTest
, self
).setUp()
1594 self
.cloud
= juju
.client
.client
.Cloud()
1595 self
.credential
= juju
.client
.client
.CloudCredential()
1597 def test_add_cloud_with_credential(
1599 mock_add_credential
,
1601 mock_disconnect_controller
,
1602 mock_get_controller
,
1604 mock_get_controller
.return_value
= juju
.controller
.Controller()
1606 cloud
= self
.loop
.run_until_complete(
1607 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1609 self
.assertEqual(cloud
, self
.cloud
)
1610 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1611 mock_add_credential
.assert_called_once_with(
1612 "cloud", credential
=self
.credential
, cloud
="cloud"
1614 mock_disconnect_controller
.assert_called_once()
1616 def test_add_cloud_no_credential(
1618 mock_add_credential
,
1620 mock_disconnect_controller
,
1621 mock_get_controller
,
1623 mock_get_controller
.return_value
= juju
.controller
.Controller()
1625 cloud
= self
.loop
.run_until_complete(
1626 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1628 self
.assertEqual(cloud
, self
.cloud
)
1629 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1630 mock_add_credential
.assert_not_called()
1631 mock_disconnect_controller
.assert_called_once()
1633 def test_add_cloud_exception(
1635 mock_add_credential
,
1637 mock_disconnect_controller
,
1638 mock_get_controller
,
1640 mock_get_controller
.return_value
= juju
.controller
.Controller()
1641 mock_add_cloud
.side_effect
= Exception()
1642 with self
.assertRaises(Exception):
1643 self
.loop
.run_until_complete(
1644 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1647 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1648 mock_add_credential
.assert_not_called()
1649 mock_disconnect_controller
.assert_called_once()
1651 def test_add_credential_exception(
1653 mock_add_credential
,
1655 mock_disconnect_controller
,
1656 mock_get_controller
,
1658 mock_get_controller
.return_value
= juju
.controller
.Controller()
1659 mock_add_credential
.side_effect
= Exception()
1660 with self
.assertRaises(Exception):
1661 self
.loop
.run_until_complete(
1662 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1665 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1666 mock_add_credential
.assert_called_once_with(
1667 "cloud", credential
=self
.credential
, cloud
="cloud"
1669 mock_disconnect_controller
.assert_called_once()
1672 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1673 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1674 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1675 class RemoveCloudTest(LibjujuTestCase
):
1677 super(RemoveCloudTest
, self
).setUp()
1679 def test_remove_cloud(
1682 mock_disconnect_controller
,
1683 mock_get_controller
,
1685 mock_get_controller
.return_value
= juju
.controller
.Controller()
1687 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1688 mock_remove_cloud
.assert_called_once_with("cloud")
1689 mock_disconnect_controller
.assert_called_once()
1691 def test_remove_cloud_exception(
1694 mock_disconnect_controller
,
1695 mock_get_controller
,
1697 mock_get_controller
.return_value
= juju
.controller
.Controller()
1698 mock_remove_cloud
.side_effect
= Exception()
1700 with self
.assertRaises(Exception):
1701 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1702 mock_remove_cloud
.assert_called_once_with("cloud")
1703 mock_disconnect_controller
.assert_called_once()
1706 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1707 class GetK8sCloudCredentials(LibjujuTestCase
):
1709 super(GetK8sCloudCredentials
, self
).setUp()
1710 self
.cert_data
= "cert"
1711 self
.token
= "token"
1713 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1714 def test_not_supported(self
, mock_exception
, mock_configuration
):
1715 mock_configuration
.username
= ""
1716 mock_configuration
.password
= ""
1717 mock_configuration
.ssl_ca_cert
= None
1718 mock_configuration
.cert_file
= None
1719 mock_configuration
.key_file
= None
1720 exception_raised
= False
1722 self
.cert_data
= None
1724 _
= self
.libjuju
.get_k8s_cloud_credential(
1729 except JujuInvalidK8sConfiguration
as e
:
1730 exception_raised
= True
1733 "authentication method not supported",
1735 self
.assertTrue(exception_raised
)
1737 def test_user_pass(self
, mock_configuration
):
1738 mock_configuration
.username
= "admin"
1739 mock_configuration
.password
= "admin"
1740 mock_configuration
.ssl_ca_cert
= None
1741 mock_configuration
.cert_file
= None
1742 mock_configuration
.key_file
= None
1744 self
.cert_data
= None
1745 credential
= self
.libjuju
.get_k8s_cloud_credential(
1752 juju
.client
._definitions
.CloudCredential(
1753 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1757 def test_user_pass_with_cert(self
, mock_configuration
):
1758 mock_configuration
.username
= "admin"
1759 mock_configuration
.password
= "admin"
1760 mock_configuration
.ssl_ca_cert
= None
1761 mock_configuration
.cert_file
= None
1762 mock_configuration
.key_file
= None
1764 credential
= self
.libjuju
.get_k8s_cloud_credential(
1771 juju
.client
._definitions
.CloudCredential(
1773 "ClientCertificateData": self
.cert_data
,
1774 "username": "admin",
1775 "password": "admin",
1777 auth_type
="userpasswithcert",
1781 def test_user_no_pass(self
, mock_configuration
):
1782 mock_configuration
.username
= "admin"
1783 mock_configuration
.password
= ""
1784 mock_configuration
.ssl_ca_cert
= None
1785 mock_configuration
.cert_file
= None
1786 mock_configuration
.key_file
= None
1788 self
.cert_data
= None
1789 with patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1790 credential
= self
.libjuju
.get_k8s_cloud_credential(
1797 juju
.client
._definitions
.CloudCredential(
1798 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1801 mock_debug
.assert_called_once_with(
1802 "credential for user admin has empty password"
1805 def test_cert(self
, mock_configuration
):
1806 mock_configuration
.username
= ""
1807 mock_configuration
.password
= ""
1808 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1809 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1810 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1811 ssl_ca_cert_file
.write("cacert")
1812 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1813 mock_configuration
.cert_file
= None
1814 mock_configuration
.key_file
= None
1815 credential
= self
.libjuju
.get_k8s_cloud_credential(
1822 juju
.client
._definitions
.CloudCredential(
1823 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1824 auth_type
="certificate",
1828 # TODO: Fix this test when oauth authentication is supported
1829 # def test_oauth2(self, mock_configuration):
1830 # mock_configuration.username = ""
1831 # mock_configuration.password = ""
1832 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1833 # key = tempfile.NamedTemporaryFile()
1834 # with open(key.name, "w") as key_file:
1835 # key_file.write("key")
1836 # mock_configuration.ssl_ca_cert = None
1837 # mock_configuration.cert_file = None
1838 # mock_configuration.key_file = key.name
1839 # credential = self.libjuju.get_k8s_cloud_credential(
1840 # mock_configuration,
1846 # juju.client._definitions.CloudCredential(
1847 # attrs={"ClientKeyData": "key", "Token": "Token"},
1848 # auth_type="oauth2",
1852 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1853 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1854 # mock_configuration.username = ""
1855 # mock_configuration.password = ""
1856 # key = tempfile.NamedTemporaryFile()
1857 # with open(key.name, "w") as key_file:
1858 # key_file.write("key")
1859 # mock_configuration.ssl_ca_cert = None
1860 # mock_configuration.cert_file = None
1861 # mock_configuration.key_file = key.name
1862 # exception_raised = False
1864 # _ = self.libjuju.get_k8s_cloud_credential(
1865 # mock_configuration,
1869 # except JujuInvalidK8sConfiguration as e:
1870 # exception_raised = True
1873 # "missing token for auth type oauth2",
1875 # self.assertTrue(exception_raised)
1877 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1878 mock_configuration
.username
= "admin"
1879 mock_configuration
.password
= "pass"
1880 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1881 mock_configuration
.ssl_ca_cert
= None
1882 mock_configuration
.cert_file
= None
1883 mock_configuration
.key_file
= None
1884 exception_raised
= False
1886 _
= self
.libjuju
.get_k8s_cloud_credential(
1891 except JujuInvalidK8sConfiguration
as e
:
1892 exception_raised
= True
1895 "Cannot set both token and user/pass",
1897 self
.assertTrue(exception_raised
)
1900 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1901 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1902 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1903 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1904 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1905 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
1906 class ScaleApplicationTest(LibjujuTestCase
):
1908 super(ScaleApplicationTest
, self
).setUp()
1910 @asynctest.mock
.patch("asyncio.sleep")
1911 def test_scale_application(
1914 mock_wait_for_model
,
1915 mock_disconnect_controller
,
1916 mock_disconnect_model
,
1917 mock_get_application
,
1919 mock_get_controller
,
1921 mock_get_model
.return_value
= juju
.model
.Model()
1922 mock_get_application
.return_value
= FakeApplication()
1923 self
.loop
.run_until_complete(self
.libjuju
.scale_application("model", "app", 2))
1924 mock_wait_for_model
.assert_called_once()
1925 mock_disconnect_controller
.assert_called_once()
1926 mock_disconnect_model
.assert_called_once()
1928 def test_no_application(
1931 mock_disconnect_controller
,
1932 mock_disconnect_model
,
1933 mock_get_application
,
1935 mock_get_controller
,
1937 mock_get_application
.return_value
= None
1938 mock_get_model
.return_value
= juju
.model
.Model()
1939 with self
.assertRaises(JujuApplicationNotFound
):
1940 self
.loop
.run_until_complete(
1941 self
.libjuju
.scale_application("model", "app", 2)
1943 mock_disconnect_controller
.assert_called()
1944 mock_disconnect_model
.assert_called()
1949 mock_disconnect_controller
,
1950 mock_disconnect_model
,
1951 mock_get_application
,
1953 mock_get_controller
,
1955 mock_get_model
.return_value
= None
1956 mock_get_application
.return_value
= FakeApplication()
1957 with self
.assertRaises(Exception):
1958 self
.loop
.run_until_complete(
1959 self
.libjuju
.scale_application("model", "app", 2, total_timeout
=0)
1961 mock_disconnect_controller
.assert_called_once()
1964 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1965 class GetUnitNumberTest(LibjujuTestCase
):
1967 super(GetUnitNumberTest
, self
).setUp()
1969 def test_successful_get_unit_number(
1971 mock_get_applications
,
1973 mock_get_applications
.return_value
= FakeApplication()
1974 model
= juju
.model
.Model()
1975 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1976 self
.assertEqual(result
, 2)
1978 def test_non_existing_application(
1980 mock_get_applications
,
1982 mock_get_applications
.return_value
= None
1983 model
= juju
.model
.Model()
1984 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1985 self
.assertEqual(result
, None)
1988 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
1989 class GetMachineInfoTest(LibjujuTestCase
):
1991 super(GetMachineInfoTest
, self
).setUp()
1993 def test_successful(
1997 machine_id
= "existing_machine"
1998 model
= juju
.model
.Model()
1999 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2000 machine
, series
= self
.libjuju
._get
_machine
_info
(
2001 machine_id
=machine_id
,
2004 self
.assertIsNotNone(machine
, series
)
2010 machine_id
= "not_existing_machine"
2011 machine
= series
= None
2012 model
= juju
.model
.Model()
2013 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2014 with self
.assertRaises(JujuMachineNotFound
):
2015 machine
, series
= self
.libjuju
._get
_machine
_info
(
2016 machine_id
=machine_id
,
2019 self
.assertIsNone(machine
, series
)
2022 class GetUnitTest(LibjujuTestCase
):
2024 super(GetUnitTest
, self
).setUp()
2026 def test_successful(self
):
2027 result
= self
.libjuju
._get
_unit
(FakeApplication(), "existing_machine_id")
2028 self
.assertIsInstance(result
, FakeUnit
)
2030 def test_return_none(self
):
2031 result
= self
.libjuju
._get
_unit
(FakeApplication(), "not_existing_machine_id")
2032 self
.assertIsNone(result
)
2035 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2036 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2037 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2038 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2039 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2040 class CheckApplicationExists(LibjujuTestCase
):
2042 super(CheckApplicationExists
, self
).setUp()
2044 def test_successful(
2046 mock_get_application
,
2047 mock_disconnect_controller
,
2048 mock_disconnect_model
,
2050 mock_get_controller
,
2052 mock_get_model
.return_value
= juju
.model
.Model()
2053 mock_get_application
.return_value
= FakeApplication()
2054 result
= self
.loop
.run_until_complete(
2055 self
.libjuju
.check_application_exists(
2060 self
.assertEqual(result
, True)
2062 mock_get_application
.assert_called_once()
2063 mock_get_controller
.assert_called_once()
2064 mock_get_model
.assert_called_once()
2065 mock_disconnect_controller
.assert_called_once()
2066 mock_disconnect_model
.assert_called_once()
2068 def test_no_application(
2070 mock_get_application
,
2071 mock_disconnect_controller
,
2072 mock_disconnect_model
,
2074 mock_get_controller
,
2076 mock_get_model
.return_value
= juju
.model
.Model()
2077 mock_get_application
.return_value
= None
2078 result
= self
.loop
.run_until_complete(
2079 self
.libjuju
.check_application_exists(
2084 self
.assertEqual(result
, False)
2086 mock_get_application
.assert_called_once()
2087 mock_get_controller
.assert_called_once()
2088 mock_get_model
.assert_called_once()
2089 mock_disconnect_controller
.assert_called_once()
2090 mock_disconnect_model
.assert_called_once()
2093 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2094 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2095 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2096 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2097 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2098 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_machine_info")
2099 class AddUnitTest(LibjujuTestCase
):
2101 super(AddUnitTest
, self
).setUp()
2103 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
2104 @asynctest.mock
.patch("asyncio.sleep")
2105 def test_successful(
2109 mock_get_machine_info
,
2110 mock_get_application
,
2111 mock_disconnect_controller
,
2112 mock_disconnect_model
,
2114 mock_get_controller
,
2116 mock_get_model
.return_value
= juju
.model
.Model()
2117 mock_get_application
.return_value
= FakeApplication()
2118 mock_get_machine_info
.return_value
= FakeMachine(), "series"
2119 self
.loop
.run_until_complete(
2120 self
.libjuju
.add_unit(
2127 mock_wait_for
.assert_called_once()
2128 mock_get_application
.assert_called_once()
2129 mock_get_controller
.assert_called_once()
2130 mock_get_model
.assert_called_once()
2131 mock_disconnect_controller
.assert_called_once()
2132 mock_disconnect_model
.assert_called_once()
2136 mock_get_machine_info
,
2137 mock_get_application
,
2138 mock_disconnect_controller
,
2139 mock_disconnect_model
,
2141 mock_get_controller
,
2143 mock_get_model
.return_value
= juju
.model
.Model()
2144 mock_get_application
.return_value
= None
2145 with self
.assertRaises(JujuApplicationNotFound
):
2146 self
.loop
.run_until_complete(
2147 self
.libjuju
.add_unit(
2154 mock_get_application
.assert_called_once()
2155 mock_get_controller
.assert_called_once()
2156 mock_get_model
.assert_called_once()
2157 mock_disconnect_controller
.assert_called_once()
2158 mock_disconnect_model
.assert_called_once()
2160 def test_no_machine(
2162 mock_get_machine_info
,
2163 mock_get_application
,
2164 mock_disconnect_controller
,
2165 mock_disconnect_model
,
2167 mock_get_controller
,
2169 mock_get_model
.return_value
= juju
.model
.Model()
2170 mock_get_application
.return_value
= FakeApplication()
2171 mock_get_machine_info
.side_effect
= JujuMachineNotFound()
2172 with self
.assertRaises(JujuMachineNotFound
):
2173 self
.loop
.run_until_complete(
2174 self
.libjuju
.add_unit(
2181 mock_get_application
.assert_called_once()
2182 mock_get_controller
.assert_called_once()
2183 mock_get_model
.assert_called_once()
2184 mock_disconnect_controller
.assert_called_once()
2185 mock_disconnect_model
.assert_called_once()
2188 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2189 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2190 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2191 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2192 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2193 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_unit")
2194 class DestroyUnitTest(LibjujuTestCase
):
2196 super(DestroyUnitTest
, self
).setUp()
2198 @asynctest.mock
.patch("asyncio.sleep")
2199 def test_successful(
2203 mock_get_application
,
2204 mock_disconnect_controller
,
2205 mock_disconnect_model
,
2207 mock_get_controller
,
2209 mock_get_model
.return_value
= juju
.model
.Model()
2210 mock_get_application
.return_value
= FakeApplication()
2212 self
.loop
.run_until_complete(
2213 self
.libjuju
.destroy_unit("app", "model", "machine", 0)
2216 mock_get_unit
.assert_called()
2217 mock_get_application
.assert_called_once()
2218 mock_get_controller
.assert_called_once()
2219 mock_get_model
.assert_called_once()
2220 mock_disconnect_controller
.assert_called_once()
2221 mock_disconnect_model
.assert_called_once()
2226 mock_get_application
,
2227 mock_disconnect_controller
,
2228 mock_disconnect_model
,
2230 mock_get_controller
,
2232 mock_get_model
.return_value
= juju
.model
.Model()
2233 mock_get_application
.return_value
= None
2235 with self
.assertRaises(JujuApplicationNotFound
):
2236 self
.loop
.run_until_complete(
2237 self
.libjuju
.destroy_unit("app", "model", "machine")
2240 mock_get_application
.assert_called_once()
2241 mock_get_controller
.assert_called_once()
2242 mock_get_model
.assert_called_once()
2243 mock_disconnect_controller
.assert_called_once()
2244 mock_disconnect_model
.assert_called_once()
2249 mock_get_application
,
2250 mock_disconnect_controller
,
2251 mock_disconnect_model
,
2253 mock_get_controller
,
2255 mock_get_model
.return_value
= juju
.model
.Model()
2256 mock_get_application
.return_value
= FakeApplication()
2257 mock_get_unit
.return_value
= None
2259 with self
.assertRaises(JujuError
):
2260 self
.loop
.run_until_complete(
2261 self
.libjuju
.destroy_unit("app", "model", "machine")
2264 mock_get_unit
.assert_called_once()
2265 mock_get_application
.assert_called_once()
2266 mock_get_controller
.assert_called_once()
2267 mock_get_model
.assert_called_once()
2268 mock_disconnect_controller
.assert_called_once()
2269 mock_disconnect_model
.assert_called_once()