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
23 from .utils
import FakeMachine
, FakeApplication
24 from n2vc
.libjuju
import Libjuju
25 from n2vc
.exceptions
import (
26 JujuControllerFailedConnecting
,
28 JujuApplicationNotFound
,
30 JujuApplicationExists
,
31 JujuInvalidK8sConfiguration
,
32 JujuLeaderUnitNotFound
,
35 from n2vc
.k8s_juju_conn
import generate_rbac_id
36 from n2vc
.tests
.unit
.utils
import AsyncMock
37 from n2vc
.vca
.connection
import Connection
38 from n2vc
.vca
.connection_data
import ConnectionData
41 cacert
= """-----BEGIN CERTIFICATE-----
43 -----END CERTIFICATE-----"""
46 @asynctest.mock
.patch("n2vc.libjuju.Controller")
47 class LibjujuTestCase(asynctest
.TestCase
):
48 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
51 mock_base64_to_cacert
=None,
53 self
.loop
= asyncio
.get_event_loop()
55 mock_base64_to_cacert
.return_value
= cacert
56 Connection
._load
_vca
_connection
_data
= Mock()
57 vca_connection
= Connection(AsyncMock())
58 vca_connection
._data
= ConnectionData(
60 "endpoints": ["1.2.3.4:17070"],
66 "lxd-credentials": "credentials",
67 "k8s-cloud": "k8s_cloud",
68 "k8s-credentials": "k8s_credentials",
70 "api-proxy": "api_proxy",
73 logging
.disable(logging
.CRITICAL
)
74 self
.libjuju
= Libjuju(vca_connection
, self
.loop
)
75 self
.loop
.run_until_complete(self
.libjuju
.disconnect())
78 @asynctest.mock
.patch("juju.controller.Controller.connect")
79 @asynctest.mock
.patch(
80 "juju.controller.Controller.api_endpoints",
81 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
83 class GetControllerTest(LibjujuTestCase
):
85 super(GetControllerTest
, self
).setUp()
87 def test_diff_endpoint(self
, mock_api_endpoints
, mock_connect
):
88 self
.libjuju
.endpoints
= []
89 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
90 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
92 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
95 mock_disconnect_controller
,
99 self
.libjuju
.endpoints
= []
101 mock_connect
.side_effect
= Exception()
103 with self
.assertRaises(JujuControllerFailedConnecting
):
104 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
105 self
.assertIsNone(controller
)
106 mock_disconnect_controller
.assert_called_once()
108 def test_same_endpoint_get_controller(self
, mock_api_endpoints
, mock_connect
):
109 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
110 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
111 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
114 class DisconnectTest(LibjujuTestCase
):
116 super(DisconnectTest
, self
).setUp()
118 @asynctest.mock
.patch("juju.model.Model.disconnect")
119 def test_disconnect_model(self
, mock_disconnect
):
120 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
121 mock_disconnect
.assert_called_once()
123 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
124 def test_disconnect_controller(self
, mock_disconnect
):
125 self
.loop
.run_until_complete(
126 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
128 mock_disconnect
.assert_called_once()
131 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
132 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
133 @asynctest.mock
.patch("juju.controller.Controller.add_model")
134 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
135 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
136 class AddModelTest(LibjujuTestCase
):
138 super(AddModelTest
, self
).setUp()
140 def test_existing_model(
142 mock_disconnect_model
,
143 mock_disconnect_controller
,
148 mock_model_exists
.return_value
= True
150 # This should not raise an exception
151 self
.loop
.run_until_complete(self
.libjuju
.add_model("existing_model", "cloud"))
153 mock_disconnect_controller
.assert_called()
155 # TODO Check two job executing at the same time and one returning without doing anything.
157 def test_non_existing_model(
159 mock_disconnect_model
,
160 mock_disconnect_controller
,
165 mock_model_exists
.return_value
= False
166 mock_get_controller
.return_value
= juju
.controller
.Controller()
168 self
.loop
.run_until_complete(
169 self
.libjuju
.add_model("nonexisting_model", Mock())
172 mock_add_model
.assert_called_once()
173 mock_disconnect_controller
.assert_called()
174 mock_disconnect_model
.assert_called()
177 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
178 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
179 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
180 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
181 @asynctest.mock
.patch(
182 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
184 @asynctest.mock
.patch("juju.model.Model.get_action_status")
185 @asynctest.mock
.patch("juju.model.Model.get_action_output")
186 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_actions")
187 class GetExecutedActionsTest(LibjujuTestCase
):
189 super(GetExecutedActionsTest
, self
).setUp()
194 mock_get_action_output
,
195 mock_get_action_status
,
197 mock_disconnect_controller
,
198 mock_disconnect_model
,
202 mock_get_model
.return_value
= None
203 with self
.assertRaises(JujuError
):
204 self
.loop
.run_until_complete(self
.libjuju
.get_executed_actions("model"))
206 mock_get_controller
.assert_called_once()
207 mock_disconnect_controller
.assert_called_once()
208 mock_get_model
.assert_called_once()
209 mock_disconnect_model
.assert_not_called()
214 mock_get_action_output
,
215 mock_get_action_status
,
217 mock_disconnect_controller
,
218 mock_disconnect_model
,
222 mock_get_model
.return_value
= juju
.model
.Model()
223 mock_applications
.return_value
= {"existing_app"}
224 mock_get_actions
.return_value
= {"action_name": "description"}
225 mock_get_action_status
.return_value
= {"id": "status"}
226 mock_get_action_output
.return_value
= {"output": "completed"}
228 executed_actions
= self
.loop
.run_until_complete(
229 self
.libjuju
.get_executed_actions("model")
234 "action": "action_name",
236 "output": "completed",
239 self
.assertListEqual(expected_result
, executed_actions
)
240 self
.assertIsInstance(executed_actions
, list)
242 mock_get_controller
.assert_called_once()
243 mock_get_model
.assert_called_once()
244 mock_disconnect_controller
.assert_called_once()
245 mock_disconnect_model
.assert_called_once()
248 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
249 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
250 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
251 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
252 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
253 class GetApplicationConfigsTest(LibjujuTestCase
):
255 super(GetApplicationConfigsTest
, self
).setUp()
259 mock_get_application
,
260 mock_disconnect_controller
,
261 mock_disconnect_model
,
265 mock_get_model
.return_value
= None
266 with self
.assertRaises(JujuError
):
267 self
.loop
.run_until_complete(
268 self
.libjuju
.get_application_configs("model", "app")
271 mock_get_controller
.assert_called_once()
272 mock_disconnect_controller
.assert_called_once()
273 mock_get_model
.assert_called_once()
274 mock_disconnect_model
.assert_not_called()
278 mock_get_application
,
279 mock_disconnect_controller
,
280 mock_disconnect_model
,
284 mock_get_application
.return_value
= FakeApplication()
285 application_configs
= self
.loop
.run_until_complete(
286 self
.libjuju
.get_application_configs("model", "app")
289 self
.assertEqual(application_configs
, ["app_config"])
291 mock_get_controller
.assert_called_once()
292 mock_get_model
.assert_called_once()
293 mock_disconnect_controller
.assert_called_once()
294 mock_disconnect_model
.assert_called_once()
297 @asynctest.mock
.patch("juju.controller.Controller.get_model")
298 class GetModelTest(LibjujuTestCase
):
300 super(GetModelTest
, self
).setUp()
306 mock_get_model
.return_value
= juju
.model
.Model()
307 model
= self
.loop
.run_until_complete(
308 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
310 self
.assertIsInstance(model
, juju
.model
.Model
)
313 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
314 @asynctest.mock
.patch("juju.controller.Controller.list_models")
315 class ModelExistsTest(LibjujuTestCase
):
317 super(ModelExistsTest
, self
).setUp()
319 async def test_existing_model(
324 mock_list_models
.return_value
= ["existing_model"]
326 await self
.libjuju
.model_exists(
327 "existing_model", juju
.controller
.Controller()
331 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
332 async def test_no_controller(
334 mock_disconnect_controller
,
338 mock_list_models
.return_value
= ["existing_model"]
339 mock_get_controller
.return_value
= juju
.controller
.Controller()
340 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
341 mock_disconnect_controller
.assert_called_once()
343 async def test_non_existing_model(
348 mock_list_models
.return_value
= ["existing_model"]
350 await self
.libjuju
.model_exists(
351 "not_existing_model", juju
.controller
.Controller()
356 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
357 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
358 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
359 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
360 @asynctest.mock
.patch("juju.model.Model.get_status")
361 class GetModelStatusTest(LibjujuTestCase
):
363 super(GetModelStatusTest
, self
).setUp()
368 mock_disconnect_controller
,
369 mock_disconnect_model
,
373 mock_get_model
.return_value
= juju
.model
.Model()
374 mock_get_status
.return_value
= {"status"}
376 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
378 mock_get_status
.assert_called_once()
379 mock_disconnect_controller
.assert_called_once()
380 mock_disconnect_model
.assert_called_once()
382 self
.assertEqual(status
, {"status"})
387 mock_disconnect_controller
,
388 mock_disconnect_model
,
392 mock_get_model
.return_value
= juju
.model
.Model()
393 mock_get_status
.side_effect
= Exception()
395 with self
.assertRaises(Exception):
396 status
= self
.loop
.run_until_complete(
397 self
.libjuju
.get_model_status("model")
400 mock_disconnect_controller
.assert_called_once()
401 mock_disconnect_model
.assert_called_once()
403 self
.assertIsNone(status
)
406 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
407 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
408 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
409 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
410 @asynctest.mock
.patch("juju.model.Model.get_machines")
411 @asynctest.mock
.patch("juju.model.Model.add_machine")
412 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
413 class CreateMachineTest(LibjujuTestCase
):
415 super(CreateMachineTest
, self
).setUp()
417 def test_existing_machine(
422 mock_disconnect_controller
,
423 mock_disconnect_model
,
427 mock_get_model
.return_value
= juju
.model
.Model()
428 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
429 machine
, bool_res
= self
.loop
.run_until_complete(
430 self
.libjuju
.create_machine("model", "existing_machine")
433 self
.assertIsInstance(machine
, FakeMachine
)
434 self
.assertFalse(bool_res
)
436 mock_disconnect_controller
.assert_called()
437 mock_disconnect_model
.assert_called()
439 def test_non_existing_machine(
444 mock_disconnect_controller
,
445 mock_disconnect_model
,
451 mock_get_model
.return_value
= juju
.model
.Model()
452 with self
.assertRaises(JujuMachineNotFound
):
453 machine
, bool_res
= self
.loop
.run_until_complete(
454 self
.libjuju
.create_machine("model", "non_existing_machine")
456 self
.assertIsNone(machine
)
457 self
.assertIsNone(bool_res
)
459 mock_disconnect_controller
.assert_called()
460 mock_disconnect_model
.assert_called()
467 mock_disconnect_controller
,
468 mock_disconnect_model
,
472 mock_get_model
.return_value
= juju
.model
.Model()
473 mock_add_machine
.return_value
= FakeMachine()
475 machine
, bool_res
= self
.loop
.run_until_complete(
476 self
.libjuju
.create_machine("model")
479 self
.assertIsInstance(machine
, FakeMachine
)
480 self
.assertTrue(bool_res
)
482 mock_wait_for
.assert_called_once()
483 mock_add_machine
.assert_called_once()
485 mock_disconnect_controller
.assert_called()
486 mock_disconnect_model
.assert_called()
489 # TODO test provision machine
492 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
493 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
494 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
495 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
496 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
497 @asynctest.mock
.patch("juju.model.Model.deploy")
498 class DeployTest(LibjujuTestCase
):
500 super(DeployTest
, self
).setUp()
506 mock_disconnect_controller
,
507 mock_disconnect_model
,
511 mock_get_model
.return_value
= juju
.model
.Model()
512 self
.loop
.run_until_complete(
513 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
515 mock_deploy
.assert_called_once()
516 mock_wait_for_model
.assert_called_once()
517 mock_disconnect_controller
.assert_called_once()
518 mock_disconnect_model
.assert_called_once()
520 def test_deploy_no_wait(
524 mock_disconnect_controller
,
525 mock_disconnect_model
,
529 mock_get_model
.return_value
= juju
.model
.Model()
530 self
.loop
.run_until_complete(
531 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
533 mock_deploy
.assert_called_once()
534 mock_wait_for_model
.assert_not_called()
535 mock_disconnect_controller
.assert_called_once()
536 mock_disconnect_model
.assert_called_once()
538 def test_deploy_exception(
542 mock_disconnect_controller
,
543 mock_disconnect_model
,
547 mock_deploy
.side_effect
= Exception()
548 mock_get_model
.return_value
= juju
.model
.Model()
549 with self
.assertRaises(Exception):
550 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
551 mock_deploy
.assert_called_once()
552 mock_wait_for_model
.assert_not_called()
553 mock_disconnect_controller
.assert_called_once()
554 mock_disconnect_model
.assert_called_once()
557 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
558 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
559 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
560 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
561 @asynctest.mock
.patch(
562 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
564 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
565 @asynctest.mock
.patch("juju.model.Model.deploy")
566 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
567 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
568 class DeployCharmTest(LibjujuTestCase
):
570 super(DeployCharmTest
, self
).setUp()
572 def test_existing_app(
579 mock_disconnect_controller
,
580 mock_disconnect_model
,
584 mock_get_model
.return_value
= juju
.model
.Model()
585 mock_applications
.return_value
= {"existing_app"}
588 with self
.assertRaises(JujuApplicationExists
):
589 application
= self
.loop
.run_until_complete(
590 self
.libjuju
.deploy_charm(
597 self
.assertIsNone(application
)
599 mock_disconnect_controller
.assert_called()
600 mock_disconnect_model
.assert_called()
602 def test_non_existing_machine(
609 mock_disconnect_controller
,
610 mock_disconnect_model
,
614 mock_get_model
.return_value
= juju
.model
.Model()
615 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
617 with self
.assertRaises(JujuMachineNotFound
):
618 application
= self
.loop
.run_until_complete(
619 self
.libjuju
.deploy_charm(
627 self
.assertIsNone(application
)
629 mock_disconnect_controller
.assert_called()
630 mock_disconnect_model
.assert_called()
639 mock_disconnect_controller
,
640 mock_disconnect_model
,
644 mock_get_model
.return_value
= juju
.model
.Model()
645 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
646 mock_create_machine
.return_value
= (FakeMachine(), "other")
647 mock_deploy
.return_value
= FakeApplication()
648 application
= self
.loop
.run_until_complete(
649 self
.libjuju
.deploy_charm(
658 self
.assertIsInstance(application
, FakeApplication
)
660 mock_deploy
.assert_called_once()
661 mock_wait_for
.assert_called_once()
663 mock_create_machine
.assert_called_once()
665 mock_disconnect_controller
.assert_called()
666 mock_disconnect_model
.assert_called()
675 mock_disconnect_controller
,
676 mock_disconnect_model
,
680 mock_get_model
.return_value
= juju
.model
.Model()
681 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
682 mock_deploy
.return_value
= FakeApplication()
683 application
= self
.loop
.run_until_complete(
684 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
687 self
.assertIsInstance(application
, FakeApplication
)
689 mock_deploy
.assert_called_once()
690 mock_wait_for
.assert_called_once()
692 mock_disconnect_controller
.assert_called()
693 mock_disconnect_model
.assert_called()
696 @asynctest.mock
.patch(
697 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
699 class GetApplicationTest(LibjujuTestCase
):
701 super(GetApplicationTest
, self
).setUp()
703 def test_existing_application(
707 mock_applications
.return_value
= {"existing_app": "exists"}
708 model
= juju
.model
.Model()
709 result
= self
.libjuju
._get
_application
(model
, "existing_app")
710 self
.assertEqual(result
, "exists")
712 def test_non_existing_application(
716 mock_applications
.return_value
= {"existing_app": "exists"}
717 model
= juju
.model
.Model()
718 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
719 self
.assertIsNone(result
)
722 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
723 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
724 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
725 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
726 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
727 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
728 @asynctest.mock
.patch("juju.model.Model.get_action_output")
729 @asynctest.mock
.patch("juju.model.Model.get_action_status")
730 class ExecuteActionTest(LibjujuTestCase
):
732 super(ExecuteActionTest
, self
).setUp()
734 def test_no_application(
736 mock_get_action_status
,
737 mock_get_action_output
,
739 mock__get_application
,
740 mock_disconnect_controller
,
741 mock_disconnect_model
,
745 mock__get_application
.return_value
= None
746 mock_get_model
.return_value
= juju
.model
.Model()
749 with self
.assertRaises(JujuApplicationNotFound
):
750 output
, status
= self
.loop
.run_until_complete(
751 self
.libjuju
.execute_action(
757 self
.assertIsNone(output
)
758 self
.assertIsNone(status
)
760 mock_disconnect_controller
.assert_called()
761 mock_disconnect_model
.assert_called()
765 mock_get_action_status
,
766 mock_get_action_output
,
768 mock__get_application
,
769 mock_disconnect_controller
,
770 mock_disconnect_model
,
775 mock_get_model
.return_value
= juju
.model
.Model()
776 mock__get_application
.return_value
= FakeApplication()
779 with self
.assertRaises(JujuActionNotFound
):
780 output
, status
= self
.loop
.run_until_complete(
781 self
.libjuju
.execute_action(
787 self
.assertIsNone(output
)
788 self
.assertIsNone(status
)
790 mock_disconnect_controller
.assert_called()
791 mock_disconnect_model
.assert_called()
793 @asynctest.mock
.patch("asyncio.sleep")
794 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
797 mock_is_leader_from_status
,
799 mock_get_action_status
,
800 mock_get_action_output
,
802 mock__get_application
,
803 mock_disconnect_controller
,
804 mock_disconnect_model
,
808 mock_get_model
.return_value
= juju
.model
.Model()
809 mock__get_application
.return_value
= FakeApplication()
810 mock_is_leader_from_status
.return_value
= False
813 with self
.assertRaises(JujuLeaderUnitNotFound
):
814 output
, status
= self
.loop
.run_until_complete(
815 self
.libjuju
.execute_action(
821 self
.assertIsNone(output
)
822 self
.assertIsNone(status
)
824 mock_disconnect_controller
.assert_called()
825 mock_disconnect_model
.assert_called()
827 def test_succesful_exec(
829 mock_get_action_status
,
830 mock_get_action_output
,
832 mock__get_application
,
833 mock_disconnect_controller
,
834 mock_disconnect_model
,
838 mock_get_model
.return_value
= juju
.model
.Model()
839 mock__get_application
.return_value
= FakeApplication()
840 mock_get_action_output
.return_value
= "output"
841 mock_get_action_status
.return_value
= {"id": "status"}
842 output
, status
= self
.loop
.run_until_complete(
843 self
.libjuju
.execute_action("app", "model", "existing_action")
845 self
.assertEqual(output
, "output")
846 self
.assertEqual(status
, "status")
848 mock_wait_for
.assert_called_once()
850 mock_disconnect_controller
.assert_called()
851 mock_disconnect_model
.assert_called()
854 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
855 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
856 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
857 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
858 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
859 class GetActionTest(LibjujuTestCase
):
861 super(GetActionTest
, self
).setUp()
865 mock_get_application
,
866 mock_disconnect_controller
,
867 mock_disconnect_model
,
871 mock_get_application
.side_effect
= Exception()
873 with self
.assertRaises(Exception):
874 actions
= self
.loop
.run_until_complete(
875 self
.libjuju
.get_actions("app", "model")
878 self
.assertIsNone(actions
)
879 mock_disconnect_controller
.assert_called_once()
880 mock_disconnect_model
.assert_called_once()
884 mock_get_application
,
885 mock_disconnect_controller
,
886 mock_disconnect_model
,
890 mock_get_application
.return_value
= FakeApplication()
892 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
894 self
.assertEqual(actions
, ["existing_action"])
896 mock_get_controller
.assert_called_once()
897 mock_get_model
.assert_called_once()
898 mock_disconnect_controller
.assert_called_once()
899 mock_disconnect_model
.assert_called_once()
902 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
903 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
904 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
905 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
906 @asynctest.mock
.patch("juju.application.Application.get_metrics")
907 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
908 class GetMetricsTest(LibjujuTestCase
):
910 super(GetMetricsTest
, self
).setUp()
912 def test_get_metrics_success(
914 mock_get_application
,
916 mock_disconnect_controller
,
917 mock_disconnect_model
,
921 mock_get_application
.return_value
= FakeApplication()
922 mock_get_model
.return_value
= juju
.model
.Model()
924 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
926 mock_disconnect_controller
.assert_called_once()
927 mock_disconnect_model
.assert_called_once()
929 def test_get_metrics_exception(
931 mock_get_application
,
933 mock_disconnect_controller
,
934 mock_disconnect_model
,
938 mock_get_model
.return_value
= juju
.model
.Model()
939 mock_get_metrics
.side_effect
= Exception()
940 with self
.assertRaises(Exception):
941 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
943 mock_disconnect_controller
.assert_called_once()
944 mock_disconnect_model
.assert_called_once()
946 def test_missing_args_exception(
948 mock_get_application
,
950 mock_disconnect_controller
,
951 mock_disconnect_model
,
955 mock_get_model
.return_value
= juju
.model
.Model()
957 with self
.assertRaises(Exception):
958 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
960 mock_get_controller
.assert_not_called()
961 mock_get_model
.assert_not_called()
962 mock_disconnect_controller
.assert_not_called()
963 mock_disconnect_model
.assert_not_called()
966 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
967 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
968 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
969 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
970 @asynctest.mock
.patch("juju.model.Model.add_relation")
971 class AddRelationTest(LibjujuTestCase
):
973 super(AddRelationTest
, self
).setUp()
975 @asynctest.mock
.patch("logging.Logger.warning")
980 mock_disconnect_controller
,
981 mock_disconnect_model
,
985 # TODO in libjuju.py should this fail only with a log message?
986 result
= {"error": "not found", "response": "response", "request-id": 1}
988 mock_get_model
.return_value
= juju
.model
.Model()
989 mock_add_relation
.side_effect
= JujuAPIError(result
)
991 self
.loop
.run_until_complete(
992 self
.libjuju
.add_relation(
999 mock_warning
.assert_called_with("Relation not found: not found")
1000 mock_disconnect_controller
.assert_called_once()
1001 mock_disconnect_model
.assert_called_once()
1003 @asynctest.mock
.patch("logging.Logger.warning")
1004 def test_already_exists(
1008 mock_disconnect_controller
,
1009 mock_disconnect_model
,
1011 mock_get_controller
,
1013 # TODO in libjuju.py should this fail silently?
1014 result
= {"error": "already exists", "response": "response", "request-id": 1}
1016 mock_get_model
.return_value
= juju
.model
.Model()
1017 mock_add_relation
.side_effect
= JujuAPIError(result
)
1019 self
.loop
.run_until_complete(
1020 self
.libjuju
.add_relation(
1027 mock_warning
.assert_called_with("Relation already exists: already exists")
1028 mock_disconnect_controller
.assert_called_once()
1029 mock_disconnect_model
.assert_called_once()
1034 mock_disconnect_controller
,
1035 mock_disconnect_model
,
1037 mock_get_controller
,
1039 mock_get_model
.return_value
= juju
.model
.Model()
1040 result
= {"error": "", "response": "response", "request-id": 1}
1041 mock_add_relation
.side_effect
= JujuAPIError(result
)
1043 with self
.assertRaises(JujuAPIError
):
1044 self
.loop
.run_until_complete(
1045 self
.libjuju
.add_relation(
1052 mock_disconnect_controller
.assert_called_once()
1053 mock_disconnect_model
.assert_called_once()
1058 mock_disconnect_controller
,
1059 mock_disconnect_model
,
1061 mock_get_controller
,
1063 mock_get_model
.return_value
= juju
.model
.Model()
1065 self
.loop
.run_until_complete(
1066 self
.libjuju
.add_relation(
1073 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1074 mock_disconnect_controller
.assert_called_once()
1075 mock_disconnect_model
.assert_called_once()
1080 mock_disconnect_controller
,
1081 mock_disconnect_model
,
1083 mock_get_controller
,
1085 mock_get_model
.return_value
= juju
.model
.Model()
1087 self
.loop
.run_until_complete(
1088 self
.libjuju
.add_relation(
1095 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1096 mock_disconnect_controller
.assert_called_once()
1097 mock_disconnect_model
.assert_called_once()
1100 # TODO destroy_model testcase
1103 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1104 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1105 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1106 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1107 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1108 class DestroyApplicationTest(LibjujuTestCase
):
1110 super(DestroyApplicationTest
, self
).setUp()
1114 mock_get_controller
,
1116 mock_disconnect_controller
,
1117 mock_get_application
,
1118 mock_disconnect_model
,
1120 mock_get_application
.return_value
= FakeApplication()
1121 mock_get_model
.return_value
= None
1122 self
.loop
.run_until_complete(
1123 self
.libjuju
.destroy_application(
1129 mock_get_application
.assert_called()
1130 mock_disconnect_controller
.assert_called_once()
1131 mock_disconnect_model
.assert_called_once()
1133 def test_no_application(
1135 mock_get_controller
,
1137 mock_disconnect_controller
,
1138 mock_get_application
,
1139 mock_disconnect_model
,
1141 mock_get_model
.return_value
= None
1142 mock_get_application
.return_value
= None
1144 self
.loop
.run_until_complete(
1145 self
.libjuju
.destroy_application(
1151 mock_get_application
.assert_called()
1155 mock_get_controller
,
1157 mock_disconnect_controller
,
1158 mock_get_application
,
1159 mock_disconnect_model
,
1161 mock_get_application
.return_value
= FakeApplication
1162 mock_get_model
.return_value
= None
1164 with self
.assertRaises(Exception):
1165 self
.loop
.run_until_complete(
1166 self
.libjuju
.destroy_application(
1172 mock_get_application
.assert_called_once()
1175 # @asynctest.mock.patch("juju.model.Model.get_machines")
1176 # @asynctest.mock.patch("logging.Logger.debug")
1177 # class DestroyMachineTest(LibjujuTestCase):
1179 # super(DestroyMachineTest, self).setUp()
1181 # def test_success_manual_machine(
1182 # self, mock_debug, mock_get_machines,
1184 # mock_get_machines.side_effect = [
1185 # {"machine": FakeManualMachine()},
1186 # {"machine": FakeManualMachine()},
1189 # self.loop.run_until_complete(
1190 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1193 # asynctest.call("Waiting for machine machine is destroyed"),
1194 # asynctest.call("Machine destroyed: machine"),
1196 # mock_debug.assert_has_calls(calls)
1198 # def test_no_machine(
1199 # self, mock_debug, mock_get_machines,
1201 # mock_get_machines.return_value = {}
1202 # self.loop.run_until_complete(
1203 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1205 # mock_debug.assert_called_with("Machine not found: machine")
1208 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1209 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1210 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1211 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1212 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1213 class ConfigureApplicationTest(LibjujuTestCase
):
1215 super(ConfigureApplicationTest
, self
).setUp()
1219 mock_get_application
,
1220 mock_disconnect_controller
,
1221 mock_disconnect_model
,
1223 mock_get_controller
,
1226 mock_get_application
.return_value
= FakeApplication()
1228 self
.loop
.run_until_complete(
1229 self
.libjuju
.configure_application(
1235 mock_get_application
.assert_called_once()
1236 mock_disconnect_controller
.assert_called_once()
1237 mock_disconnect_model
.assert_called_once()
1241 mock_get_application
,
1242 mock_disconnect_controller
,
1243 mock_disconnect_model
,
1245 mock_get_controller
,
1248 mock_get_application
.side_effect
= Exception()
1250 with self
.assertRaises(Exception):
1251 self
.loop
.run_until_complete(
1252 self
.libjuju
.configure_application(
1258 mock_disconnect_controller
.assert_called_once()
1259 mock_disconnect_model
.assert_called_once()
1261 def test_controller_exception(
1263 mock_get_application
,
1264 mock_disconnect_controller
,
1265 mock_disconnect_model
,
1267 mock_get_controller
,
1270 result
= {"error": "not found", "response": "response", "request-id": 1}
1272 mock_get_controller
.side_effect
= JujuAPIError(result
)
1274 with self
.assertRaises(JujuAPIError
):
1275 self
.loop
.run_until_complete(
1276 self
.libjuju
.configure_application(
1282 mock_get_model
.assert_not_called()
1283 mock_disconnect_controller
.assert_not_called()
1284 mock_disconnect_model
.assert_not_called()
1286 def test_get_model_exception(
1288 mock_get_application
,
1289 mock_disconnect_controller
,
1290 mock_disconnect_model
,
1292 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_succesful_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)