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
import mock
21 from juju
.errors
import JujuAPIError
23 from .utils
import FakeN2VC
, FakeMachine
, FakeApplication
24 from n2vc
.libjuju
import Libjuju
25 from n2vc
.exceptions
import (
26 JujuControllerFailedConnecting
,
27 JujuModelAlreadyExists
,
29 JujuApplicationNotFound
,
31 JujuApplicationExists
,
32 JujuInvalidK8sConfiguration
,
33 JujuLeaderUnitNotFound
,
35 from n2vc
.k8s_juju_conn
import generate_rbac_id
38 class LibjujuTestCase(asynctest
.TestCase
):
39 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
40 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
41 @asynctest.mock
.patch("juju.controller.Controller.connection")
42 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
45 mock__get_api_endpoints_db
=None,
48 mock_update_endpoints
=None,
50 loop
= asyncio
.get_event_loop()
52 mock__get_api_endpoints_db
.return_value
= ["127.0.0.1:17070"]
53 endpoints
= "127.0.0.1:17070"
57 -----BEGIN CERTIFICATE-----
59 -----END CERTIFICATE-----"""
60 self
.libjuju
= Libjuju(
62 "192.168.0.155:17070",
70 apt_mirror
="192.168.0.100",
71 enable_os_upgrade
=True,
73 logging
.disable(logging
.CRITICAL
)
74 loop
.run_until_complete(self
.libjuju
.disconnect())
77 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._create_health_check_task")
78 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
79 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
80 class LibjujuInitTestCase(asynctest
.TestCase
):
82 self
.loop
= asyncio
.get_event_loop()
83 self
.n2vc
= FakeN2VC()
84 self
.endpoint
= "192.168.100.100:17070"
85 self
.username
= "admin"
86 self
.password
= "secret"
88 -----BEGIN CERTIFICATE-----
90 -----END CERTIFICATE-----"""
92 def test_endpoint_not_in_db(
94 mock__get_api_endpoints_db
,
95 mock_update_endpoints
,
96 mock_create_health_check_task
,
98 mock__get_api_endpoints_db
.return_value
= ["another_ip"]
101 "192.168.0.155:17070",
109 apt_mirror
="192.168.0.100",
110 enable_os_upgrade
=True,
112 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
113 mock__get_api_endpoints_db
.assert_called_once()
115 def test_endpoint_in_db(
117 mock__get_api_endpoints_db
,
118 mock_update_endpoints
,
119 mock_create_health_check_task
,
121 mock__get_api_endpoints_db
.return_value
= [self
.endpoint
, "another_ip"]
124 "192.168.0.155:17070",
132 apt_mirror
="192.168.0.100",
133 enable_os_upgrade
=True,
135 mock_update_endpoints
.assert_not_called()
136 mock__get_api_endpoints_db
.assert_called_once()
138 def test_no_db_endpoints(
140 mock__get_api_endpoints_db
,
141 mock_update_endpoints
,
142 mock_create_health_check_task
,
144 mock__get_api_endpoints_db
.return_value
= None
147 "192.168.0.155:17070",
155 apt_mirror
="192.168.0.100",
156 enable_os_upgrade
=True,
158 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
159 mock__get_api_endpoints_db
.assert_called_once()
162 @asynctest.mock
.patch("juju.controller.Controller.connect")
163 @asynctest.mock
.patch(
164 "juju.controller.Controller.api_endpoints",
165 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
167 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
168 class GetControllerTest(LibjujuTestCase
):
170 super(GetControllerTest
, self
).setUp()
172 def test_diff_endpoint(
173 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
175 self
.libjuju
.endpoints
= []
176 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
177 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
178 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
180 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
183 mock_disconnect_controller
,
184 mock__update_api_endpoints_db
,
188 self
.libjuju
.endpoints
= []
189 mock__update_api_endpoints_db
.side_effect
= Exception()
191 with self
.assertRaises(JujuControllerFailedConnecting
):
192 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
193 self
.assertIsNone(controller
)
194 mock_disconnect_controller
.assert_called_once()
196 def test_same_endpoint_get_controller(
197 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
199 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
200 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
201 mock__update_api_endpoints_db
.assert_not_called()
202 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
205 class DisconnectTest(LibjujuTestCase
):
207 super(DisconnectTest
, self
).setUp()
209 @asynctest.mock
.patch("juju.model.Model.disconnect")
210 def test_disconnect_model(self
, mock_disconnect
):
211 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
212 mock_disconnect
.assert_called_once()
214 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
215 def test_disconnect_controller(self
, mock_disconnect
):
216 self
.loop
.run_until_complete(
217 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
219 mock_disconnect
.assert_called_once()
222 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
223 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
224 @asynctest.mock
.patch("juju.controller.Controller.add_model")
225 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
226 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
227 class AddModelTest(LibjujuTestCase
):
229 super(AddModelTest
, self
).setUp()
231 def test_existing_model(
233 mock_disconnect_model
,
234 mock_disconnect_controller
,
239 mock_model_exists
.return_value
= True
241 with self
.assertRaises(JujuModelAlreadyExists
):
242 self
.loop
.run_until_complete(
243 self
.libjuju
.add_model("existing_model", "cloud")
246 mock_disconnect_controller
.assert_called()
248 # TODO Check two job executing at the same time and one returning without doing anything.
250 def test_non_existing_model(
252 mock_disconnect_model
,
253 mock_disconnect_controller
,
258 mock_model_exists
.return_value
= False
259 mock_get_controller
.return_value
= juju
.controller
.Controller()
261 self
.loop
.run_until_complete(
262 self
.libjuju
.add_model("nonexisting_model", "cloud")
265 mock_add_model
.assert_called_once()
266 mock_disconnect_controller
.assert_called()
267 mock_disconnect_model
.assert_called()
270 @asynctest.mock
.patch("juju.controller.Controller.get_model")
271 class GetModelTest(LibjujuTestCase
):
273 super(GetModelTest
, self
).setUp()
279 mock_get_model
.return_value
= juju
.model
.Model()
280 model
= self
.loop
.run_until_complete(
281 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
283 self
.assertIsInstance(model
, juju
.model
.Model
)
286 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
287 @asynctest.mock
.patch("juju.controller.Controller.list_models")
288 class ModelExistsTest(LibjujuTestCase
):
290 super(ModelExistsTest
, self
).setUp()
292 async def test_existing_model(
297 mock_list_models
.return_value
= ["existing_model"]
299 await self
.libjuju
.model_exists(
300 "existing_model", juju
.controller
.Controller()
304 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
305 async def test_no_controller(
307 mock_disconnect_controller
,
311 mock_list_models
.return_value
= ["existing_model"]
312 mock_get_controller
.return_value
= juju
.controller
.Controller()
313 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
314 mock_disconnect_controller
.assert_called_once()
316 async def test_non_existing_model(
321 mock_list_models
.return_value
= ["existing_model"]
323 await self
.libjuju
.model_exists(
324 "not_existing_model", juju
.controller
.Controller()
329 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
330 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
331 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
332 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
333 @asynctest.mock
.patch("juju.model.Model.get_status")
334 class GetModelStatusTest(LibjujuTestCase
):
336 super(GetModelStatusTest
, self
).setUp()
341 mock_disconnect_controller
,
342 mock_disconnect_model
,
346 mock_get_model
.return_value
= juju
.model
.Model()
347 mock_get_status
.return_value
= {"status"}
349 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
351 mock_get_status
.assert_called_once()
352 mock_disconnect_controller
.assert_called_once()
353 mock_disconnect_model
.assert_called_once()
355 self
.assertEqual(status
, {"status"})
360 mock_disconnect_controller
,
361 mock_disconnect_model
,
365 mock_get_model
.return_value
= juju
.model
.Model()
366 mock_get_status
.side_effect
= Exception()
368 with self
.assertRaises(Exception):
369 status
= self
.loop
.run_until_complete(
370 self
.libjuju
.get_model_status("model")
373 mock_disconnect_controller
.assert_called_once()
374 mock_disconnect_model
.assert_called_once()
376 self
.assertIsNone(status
)
379 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
380 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
381 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
382 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
383 @asynctest.mock
.patch("juju.model.Model.get_machines")
384 @asynctest.mock
.patch("juju.model.Model.add_machine")
385 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
386 class CreateMachineTest(LibjujuTestCase
):
388 super(CreateMachineTest
, self
).setUp()
390 def test_existing_machine(
395 mock_disconnect_controller
,
396 mock_disconnect_model
,
400 mock_get_model
.return_value
= juju
.model
.Model()
401 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
402 machine
, bool_res
= self
.loop
.run_until_complete(
403 self
.libjuju
.create_machine("model", "existing_machine")
406 self
.assertIsInstance(machine
, FakeMachine
)
407 self
.assertFalse(bool_res
)
409 mock_disconnect_controller
.assert_called()
410 mock_disconnect_model
.assert_called()
412 def test_non_existing_machine(
417 mock_disconnect_controller
,
418 mock_disconnect_model
,
424 mock_get_model
.return_value
= juju
.model
.Model()
425 with self
.assertRaises(JujuMachineNotFound
):
426 machine
, bool_res
= self
.loop
.run_until_complete(
427 self
.libjuju
.create_machine("model", "non_existing_machine")
429 self
.assertIsNone(machine
)
430 self
.assertIsNone(bool_res
)
432 mock_disconnect_controller
.assert_called()
433 mock_disconnect_model
.assert_called()
440 mock_disconnect_controller
,
441 mock_disconnect_model
,
445 mock_get_model
.return_value
= juju
.model
.Model()
446 mock_add_machine
.return_value
= FakeMachine()
448 machine
, bool_res
= self
.loop
.run_until_complete(
449 self
.libjuju
.create_machine("model")
452 self
.assertIsInstance(machine
, FakeMachine
)
453 self
.assertTrue(bool_res
)
455 mock_wait_for
.assert_called_once()
456 mock_add_machine
.assert_called_once()
458 mock_disconnect_controller
.assert_called()
459 mock_disconnect_model
.assert_called()
462 # TODO test provision machine
465 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
466 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
467 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
468 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
469 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
470 @asynctest.mock
.patch("juju.model.Model.deploy")
471 class DeployTest(LibjujuTestCase
):
473 super(DeployTest
, self
).setUp()
479 mock_disconnect_controller
,
480 mock_disconnect_model
,
484 mock_get_model
.return_value
= juju
.model
.Model()
485 self
.loop
.run_until_complete(
486 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
488 mock_deploy
.assert_called_once()
489 mock_wait_for_model
.assert_called_once()
490 mock_disconnect_controller
.assert_called_once()
491 mock_disconnect_model
.assert_called_once()
493 def test_deploy_no_wait(
497 mock_disconnect_controller
,
498 mock_disconnect_model
,
502 mock_get_model
.return_value
= juju
.model
.Model()
503 self
.loop
.run_until_complete(
504 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
506 mock_deploy
.assert_called_once()
507 mock_wait_for_model
.assert_not_called()
508 mock_disconnect_controller
.assert_called_once()
509 mock_disconnect_model
.assert_called_once()
511 def test_deploy_exception(
515 mock_disconnect_controller
,
516 mock_disconnect_model
,
520 mock_deploy
.side_effect
= Exception()
521 mock_get_model
.return_value
= juju
.model
.Model()
522 with self
.assertRaises(Exception):
523 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
524 mock_deploy
.assert_called_once()
525 mock_wait_for_model
.assert_not_called()
526 mock_disconnect_controller
.assert_called_once()
527 mock_disconnect_model
.assert_called_once()
530 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
531 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
532 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
533 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
534 @asynctest.mock
.patch(
535 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
537 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
538 @asynctest.mock
.patch("juju.model.Model.deploy")
539 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
540 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
541 class DeployCharmTest(LibjujuTestCase
):
543 super(DeployCharmTest
, self
).setUp()
545 def test_existing_app(
552 mock_disconnect_controller
,
553 mock_disconnect_model
,
557 mock_get_model
.return_value
= juju
.model
.Model()
558 mock_applications
.return_value
= {"existing_app"}
561 with self
.assertRaises(JujuApplicationExists
):
562 application
= self
.loop
.run_until_complete(
563 self
.libjuju
.deploy_charm(
570 self
.assertIsNone(application
)
572 mock_disconnect_controller
.assert_called()
573 mock_disconnect_model
.assert_called()
575 def test_non_existing_machine(
582 mock_disconnect_controller
,
583 mock_disconnect_model
,
587 mock_get_model
.return_value
= juju
.model
.Model()
588 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
590 with self
.assertRaises(JujuMachineNotFound
):
591 application
= self
.loop
.run_until_complete(
592 self
.libjuju
.deploy_charm(
600 self
.assertIsNone(application
)
602 mock_disconnect_controller
.assert_called()
603 mock_disconnect_model
.assert_called()
612 mock_disconnect_controller
,
613 mock_disconnect_model
,
617 mock_get_model
.return_value
= juju
.model
.Model()
618 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
619 mock_create_machine
.return_value
= (FakeMachine(), "other")
620 mock_deploy
.return_value
= FakeApplication()
621 application
= self
.loop
.run_until_complete(
622 self
.libjuju
.deploy_charm(
631 self
.assertIsInstance(application
, FakeApplication
)
633 mock_deploy
.assert_called_once()
634 mock_wait_for
.assert_called_once()
636 mock_create_machine
.assert_called_once()
638 mock_disconnect_controller
.assert_called()
639 mock_disconnect_model
.assert_called()
648 mock_disconnect_controller
,
649 mock_disconnect_model
,
653 mock_get_model
.return_value
= juju
.model
.Model()
654 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
655 mock_deploy
.return_value
= FakeApplication()
656 application
= self
.loop
.run_until_complete(
657 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
660 self
.assertIsInstance(application
, FakeApplication
)
662 mock_deploy
.assert_called_once()
663 mock_wait_for
.assert_called_once()
665 mock_disconnect_controller
.assert_called()
666 mock_disconnect_model
.assert_called()
669 @asynctest.mock
.patch(
670 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
672 class GetApplicationTest(LibjujuTestCase
):
674 super(GetApplicationTest
, self
).setUp()
676 def test_existing_application(
680 mock_applications
.return_value
= {"existing_app": "exists"}
681 model
= juju
.model
.Model()
682 result
= self
.libjuju
._get
_application
(model
, "existing_app")
683 self
.assertEqual(result
, "exists")
685 def test_non_existing_application(
689 mock_applications
.return_value
= {"existing_app": "exists"}
690 model
= juju
.model
.Model()
691 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
692 self
.assertIsNone(result
)
695 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
696 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
697 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
698 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
699 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
700 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
701 @asynctest.mock
.patch("juju.model.Model.get_action_output")
702 @asynctest.mock
.patch("juju.model.Model.get_action_status")
703 class ExecuteActionTest(LibjujuTestCase
):
705 super(ExecuteActionTest
, self
).setUp()
707 def test_no_application(
709 mock_get_action_status
,
710 mock_get_action_output
,
712 mock__get_application
,
713 mock_disconnect_controller
,
714 mock_disconnect_model
,
718 mock__get_application
.return_value
= None
719 mock_get_model
.return_value
= juju
.model
.Model()
722 with self
.assertRaises(JujuApplicationNotFound
):
723 output
, status
= self
.loop
.run_until_complete(
724 self
.libjuju
.execute_action(
730 self
.assertIsNone(output
)
731 self
.assertIsNone(status
)
733 mock_disconnect_controller
.assert_called()
734 mock_disconnect_model
.assert_called()
738 mock_get_action_status
,
739 mock_get_action_output
,
741 mock__get_application
,
742 mock_disconnect_controller
,
743 mock_disconnect_model
,
748 mock_get_model
.return_value
= juju
.model
.Model()
749 mock__get_application
.return_value
= FakeApplication()
752 with self
.assertRaises(JujuActionNotFound
):
753 output
, status
= self
.loop
.run_until_complete(
754 self
.libjuju
.execute_action(
760 self
.assertIsNone(output
)
761 self
.assertIsNone(status
)
763 mock_disconnect_controller
.assert_called()
764 mock_disconnect_model
.assert_called()
766 @asynctest.mock
.patch("asyncio.sleep")
767 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
770 mock_is_leader_from_status
,
772 mock_get_action_status
,
773 mock_get_action_output
,
775 mock__get_application
,
776 mock_disconnect_controller
,
777 mock_disconnect_model
,
781 mock_get_model
.return_value
= juju
.model
.Model()
782 mock__get_application
.return_value
= FakeApplication()
783 mock_is_leader_from_status
.return_value
= False
786 with self
.assertRaises(JujuLeaderUnitNotFound
):
787 output
, status
= self
.loop
.run_until_complete(
788 self
.libjuju
.execute_action(
794 self
.assertIsNone(output
)
795 self
.assertIsNone(status
)
797 mock_disconnect_controller
.assert_called()
798 mock_disconnect_model
.assert_called()
800 def test_succesful_exec(
802 mock_get_action_status
,
803 mock_get_action_output
,
805 mock__get_application
,
806 mock_disconnect_controller
,
807 mock_disconnect_model
,
811 mock_get_model
.return_value
= juju
.model
.Model()
812 mock__get_application
.return_value
= FakeApplication()
813 mock_get_action_output
.return_value
= "output"
814 mock_get_action_status
.return_value
= {"id": "status"}
815 output
, status
= self
.loop
.run_until_complete(
816 self
.libjuju
.execute_action("app", "model", "existing_action")
818 self
.assertEqual(output
, "output")
819 self
.assertEqual(status
, "status")
821 mock_wait_for
.assert_called_once()
823 mock_disconnect_controller
.assert_called()
824 mock_disconnect_model
.assert_called()
827 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
828 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
829 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
830 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
831 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
832 class GetActionTest(LibjujuTestCase
):
834 super(GetActionTest
, self
).setUp()
838 mock_get_application
,
839 mock_disconnect_controller
,
840 mock_disconnect_model
,
844 mock_get_application
.side_effect
= Exception()
846 with self
.assertRaises(Exception):
847 actions
= self
.loop
.run_until_complete(
848 self
.libjuju
.get_actions("app", "model")
851 self
.assertIsNone(actions
)
852 mock_disconnect_controller
.assert_called_once()
853 mock_disconnect_model
.assert_called_once()
857 mock_get_application
,
858 mock_disconnect_controller
,
859 mock_disconnect_model
,
863 mock_get_application
.return_value
= FakeApplication()
865 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
867 self
.assertEqual(actions
, ["existing_action"])
869 mock_get_controller
.assert_called_once()
870 mock_get_model
.assert_called_once()
871 mock_disconnect_controller
.assert_called_once()
872 mock_disconnect_model
.assert_called_once()
875 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
876 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
877 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
878 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
879 @asynctest.mock
.patch("juju.application.Application.get_metrics")
880 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
881 class GetMetricsTest(LibjujuTestCase
):
883 super(GetMetricsTest
, self
).setUp()
885 def test_get_metrics_success(
887 mock_get_application
,
889 mock_disconnect_controller
,
890 mock_disconnect_model
,
894 mock_get_application
.return_value
= FakeApplication()
895 mock_get_model
.return_value
= juju
.model
.Model()
897 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
899 mock_disconnect_controller
.assert_called_once()
900 mock_disconnect_model
.assert_called_once()
902 def test_get_metrics_exception(
904 mock_get_application
,
906 mock_disconnect_controller
,
907 mock_disconnect_model
,
911 mock_get_model
.return_value
= juju
.model
.Model()
912 mock_get_metrics
.side_effect
= Exception()
913 with self
.assertRaises(Exception):
914 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
916 mock_disconnect_controller
.assert_called_once()
917 mock_disconnect_model
.assert_called_once()
919 def test_missing_args_exception(
921 mock_get_application
,
923 mock_disconnect_controller
,
924 mock_disconnect_model
,
928 mock_get_model
.return_value
= juju
.model
.Model()
930 with self
.assertRaises(Exception):
931 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
933 mock_get_controller
.assert_not_called()
934 mock_get_model
.assert_not_called()
935 mock_disconnect_controller
.assert_not_called()
936 mock_disconnect_model
.assert_not_called()
939 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
940 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
941 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
942 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
943 @asynctest.mock
.patch("juju.model.Model.add_relation")
944 class AddRelationTest(LibjujuTestCase
):
946 super(AddRelationTest
, self
).setUp()
948 @asynctest.mock
.patch("logging.Logger.warning")
953 mock_disconnect_controller
,
954 mock_disconnect_model
,
958 # TODO in libjuju.py should this fail only with a log message?
959 result
= {"error": "not found", "response": "response", "request-id": 1}
961 mock_get_model
.return_value
= juju
.model
.Model()
962 mock_add_relation
.side_effect
= JujuAPIError(result
)
964 self
.loop
.run_until_complete(
965 self
.libjuju
.add_relation(
972 mock_warning
.assert_called_with("Relation not found: not found")
973 mock_disconnect_controller
.assert_called_once()
974 mock_disconnect_model
.assert_called_once()
976 @asynctest.mock
.patch("logging.Logger.warning")
977 def test_already_exists(
981 mock_disconnect_controller
,
982 mock_disconnect_model
,
986 # TODO in libjuju.py should this fail silently?
987 result
= {"error": "already exists", "response": "response", "request-id": 1}
989 mock_get_model
.return_value
= juju
.model
.Model()
990 mock_add_relation
.side_effect
= JujuAPIError(result
)
992 self
.loop
.run_until_complete(
993 self
.libjuju
.add_relation(
1000 mock_warning
.assert_called_with("Relation already exists: already exists")
1001 mock_disconnect_controller
.assert_called_once()
1002 mock_disconnect_model
.assert_called_once()
1007 mock_disconnect_controller
,
1008 mock_disconnect_model
,
1010 mock_get_controller
,
1012 mock_get_model
.return_value
= juju
.model
.Model()
1013 result
= {"error": "", "response": "response", "request-id": 1}
1014 mock_add_relation
.side_effect
= JujuAPIError(result
)
1016 with self
.assertRaises(JujuAPIError
):
1017 self
.loop
.run_until_complete(
1018 self
.libjuju
.add_relation(
1025 mock_disconnect_controller
.assert_called_once()
1026 mock_disconnect_model
.assert_called_once()
1031 mock_disconnect_controller
,
1032 mock_disconnect_model
,
1034 mock_get_controller
,
1036 mock_get_model
.return_value
= juju
.model
.Model()
1038 self
.loop
.run_until_complete(
1039 self
.libjuju
.add_relation(
1046 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1047 mock_disconnect_controller
.assert_called_once()
1048 mock_disconnect_model
.assert_called_once()
1053 mock_disconnect_controller
,
1054 mock_disconnect_model
,
1056 mock_get_controller
,
1058 mock_get_model
.return_value
= juju
.model
.Model()
1060 self
.loop
.run_until_complete(
1061 self
.libjuju
.add_relation(
1068 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1069 mock_disconnect_controller
.assert_called_once()
1070 mock_disconnect_model
.assert_called_once()
1073 # TODO destroy_model testcase
1076 # @asynctest.mock.patch("juju.model.Model.get_machines")
1077 # @asynctest.mock.patch("logging.Logger.debug")
1078 # class DestroyMachineTest(LibjujuTestCase):
1080 # super(DestroyMachineTest, self).setUp()
1082 # def test_success_manual_machine(
1083 # self, mock_debug, mock_get_machines,
1085 # mock_get_machines.side_effect = [
1086 # {"machine": FakeManualMachine()},
1087 # {"machine": FakeManualMachine()},
1090 # self.loop.run_until_complete(
1091 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1094 # asynctest.call("Waiting for machine machine is destroyed"),
1095 # asynctest.call("Machine destroyed: machine"),
1097 # mock_debug.assert_has_calls(calls)
1099 # def test_no_machine(
1100 # self, mock_debug, mock_get_machines,
1102 # mock_get_machines.return_value = {}
1103 # self.loop.run_until_complete(
1104 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1106 # mock_debug.assert_called_with("Machine not found: machine")
1109 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1112 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1113 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1114 class ConfigureApplicationTest(LibjujuTestCase
):
1116 super(ConfigureApplicationTest
, self
).setUp()
1120 mock_get_application
,
1121 mock_disconnect_controller
,
1122 mock_disconnect_model
,
1124 mock_get_controller
,
1127 mock_get_application
.return_value
= FakeApplication()
1129 self
.loop
.run_until_complete(
1130 self
.libjuju
.configure_application(
1136 mock_get_application
.assert_called_once()
1137 mock_disconnect_controller
.assert_called_once()
1138 mock_disconnect_model
.assert_called_once()
1142 mock_get_application
,
1143 mock_disconnect_controller
,
1144 mock_disconnect_model
,
1146 mock_get_controller
,
1149 mock_get_application
.side_effect
= Exception()
1151 with self
.assertRaises(Exception):
1152 self
.loop
.run_until_complete(
1153 self
.libjuju
.configure_application(
1159 mock_disconnect_controller
.assert_called_once()
1160 mock_disconnect_model
.assert_called_once()
1162 def test_controller_exception(
1164 mock_get_application
,
1165 mock_disconnect_controller
,
1166 mock_disconnect_model
,
1168 mock_get_controller
,
1171 result
= {"error": "not found", "response": "response", "request-id": 1}
1173 mock_get_controller
.side_effect
= JujuAPIError(result
)
1175 with self
.assertRaises(JujuAPIError
):
1176 self
.loop
.run_until_complete(
1177 self
.libjuju
.configure_application(
1183 mock_get_model
.assert_not_called()
1184 mock_disconnect_controller
.assert_not_called()
1185 mock_disconnect_model
.assert_not_called()
1187 def test_get_model_exception(
1189 mock_get_application
,
1190 mock_disconnect_controller
,
1191 mock_disconnect_model
,
1193 mock_get_controller
,
1196 result
= {"error": "not found", "response": "response", "request-id": 1}
1197 mock_get_model
.side_effect
= JujuAPIError(result
)
1199 with self
.assertRaises(JujuAPIError
):
1200 self
.loop
.run_until_complete(
1201 self
.libjuju
.configure_application(
1207 mock_get_model
.assert_called_once()
1208 mock_disconnect_controller
.assert_called_once()
1209 mock_disconnect_model
.assert_not_called()
1212 # TODO _get_api_endpoints_db test case
1213 # TODO _update_api_endpoints_db test case
1214 # TODO healthcheck test case
1217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1218 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1219 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1220 class ListModelsTest(LibjujuTestCase
):
1222 super(ListModelsTest
, self
).setUp()
1224 def test_containing(
1227 mock_disconnect_controller
,
1228 mock_get_controller
,
1230 mock_get_controller
.return_value
= juju
.controller
.Controller()
1231 mock_list_models
.return_value
= ["existingmodel"]
1232 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1234 mock_disconnect_controller
.assert_called_once()
1235 self
.assertEquals(models
, ["existingmodel"])
1237 def test_not_containing(
1240 mock_disconnect_controller
,
1241 mock_get_controller
,
1243 mock_get_controller
.return_value
= juju
.controller
.Controller()
1244 mock_list_models
.return_value
= ["existingmodel", "model"]
1245 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1247 mock_disconnect_controller
.assert_called_once()
1248 self
.assertEquals(models
, [])
1250 def test_no_contains_arg(
1253 mock_disconnect_controller
,
1254 mock_get_controller
,
1256 mock_get_controller
.return_value
= juju
.controller
.Controller()
1257 mock_list_models
.return_value
= ["existingmodel", "model"]
1258 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1260 mock_disconnect_controller
.assert_called_once()
1261 self
.assertEquals(models
, ["existingmodel", "model"])
1264 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1265 class ModelsExistTest(LibjujuTestCase
):
1267 super(ModelsExistTest
, self
).setUp()
1269 def test_model_names_none(self
, mock_list_models
):
1270 mock_list_models
.return_value
= []
1271 with self
.assertRaises(Exception):
1272 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1274 def test_model_names_empty(self
, mock_list_models
):
1275 mock_list_models
.return_value
= []
1276 with self
.assertRaises(Exception):
1277 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1278 self
.libjuju
.models_exist([])
1281 def test_model_names_not_existing(self
, mock_list_models
):
1282 mock_list_models
.return_value
= ["prometheus", "grafana"]
1283 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1284 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1286 self
.assertFalse(exist
)
1287 self
.assertEqual(non_existing_models
, ["prometheus2"])
1289 def test_model_names_exist(self
, mock_list_models
):
1290 mock_list_models
.return_value
= ["prometheus", "grafana"]
1291 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1292 self
.libjuju
.models_exist(["prometheus", "grafana"])
1294 self
.assertTrue(exist
)
1295 self
.assertEqual(non_existing_models
, [])
1298 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1299 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1300 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1301 class ListOffers(LibjujuTestCase
):
1303 super(ListOffers
, self
).setUp()
1305 def test_disconnect_controller(
1308 mock_disconnect_controller
,
1309 mock_get_controller
,
1311 mock_get_controller
.return_value
= juju
.controller
.Controller()
1312 mock_list_offers
.side_effect
= Exception()
1313 with self
.assertRaises(Exception):
1314 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1315 mock_disconnect_controller
.assert_called_once()
1317 def test_empty_list(
1320 mock_disconnect_controller
,
1321 mock_get_controller
,
1323 mock_get_controller
.return_value
= juju
.controller
.Controller()
1324 mock_list_offers
.return_value
= []
1325 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1326 self
.assertEqual(offers
, [])
1327 mock_disconnect_controller
.assert_called_once()
1329 def test_non_empty_list(
1332 mock_disconnect_controller
,
1333 mock_get_controller
,
1335 mock_get_controller
.return_value
= juju
.controller
.Controller()
1336 mock_list_offers
.return_value
= ["offer"]
1337 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1338 self
.assertEqual(offers
, ["offer"])
1339 mock_disconnect_controller
.assert_called_once()
1342 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1343 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1344 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1345 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1346 @asynctest.mock
.patch("juju.model.Model.consume")
1347 class ConsumeTest(LibjujuTestCase
):
1349 super(ConsumeTest
, self
).setUp()
1354 mock_disconnect_controller
,
1355 mock_disconnect_model
,
1357 mock_get_controller
,
1359 mock_get_controller
.return_value
= juju
.controller
.Controller()
1360 mock_get_model
.return_value
= juju
.model
.Model()
1362 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1363 mock_consume
.assert_called_once()
1364 mock_disconnect_model
.assert_called_once()
1365 mock_disconnect_controller
.assert_called_once()
1367 def test_parsing_error_exception(
1370 mock_disconnect_controller
,
1371 mock_disconnect_model
,
1373 mock_get_controller
,
1375 mock_get_controller
.return_value
= juju
.controller
.Controller()
1376 mock_get_model
.return_value
= juju
.model
.Model()
1377 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1379 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1380 self
.loop
.run_until_complete(
1381 self
.libjuju
.consume("offer_url", "model_name")
1383 mock_consume
.assert_called_once()
1384 mock_disconnect_model
.assert_called_once()
1385 mock_disconnect_controller
.assert_called_once()
1387 def test_juju_error_exception(
1390 mock_disconnect_controller
,
1391 mock_disconnect_model
,
1393 mock_get_controller
,
1395 mock_get_controller
.return_value
= juju
.controller
.Controller()
1396 mock_get_model
.return_value
= juju
.model
.Model()
1397 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1399 with self
.assertRaises(juju
.errors
.JujuError
):
1400 self
.loop
.run_until_complete(
1401 self
.libjuju
.consume("offer_url", "model_name")
1403 mock_consume
.assert_called_once()
1404 mock_disconnect_model
.assert_called_once()
1405 mock_disconnect_controller
.assert_called_once()
1407 def test_juju_api_error_exception(
1410 mock_disconnect_controller
,
1411 mock_disconnect_model
,
1413 mock_get_controller
,
1415 mock_get_controller
.return_value
= juju
.controller
.Controller()
1416 mock_get_model
.return_value
= juju
.model
.Model()
1417 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1418 {"error": "", "response": "", "request-id": ""}
1421 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1422 self
.loop
.run_until_complete(
1423 self
.libjuju
.consume("offer_url", "model_name")
1425 mock_consume
.assert_called_once()
1426 mock_disconnect_model
.assert_called_once()
1427 mock_disconnect_controller
.assert_called_once()
1430 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1431 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1432 class AddK8sTest(LibjujuTestCase
):
1434 super(AddK8sTest
, self
).setUp()
1436 rbac_id
= generate_rbac_id()
1438 client_cert_data
= "cert"
1439 configuration
= kubernetes
.client
.configuration
.Configuration()
1440 storage_class
= "storage_class"
1441 credential_name
= name
1443 self
._add
_k
8s
_args
= {
1447 "client_cert_data": client_cert_data
,
1448 "configuration": configuration
,
1449 "storage_class": storage_class
,
1450 "credential_name": credential_name
,
1453 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1454 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1455 mock_add_cloud
.assert_called_once()
1456 mock_get_k8s_cloud_credential
.assert_called_once()
1458 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1459 mock_add_cloud
.side_effect
= Exception()
1460 with self
.assertRaises(Exception):
1461 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1462 mock_add_cloud
.assert_called_once()
1463 mock_get_k8s_cloud_credential
.assert_called_once()
1465 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1466 self
._add
_k
8s
_args
["name"] = ""
1467 with self
.assertRaises(Exception):
1468 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1469 mock_add_cloud
.assert_not_called()
1471 def test_add_k8s_missing_storage_name(
1472 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1474 self
._add
_k
8s
_args
["storage_class"] = ""
1475 with self
.assertRaises(Exception):
1476 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1477 mock_add_cloud
.assert_not_called()
1479 def test_add_k8s_missing_configuration_keys(
1480 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1482 self
._add
_k
8s
_args
["configuration"] = None
1483 with self
.assertRaises(Exception):
1484 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1485 mock_add_cloud
.assert_not_called()
1488 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1489 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1490 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1491 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1492 class AddCloudTest(LibjujuTestCase
):
1494 super(AddCloudTest
, self
).setUp()
1495 self
.cloud
= juju
.client
.client
.Cloud()
1496 self
.credential
= juju
.client
.client
.CloudCredential()
1498 def test_add_cloud_with_credential(
1500 mock_add_credential
,
1502 mock_disconnect_controller
,
1503 mock_get_controller
,
1505 mock_get_controller
.return_value
= juju
.controller
.Controller()
1507 cloud
= self
.loop
.run_until_complete(
1508 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1510 self
.assertEqual(cloud
, self
.cloud
)
1511 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1512 mock_add_credential
.assert_called_once_with(
1513 "cloud", credential
=self
.credential
, cloud
="cloud"
1515 mock_disconnect_controller
.assert_called_once()
1517 def test_add_cloud_no_credential(
1519 mock_add_credential
,
1521 mock_disconnect_controller
,
1522 mock_get_controller
,
1524 mock_get_controller
.return_value
= juju
.controller
.Controller()
1526 cloud
= self
.loop
.run_until_complete(
1527 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1529 self
.assertEqual(cloud
, self
.cloud
)
1530 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1531 mock_add_credential
.assert_not_called()
1532 mock_disconnect_controller
.assert_called_once()
1534 def test_add_cloud_exception(
1536 mock_add_credential
,
1538 mock_disconnect_controller
,
1539 mock_get_controller
,
1541 mock_get_controller
.return_value
= juju
.controller
.Controller()
1542 mock_add_cloud
.side_effect
= Exception()
1543 with self
.assertRaises(Exception):
1544 self
.loop
.run_until_complete(
1545 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1548 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1549 mock_add_credential
.assert_not_called()
1550 mock_disconnect_controller
.assert_called_once()
1552 def test_add_credential_exception(
1554 mock_add_credential
,
1556 mock_disconnect_controller
,
1557 mock_get_controller
,
1559 mock_get_controller
.return_value
= juju
.controller
.Controller()
1560 mock_add_credential
.side_effect
= Exception()
1561 with self
.assertRaises(Exception):
1562 self
.loop
.run_until_complete(
1563 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1566 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1567 mock_add_credential
.assert_called_once_with(
1568 "cloud", credential
=self
.credential
, cloud
="cloud"
1570 mock_disconnect_controller
.assert_called_once()
1573 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1574 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1575 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1576 class RemoveCloudTest(LibjujuTestCase
):
1578 super(RemoveCloudTest
, self
).setUp()
1580 def test_remove_cloud(
1583 mock_disconnect_controller
,
1584 mock_get_controller
,
1586 mock_get_controller
.return_value
= juju
.controller
.Controller()
1588 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1589 mock_remove_cloud
.assert_called_once_with("cloud")
1590 mock_disconnect_controller
.assert_called_once()
1592 def test_remove_cloud_exception(
1595 mock_disconnect_controller
,
1596 mock_get_controller
,
1598 mock_get_controller
.return_value
= juju
.controller
.Controller()
1599 mock_remove_cloud
.side_effect
= Exception()
1601 with self
.assertRaises(Exception):
1602 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1603 mock_remove_cloud
.assert_called_once_with("cloud")
1604 mock_disconnect_controller
.assert_called_once()
1607 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1608 class GetK8sCloudCredentials(LibjujuTestCase
):
1610 super(GetK8sCloudCredentials
, self
).setUp()
1611 self
.cert_data
= "cert"
1612 self
.token
= "token"
1614 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1615 def test_not_supported(self
, mock_exception
, mock_configuration
):
1616 mock_configuration
.username
= ""
1617 mock_configuration
.password
= ""
1618 mock_configuration
.ssl_ca_cert
= None
1619 mock_configuration
.cert_file
= None
1620 mock_configuration
.key_file
= None
1621 exception_raised
= False
1623 self
.cert_data
= None
1625 _
= self
.libjuju
.get_k8s_cloud_credential(
1630 except JujuInvalidK8sConfiguration
as e
:
1631 exception_raised
= True
1634 "authentication method not supported",
1636 self
.assertTrue(exception_raised
)
1638 def test_user_pass(self
, mock_configuration
):
1639 mock_configuration
.username
= "admin"
1640 mock_configuration
.password
= "admin"
1641 mock_configuration
.ssl_ca_cert
= None
1642 mock_configuration
.cert_file
= None
1643 mock_configuration
.key_file
= None
1645 self
.cert_data
= None
1646 credential
= self
.libjuju
.get_k8s_cloud_credential(
1653 juju
.client
._definitions
.CloudCredential(
1654 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1658 def test_user_pass_with_cert(self
, mock_configuration
):
1659 mock_configuration
.username
= "admin"
1660 mock_configuration
.password
= "admin"
1661 mock_configuration
.ssl_ca_cert
= None
1662 mock_configuration
.cert_file
= None
1663 mock_configuration
.key_file
= None
1665 credential
= self
.libjuju
.get_k8s_cloud_credential(
1672 juju
.client
._definitions
.CloudCredential(
1674 "ClientCertificateData": self
.cert_data
,
1675 "username": "admin",
1676 "password": "admin",
1678 auth_type
="userpasswithcert",
1682 def test_user_no_pass(self
, mock_configuration
):
1683 mock_configuration
.username
= "admin"
1684 mock_configuration
.password
= ""
1685 mock_configuration
.ssl_ca_cert
= None
1686 mock_configuration
.cert_file
= None
1687 mock_configuration
.key_file
= None
1689 self
.cert_data
= None
1690 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1691 credential
= self
.libjuju
.get_k8s_cloud_credential(
1698 juju
.client
._definitions
.CloudCredential(
1699 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1702 mock_debug
.assert_called_once_with(
1703 "credential for user admin has empty password"
1706 def test_cert(self
, mock_configuration
):
1707 mock_configuration
.username
= ""
1708 mock_configuration
.password
= ""
1709 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1710 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1711 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1712 ssl_ca_cert_file
.write("cacert")
1713 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1714 mock_configuration
.cert_file
= None
1715 mock_configuration
.key_file
= None
1716 credential
= self
.libjuju
.get_k8s_cloud_credential(
1723 juju
.client
._definitions
.CloudCredential(
1724 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1725 auth_type
="certificate",
1729 # TODO: Fix this test when oauth authentication is supported
1730 # def test_oauth2(self, mock_configuration):
1731 # mock_configuration.username = ""
1732 # mock_configuration.password = ""
1733 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1734 # key = tempfile.NamedTemporaryFile()
1735 # with open(key.name, "w") as key_file:
1736 # key_file.write("key")
1737 # mock_configuration.ssl_ca_cert = None
1738 # mock_configuration.cert_file = None
1739 # mock_configuration.key_file = key.name
1740 # credential = self.libjuju.get_k8s_cloud_credential(
1741 # mock_configuration,
1747 # juju.client._definitions.CloudCredential(
1748 # attrs={"ClientKeyData": "key", "Token": "Token"},
1749 # auth_type="oauth2",
1753 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1754 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1755 # mock_configuration.username = ""
1756 # mock_configuration.password = ""
1757 # key = tempfile.NamedTemporaryFile()
1758 # with open(key.name, "w") as key_file:
1759 # key_file.write("key")
1760 # mock_configuration.ssl_ca_cert = None
1761 # mock_configuration.cert_file = None
1762 # mock_configuration.key_file = key.name
1763 # exception_raised = False
1765 # _ = self.libjuju.get_k8s_cloud_credential(
1766 # mock_configuration,
1770 # except JujuInvalidK8sConfiguration as e:
1771 # exception_raised = True
1774 # "missing token for auth type oauth2",
1776 # self.assertTrue(exception_raised)
1778 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1779 mock_configuration
.username
= "admin"
1780 mock_configuration
.password
= "pass"
1781 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1782 mock_configuration
.ssl_ca_cert
= None
1783 mock_configuration
.cert_file
= None
1784 mock_configuration
.key_file
= None
1785 exception_raised
= False
1787 _
= self
.libjuju
.get_k8s_cloud_credential(
1792 except JujuInvalidK8sConfiguration
as e
:
1793 exception_raised
= True
1796 "Cannot set both token and user/pass",
1798 self
.assertTrue(exception_raised
)