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
,
37 class LibjujuTestCase(asynctest
.TestCase
):
38 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
39 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
40 @asynctest.mock
.patch("juju.controller.Controller.connection")
41 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
44 mock__get_api_endpoints_db
=None,
47 mock_update_endpoints
=None,
49 loop
= asyncio
.get_event_loop()
51 mock__get_api_endpoints_db
.return_value
= ["127.0.0.1:17070"]
52 endpoints
= "127.0.0.1:17070"
56 -----BEGIN CERTIFICATE-----
58 -----END CERTIFICATE-----"""
59 self
.libjuju
= Libjuju(
61 "192.168.0.155:17070",
69 apt_mirror
="192.168.0.100",
70 enable_os_upgrade
=True,
72 logging
.disable(logging
.CRITICAL
)
73 loop
.run_until_complete(self
.libjuju
.disconnect())
76 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._create_health_check_task")
77 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
78 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
79 class LibjujuInitTestCase(asynctest
.TestCase
):
81 self
.loop
= asyncio
.get_event_loop()
82 self
.n2vc
= FakeN2VC()
83 self
.endpoint
= "192.168.100.100:17070"
84 self
.username
= "admin"
85 self
.password
= "secret"
87 -----BEGIN CERTIFICATE-----
89 -----END CERTIFICATE-----"""
91 def test_endpoint_not_in_db(
93 mock__get_api_endpoints_db
,
94 mock_update_endpoints
,
95 mock_create_health_check_task
,
97 mock__get_api_endpoints_db
.return_value
= ["another_ip"]
100 "192.168.0.155:17070",
108 apt_mirror
="192.168.0.100",
109 enable_os_upgrade
=True,
111 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
112 mock__get_api_endpoints_db
.assert_called_once()
114 def test_endpoint_in_db(
116 mock__get_api_endpoints_db
,
117 mock_update_endpoints
,
118 mock_create_health_check_task
,
120 mock__get_api_endpoints_db
.return_value
= [self
.endpoint
, "another_ip"]
123 "192.168.0.155:17070",
131 apt_mirror
="192.168.0.100",
132 enable_os_upgrade
=True,
134 mock_update_endpoints
.assert_not_called()
135 mock__get_api_endpoints_db
.assert_called_once()
137 def test_no_db_endpoints(
139 mock__get_api_endpoints_db
,
140 mock_update_endpoints
,
141 mock_create_health_check_task
,
143 mock__get_api_endpoints_db
.return_value
= None
146 "192.168.0.155:17070",
154 apt_mirror
="192.168.0.100",
155 enable_os_upgrade
=True,
157 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
158 mock__get_api_endpoints_db
.assert_called_once()
161 @asynctest.mock
.patch("juju.controller.Controller.connect")
162 @asynctest.mock
.patch(
163 "juju.controller.Controller.api_endpoints",
164 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
166 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
167 class GetControllerTest(LibjujuTestCase
):
169 super(GetControllerTest
, self
).setUp()
171 def test_diff_endpoint(
172 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
174 self
.libjuju
.endpoints
= []
175 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
176 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
177 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
179 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
182 mock_disconnect_controller
,
183 mock__update_api_endpoints_db
,
187 self
.libjuju
.endpoints
= []
188 mock__update_api_endpoints_db
.side_effect
= Exception()
190 with self
.assertRaises(JujuControllerFailedConnecting
):
191 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
192 self
.assertIsNone(controller
)
193 mock_disconnect_controller
.assert_called_once()
195 def test_same_endpoint_get_controller(
196 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
198 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
199 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
200 mock__update_api_endpoints_db
.assert_not_called()
201 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
204 class DisconnectTest(LibjujuTestCase
):
206 super(DisconnectTest
, self
).setUp()
208 @asynctest.mock
.patch("juju.model.Model.disconnect")
209 def test_disconnect_model(self
, mock_disconnect
):
210 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
211 mock_disconnect
.assert_called_once()
213 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
214 def test_disconnect_controller(self
, mock_disconnect
):
215 self
.loop
.run_until_complete(
216 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
218 mock_disconnect
.assert_called_once()
221 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
222 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
223 @asynctest.mock
.patch("juju.controller.Controller.add_model")
224 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
225 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
226 class AddModelTest(LibjujuTestCase
):
228 super(AddModelTest
, self
).setUp()
230 def test_existing_model(
232 mock_disconnect_model
,
233 mock_disconnect_controller
,
238 mock_model_exists
.return_value
= True
240 with self
.assertRaises(JujuModelAlreadyExists
):
241 self
.loop
.run_until_complete(
242 self
.libjuju
.add_model("existing_model", "cloud")
245 mock_disconnect_controller
.assert_called()
247 # TODO Check two job executing at the same time and one returning without doing anything.
249 def test_non_existing_model(
251 mock_disconnect_model
,
252 mock_disconnect_controller
,
257 mock_model_exists
.return_value
= False
258 mock_get_controller
.return_value
= juju
.controller
.Controller()
260 self
.loop
.run_until_complete(
261 self
.libjuju
.add_model("nonexisting_model", "cloud")
264 mock_add_model
.assert_called_once()
265 mock_disconnect_controller
.assert_called()
266 mock_disconnect_model
.assert_called()
269 @asynctest.mock
.patch("juju.controller.Controller.get_model")
270 class GetModelTest(LibjujuTestCase
):
272 super(GetModelTest
, self
).setUp()
278 mock_get_model
.return_value
= juju
.model
.Model()
279 model
= self
.loop
.run_until_complete(
280 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
282 self
.assertIsInstance(model
, juju
.model
.Model
)
285 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
286 @asynctest.mock
.patch("juju.controller.Controller.list_models")
287 class ModelExistsTest(LibjujuTestCase
):
289 super(ModelExistsTest
, self
).setUp()
291 async def test_existing_model(
296 mock_list_models
.return_value
= ["existing_model"]
298 await self
.libjuju
.model_exists(
299 "existing_model", juju
.controller
.Controller()
303 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
304 async def test_no_controller(
306 mock_disconnect_controller
,
310 mock_list_models
.return_value
= ["existing_model"]
311 mock_get_controller
.return_value
= juju
.controller
.Controller()
312 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
313 mock_disconnect_controller
.assert_called_once()
315 async def test_non_existing_model(
320 mock_list_models
.return_value
= ["existing_model"]
322 await self
.libjuju
.model_exists(
323 "not_existing_model", juju
.controller
.Controller()
328 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
329 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
330 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
331 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
332 @asynctest.mock
.patch("juju.model.Model.get_status")
333 class GetModelStatusTest(LibjujuTestCase
):
335 super(GetModelStatusTest
, self
).setUp()
340 mock_disconnect_controller
,
341 mock_disconnect_model
,
345 mock_get_model
.return_value
= juju
.model
.Model()
346 mock_get_status
.return_value
= {"status"}
348 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
350 mock_get_status
.assert_called_once()
351 mock_disconnect_controller
.assert_called_once()
352 mock_disconnect_model
.assert_called_once()
354 self
.assertEqual(status
, {"status"})
359 mock_disconnect_controller
,
360 mock_disconnect_model
,
364 mock_get_model
.return_value
= juju
.model
.Model()
365 mock_get_status
.side_effect
= Exception()
367 with self
.assertRaises(Exception):
368 status
= self
.loop
.run_until_complete(
369 self
.libjuju
.get_model_status("model")
372 mock_disconnect_controller
.assert_called_once()
373 mock_disconnect_model
.assert_called_once()
375 self
.assertIsNone(status
)
378 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
379 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
380 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
381 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
382 @asynctest.mock
.patch("juju.model.Model.get_machines")
383 @asynctest.mock
.patch("juju.model.Model.add_machine")
384 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
385 class CreateMachineTest(LibjujuTestCase
):
387 super(CreateMachineTest
, self
).setUp()
389 def test_existing_machine(
394 mock_disconnect_controller
,
395 mock_disconnect_model
,
399 mock_get_model
.return_value
= juju
.model
.Model()
400 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
401 machine
, bool_res
= self
.loop
.run_until_complete(
402 self
.libjuju
.create_machine("model", "existing_machine")
405 self
.assertIsInstance(machine
, FakeMachine
)
406 self
.assertFalse(bool_res
)
408 mock_disconnect_controller
.assert_called()
409 mock_disconnect_model
.assert_called()
411 def test_non_existing_machine(
416 mock_disconnect_controller
,
417 mock_disconnect_model
,
423 mock_get_model
.return_value
= juju
.model
.Model()
424 with self
.assertRaises(JujuMachineNotFound
):
425 machine
, bool_res
= self
.loop
.run_until_complete(
426 self
.libjuju
.create_machine("model", "non_existing_machine")
428 self
.assertIsNone(machine
)
429 self
.assertIsNone(bool_res
)
431 mock_disconnect_controller
.assert_called()
432 mock_disconnect_model
.assert_called()
439 mock_disconnect_controller
,
440 mock_disconnect_model
,
444 mock_get_model
.return_value
= juju
.model
.Model()
445 mock_add_machine
.return_value
= FakeMachine()
447 machine
, bool_res
= self
.loop
.run_until_complete(
448 self
.libjuju
.create_machine("model")
451 self
.assertIsInstance(machine
, FakeMachine
)
452 self
.assertTrue(bool_res
)
454 mock_wait_for
.assert_called_once()
455 mock_add_machine
.assert_called_once()
457 mock_disconnect_controller
.assert_called()
458 mock_disconnect_model
.assert_called()
461 # TODO test provision machine
464 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
465 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
466 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
467 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
468 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
469 @asynctest.mock
.patch("juju.model.Model.deploy")
470 class DeployTest(LibjujuTestCase
):
472 super(DeployTest
, self
).setUp()
478 mock_disconnect_controller
,
479 mock_disconnect_model
,
483 mock_get_model
.return_value
= juju
.model
.Model()
484 self
.loop
.run_until_complete(
485 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
487 mock_deploy
.assert_called_once()
488 mock_wait_for_model
.assert_called_once()
489 mock_disconnect_controller
.assert_called_once()
490 mock_disconnect_model
.assert_called_once()
492 def test_deploy_no_wait(
496 mock_disconnect_controller
,
497 mock_disconnect_model
,
501 mock_get_model
.return_value
= juju
.model
.Model()
502 self
.loop
.run_until_complete(
503 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
505 mock_deploy
.assert_called_once()
506 mock_wait_for_model
.assert_not_called()
507 mock_disconnect_controller
.assert_called_once()
508 mock_disconnect_model
.assert_called_once()
510 def test_deploy_exception(
514 mock_disconnect_controller
,
515 mock_disconnect_model
,
519 mock_deploy
.side_effect
= Exception()
520 mock_get_model
.return_value
= juju
.model
.Model()
521 with self
.assertRaises(Exception):
522 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
523 mock_deploy
.assert_called_once()
524 mock_wait_for_model
.assert_not_called()
525 mock_disconnect_controller
.assert_called_once()
526 mock_disconnect_model
.assert_called_once()
529 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
530 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
531 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
532 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
533 @asynctest.mock
.patch(
534 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
536 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
537 @asynctest.mock
.patch("juju.model.Model.deploy")
538 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
539 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
540 class DeployCharmTest(LibjujuTestCase
):
542 super(DeployCharmTest
, self
).setUp()
544 def test_existing_app(
551 mock_disconnect_controller
,
552 mock_disconnect_model
,
556 mock_get_model
.return_value
= juju
.model
.Model()
557 mock_applications
.return_value
= {"existing_app"}
560 with self
.assertRaises(JujuApplicationExists
):
561 application
= self
.loop
.run_until_complete(
562 self
.libjuju
.deploy_charm(
569 self
.assertIsNone(application
)
571 mock_disconnect_controller
.assert_called()
572 mock_disconnect_model
.assert_called()
574 def test_non_existing_machine(
581 mock_disconnect_controller
,
582 mock_disconnect_model
,
586 mock_get_model
.return_value
= juju
.model
.Model()
587 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
589 with self
.assertRaises(JujuMachineNotFound
):
590 application
= self
.loop
.run_until_complete(
591 self
.libjuju
.deploy_charm(
599 self
.assertIsNone(application
)
601 mock_disconnect_controller
.assert_called()
602 mock_disconnect_model
.assert_called()
611 mock_disconnect_controller
,
612 mock_disconnect_model
,
616 mock_get_model
.return_value
= juju
.model
.Model()
617 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
618 mock_create_machine
.return_value
= (FakeMachine(), "other")
619 mock_deploy
.return_value
= FakeApplication()
620 application
= self
.loop
.run_until_complete(
621 self
.libjuju
.deploy_charm(
630 self
.assertIsInstance(application
, FakeApplication
)
632 mock_deploy
.assert_called_once()
633 mock_wait_for
.assert_called_once()
635 mock_create_machine
.assert_called_once()
637 mock_disconnect_controller
.assert_called()
638 mock_disconnect_model
.assert_called()
647 mock_disconnect_controller
,
648 mock_disconnect_model
,
652 mock_get_model
.return_value
= juju
.model
.Model()
653 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
654 mock_deploy
.return_value
= FakeApplication()
655 application
= self
.loop
.run_until_complete(
656 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
659 self
.assertIsInstance(application
, FakeApplication
)
661 mock_deploy
.assert_called_once()
662 mock_wait_for
.assert_called_once()
664 mock_disconnect_controller
.assert_called()
665 mock_disconnect_model
.assert_called()
668 @asynctest.mock
.patch(
669 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
671 class GetApplicationTest(LibjujuTestCase
):
673 super(GetApplicationTest
, self
).setUp()
675 def test_existing_application(
679 mock_applications
.return_value
= {"existing_app": "exists"}
680 model
= juju
.model
.Model()
681 result
= self
.libjuju
._get
_application
(model
, "existing_app")
682 self
.assertEqual(result
, "exists")
684 def test_non_existing_application(
688 mock_applications
.return_value
= {"existing_app": "exists"}
689 model
= juju
.model
.Model()
690 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
691 self
.assertIsNone(result
)
694 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
695 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
696 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
697 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
698 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
699 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
700 @asynctest.mock
.patch("juju.model.Model.get_action_output")
701 @asynctest.mock
.patch("juju.model.Model.get_action_status")
702 class ExecuteActionTest(LibjujuTestCase
):
704 super(ExecuteActionTest
, self
).setUp()
706 def test_no_application(
708 mock_get_action_status
,
709 mock_get_action_output
,
711 mock__get_application
,
712 mock_disconnect_controller
,
713 mock_disconnect_model
,
717 mock__get_application
.return_value
= None
718 mock_get_model
.return_value
= juju
.model
.Model()
721 with self
.assertRaises(JujuApplicationNotFound
):
722 output
, status
= self
.loop
.run_until_complete(
723 self
.libjuju
.execute_action(
729 self
.assertIsNone(output
)
730 self
.assertIsNone(status
)
732 mock_disconnect_controller
.assert_called()
733 mock_disconnect_model
.assert_called()
737 mock_get_action_status
,
738 mock_get_action_output
,
740 mock__get_application
,
741 mock_disconnect_controller
,
742 mock_disconnect_model
,
747 mock_get_model
.return_value
= juju
.model
.Model()
748 mock__get_application
.return_value
= FakeApplication()
751 with self
.assertRaises(JujuActionNotFound
):
752 output
, status
= self
.loop
.run_until_complete(
753 self
.libjuju
.execute_action(
759 self
.assertIsNone(output
)
760 self
.assertIsNone(status
)
762 mock_disconnect_controller
.assert_called()
763 mock_disconnect_model
.assert_called()
765 @asynctest.mock
.patch("asyncio.sleep")
766 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
769 mock_is_leader_from_status
,
771 mock_get_action_status
,
772 mock_get_action_output
,
774 mock__get_application
,
775 mock_disconnect_controller
,
776 mock_disconnect_model
,
780 mock_get_model
.return_value
= juju
.model
.Model()
781 mock__get_application
.return_value
= FakeApplication()
782 mock_is_leader_from_status
.return_value
= False
785 with self
.assertRaises(JujuLeaderUnitNotFound
):
786 output
, status
= self
.loop
.run_until_complete(
787 self
.libjuju
.execute_action(
793 self
.assertIsNone(output
)
794 self
.assertIsNone(status
)
796 mock_disconnect_controller
.assert_called()
797 mock_disconnect_model
.assert_called()
799 def test_succesful_exec(
801 mock_get_action_status
,
802 mock_get_action_output
,
804 mock__get_application
,
805 mock_disconnect_controller
,
806 mock_disconnect_model
,
810 mock_get_model
.return_value
= juju
.model
.Model()
811 mock__get_application
.return_value
= FakeApplication()
812 mock_get_action_output
.return_value
= "output"
813 mock_get_action_status
.return_value
= {"id": "status"}
814 output
, status
= self
.loop
.run_until_complete(
815 self
.libjuju
.execute_action("app", "model", "existing_action")
817 self
.assertEqual(output
, "output")
818 self
.assertEqual(status
, "status")
820 mock_wait_for
.assert_called_once()
822 mock_disconnect_controller
.assert_called()
823 mock_disconnect_model
.assert_called()
826 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
827 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
828 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
829 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
830 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
831 class GetActionTest(LibjujuTestCase
):
833 super(GetActionTest
, self
).setUp()
837 mock_get_application
,
838 mock_disconnect_controller
,
839 mock_disconnect_model
,
843 mock_get_application
.side_effect
= Exception()
845 with self
.assertRaises(Exception):
846 actions
= self
.loop
.run_until_complete(
847 self
.libjuju
.get_actions("app", "model")
850 self
.assertIsNone(actions
)
851 mock_disconnect_controller
.assert_called_once()
852 mock_disconnect_model
.assert_called_once()
856 mock_get_application
,
857 mock_disconnect_controller
,
858 mock_disconnect_model
,
862 mock_get_application
.return_value
= FakeApplication()
864 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
866 self
.assertEqual(actions
, ["existing_action"])
868 mock_get_controller
.assert_called_once()
869 mock_get_model
.assert_called_once()
870 mock_disconnect_controller
.assert_called_once()
871 mock_disconnect_model
.assert_called_once()
874 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
875 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
876 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
877 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
878 @asynctest.mock
.patch("juju.application.Application.get_metrics")
879 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
880 class GetMetricsTest(LibjujuTestCase
):
882 super(GetMetricsTest
, self
).setUp()
884 def test_get_metrics_success(
886 mock_get_application
,
888 mock_disconnect_controller
,
889 mock_disconnect_model
,
893 mock_get_application
.return_value
= FakeApplication()
894 mock_get_model
.return_value
= juju
.model
.Model()
896 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
898 mock_disconnect_controller
.assert_called_once()
899 mock_disconnect_model
.assert_called_once()
901 def test_get_metrics_exception(
903 mock_get_application
,
905 mock_disconnect_controller
,
906 mock_disconnect_model
,
910 mock_get_model
.return_value
= juju
.model
.Model()
911 mock_get_metrics
.side_effect
= Exception()
912 with self
.assertRaises(Exception):
913 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
915 mock_disconnect_controller
.assert_called_once()
916 mock_disconnect_model
.assert_called_once()
918 def test_missing_args_exception(
920 mock_get_application
,
922 mock_disconnect_controller
,
923 mock_disconnect_model
,
927 mock_get_model
.return_value
= juju
.model
.Model()
929 with self
.assertRaises(Exception):
930 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
932 mock_get_controller
.assert_not_called()
933 mock_get_model
.assert_not_called()
934 mock_disconnect_controller
.assert_not_called()
935 mock_disconnect_model
.assert_not_called()
938 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
939 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
940 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
941 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
942 @asynctest.mock
.patch("juju.model.Model.add_relation")
943 class AddRelationTest(LibjujuTestCase
):
945 super(AddRelationTest
, self
).setUp()
947 @asynctest.mock
.patch("logging.Logger.warning")
952 mock_disconnect_controller
,
953 mock_disconnect_model
,
957 # TODO in libjuju.py should this fail only with a log message?
958 result
= {"error": "not found", "response": "response", "request-id": 1}
960 mock_get_model
.return_value
= juju
.model
.Model()
961 mock_add_relation
.side_effect
= JujuAPIError(result
)
963 self
.loop
.run_until_complete(
964 self
.libjuju
.add_relation(
971 mock_warning
.assert_called_with("Relation not found: not found")
972 mock_disconnect_controller
.assert_called_once()
973 mock_disconnect_model
.assert_called_once()
975 @asynctest.mock
.patch("logging.Logger.warning")
976 def test_already_exists(
980 mock_disconnect_controller
,
981 mock_disconnect_model
,
985 # TODO in libjuju.py should this fail silently?
986 result
= {"error": "already exists", "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 already exists: already exists")
1000 mock_disconnect_controller
.assert_called_once()
1001 mock_disconnect_model
.assert_called_once()
1006 mock_disconnect_controller
,
1007 mock_disconnect_model
,
1009 mock_get_controller
,
1011 mock_get_model
.return_value
= juju
.model
.Model()
1012 result
= {"error": "", "response": "response", "request-id": 1}
1013 mock_add_relation
.side_effect
= JujuAPIError(result
)
1015 with self
.assertRaises(JujuAPIError
):
1016 self
.loop
.run_until_complete(
1017 self
.libjuju
.add_relation(
1024 mock_disconnect_controller
.assert_called_once()
1025 mock_disconnect_model
.assert_called_once()
1030 mock_disconnect_controller
,
1031 mock_disconnect_model
,
1033 mock_get_controller
,
1035 mock_get_model
.return_value
= juju
.model
.Model()
1037 self
.loop
.run_until_complete(
1038 self
.libjuju
.add_relation(
1045 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1046 mock_disconnect_controller
.assert_called_once()
1047 mock_disconnect_model
.assert_called_once()
1052 mock_disconnect_controller
,
1053 mock_disconnect_model
,
1055 mock_get_controller
,
1057 mock_get_model
.return_value
= juju
.model
.Model()
1059 self
.loop
.run_until_complete(
1060 self
.libjuju
.add_relation(
1067 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1068 mock_disconnect_controller
.assert_called_once()
1069 mock_disconnect_model
.assert_called_once()
1072 # TODO destroy_model testcase
1075 # @asynctest.mock.patch("juju.model.Model.get_machines")
1076 # @asynctest.mock.patch("logging.Logger.debug")
1077 # class DestroyMachineTest(LibjujuTestCase):
1079 # super(DestroyMachineTest, self).setUp()
1081 # def test_success_manual_machine(
1082 # self, mock_debug, mock_get_machines,
1084 # mock_get_machines.side_effect = [
1085 # {"machine": FakeManualMachine()},
1086 # {"machine": FakeManualMachine()},
1089 # self.loop.run_until_complete(
1090 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1093 # asynctest.call("Waiting for machine machine is destroyed"),
1094 # asynctest.call("Machine destroyed: machine"),
1096 # mock_debug.assert_has_calls(calls)
1098 # def test_no_machine(
1099 # self, mock_debug, mock_get_machines,
1101 # mock_get_machines.return_value = {}
1102 # self.loop.run_until_complete(
1103 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1105 # mock_debug.assert_called_with("Machine not found: machine")
1108 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1109 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1112 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1113 class ConfigureApplicationTest(LibjujuTestCase
):
1115 super(ConfigureApplicationTest
, self
).setUp()
1119 mock_get_application
,
1120 mock_disconnect_controller
,
1121 mock_disconnect_model
,
1123 mock_get_controller
,
1126 mock_get_application
.return_value
= FakeApplication()
1128 self
.loop
.run_until_complete(
1129 self
.libjuju
.configure_application(
1135 mock_get_application
.assert_called_once()
1136 mock_disconnect_controller
.assert_called_once()
1137 mock_disconnect_model
.assert_called_once()
1141 mock_get_application
,
1142 mock_disconnect_controller
,
1143 mock_disconnect_model
,
1145 mock_get_controller
,
1148 mock_get_application
.side_effect
= Exception()
1150 with self
.assertRaises(Exception):
1151 self
.loop
.run_until_complete(
1152 self
.libjuju
.configure_application(
1158 mock_disconnect_controller
.assert_called_once()
1159 mock_disconnect_model
.assert_called_once()
1161 def test_controller_exception(
1163 mock_get_application
,
1164 mock_disconnect_controller
,
1165 mock_disconnect_model
,
1167 mock_get_controller
,
1170 result
= {"error": "not found", "response": "response", "request-id": 1}
1172 mock_get_controller
.side_effect
= JujuAPIError(result
)
1174 with self
.assertRaises(JujuAPIError
):
1175 self
.loop
.run_until_complete(
1176 self
.libjuju
.configure_application(
1182 mock_get_model
.assert_not_called()
1183 mock_disconnect_controller
.assert_not_called()
1184 mock_disconnect_model
.assert_not_called()
1186 def test_get_model_exception(
1188 mock_get_application
,
1189 mock_disconnect_controller
,
1190 mock_disconnect_model
,
1192 mock_get_controller
,
1195 result
= {"error": "not found", "response": "response", "request-id": 1}
1196 mock_get_model
.side_effect
= JujuAPIError(result
)
1198 with self
.assertRaises(JujuAPIError
):
1199 self
.loop
.run_until_complete(
1200 self
.libjuju
.configure_application(
1206 mock_get_model
.assert_called_once()
1207 mock_disconnect_controller
.assert_called_once()
1208 mock_disconnect_model
.assert_not_called()
1211 # TODO _get_api_endpoints_db test case
1212 # TODO _update_api_endpoints_db test case
1213 # TODO healthcheck test case
1216 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1218 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1219 class ListModelsTest(LibjujuTestCase
):
1221 super(ListModelsTest
, self
).setUp()
1223 def test_containing(
1226 mock_disconnect_controller
,
1227 mock_get_controller
,
1229 mock_get_controller
.return_value
= juju
.controller
.Controller()
1230 mock_list_models
.return_value
= ["existingmodel"]
1231 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1233 mock_disconnect_controller
.assert_called_once()
1234 self
.assertEquals(models
, ["existingmodel"])
1236 def test_not_containing(
1239 mock_disconnect_controller
,
1240 mock_get_controller
,
1242 mock_get_controller
.return_value
= juju
.controller
.Controller()
1243 mock_list_models
.return_value
= ["existingmodel", "model"]
1244 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1246 mock_disconnect_controller
.assert_called_once()
1247 self
.assertEquals(models
, [])
1249 def test_no_contains_arg(
1252 mock_disconnect_controller
,
1253 mock_get_controller
,
1255 mock_get_controller
.return_value
= juju
.controller
.Controller()
1256 mock_list_models
.return_value
= ["existingmodel", "model"]
1257 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1259 mock_disconnect_controller
.assert_called_once()
1260 self
.assertEquals(models
, ["existingmodel", "model"])
1263 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1264 class ModelsExistTest(LibjujuTestCase
):
1266 super(ModelsExistTest
, self
).setUp()
1268 def test_model_names_none(self
, mock_list_models
):
1269 mock_list_models
.return_value
= []
1270 with self
.assertRaises(Exception):
1271 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1273 def test_model_names_empty(self
, mock_list_models
):
1274 mock_list_models
.return_value
= []
1275 with self
.assertRaises(Exception):
1276 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1277 self
.libjuju
.models_exist([])
1280 def test_model_names_not_existing(self
, mock_list_models
):
1281 mock_list_models
.return_value
= ["prometheus", "grafana"]
1282 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1283 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1285 self
.assertFalse(exist
)
1286 self
.assertEqual(non_existing_models
, ["prometheus2"])
1288 def test_model_names_exist(self
, mock_list_models
):
1289 mock_list_models
.return_value
= ["prometheus", "grafana"]
1290 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1291 self
.libjuju
.models_exist(["prometheus", "grafana"])
1293 self
.assertTrue(exist
)
1294 self
.assertEqual(non_existing_models
, [])
1297 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1298 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1299 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1300 class ListOffers(LibjujuTestCase
):
1302 super(ListOffers
, self
).setUp()
1304 def test_disconnect_controller(
1307 mock_disconnect_controller
,
1308 mock_get_controller
,
1310 mock_get_controller
.return_value
= juju
.controller
.Controller()
1311 mock_list_offers
.side_effect
= Exception()
1312 with self
.assertRaises(Exception):
1313 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1314 mock_disconnect_controller
.assert_called_once()
1316 def test_empty_list(
1319 mock_disconnect_controller
,
1320 mock_get_controller
,
1322 mock_get_controller
.return_value
= juju
.controller
.Controller()
1323 mock_list_offers
.return_value
= []
1324 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1325 self
.assertEqual(offers
, [])
1326 mock_disconnect_controller
.assert_called_once()
1328 def test_non_empty_list(
1331 mock_disconnect_controller
,
1332 mock_get_controller
,
1334 mock_get_controller
.return_value
= juju
.controller
.Controller()
1335 mock_list_offers
.return_value
= ["offer"]
1336 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1337 self
.assertEqual(offers
, ["offer"])
1338 mock_disconnect_controller
.assert_called_once()
1341 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1342 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1343 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1344 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1345 @asynctest.mock
.patch("juju.model.Model.consume")
1346 class ConsumeTest(LibjujuTestCase
):
1348 super(ConsumeTest
, self
).setUp()
1353 mock_disconnect_controller
,
1354 mock_disconnect_model
,
1356 mock_get_controller
,
1358 mock_get_controller
.return_value
= juju
.controller
.Controller()
1359 mock_get_model
.return_value
= juju
.model
.Model()
1361 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1362 mock_consume
.assert_called_once()
1363 mock_disconnect_model
.assert_called_once()
1364 mock_disconnect_controller
.assert_called_once()
1366 def test_parsing_error_exception(
1369 mock_disconnect_controller
,
1370 mock_disconnect_model
,
1372 mock_get_controller
,
1374 mock_get_controller
.return_value
= juju
.controller
.Controller()
1375 mock_get_model
.return_value
= juju
.model
.Model()
1376 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1378 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1379 self
.loop
.run_until_complete(
1380 self
.libjuju
.consume("offer_url", "model_name")
1382 mock_consume
.assert_called_once()
1383 mock_disconnect_model
.assert_called_once()
1384 mock_disconnect_controller
.assert_called_once()
1386 def test_juju_error_exception(
1389 mock_disconnect_controller
,
1390 mock_disconnect_model
,
1392 mock_get_controller
,
1394 mock_get_controller
.return_value
= juju
.controller
.Controller()
1395 mock_get_model
.return_value
= juju
.model
.Model()
1396 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1398 with self
.assertRaises(juju
.errors
.JujuError
):
1399 self
.loop
.run_until_complete(
1400 self
.libjuju
.consume("offer_url", "model_name")
1402 mock_consume
.assert_called_once()
1403 mock_disconnect_model
.assert_called_once()
1404 mock_disconnect_controller
.assert_called_once()
1406 def test_juju_api_error_exception(
1409 mock_disconnect_controller
,
1410 mock_disconnect_model
,
1412 mock_get_controller
,
1414 mock_get_controller
.return_value
= juju
.controller
.Controller()
1415 mock_get_model
.return_value
= juju
.model
.Model()
1416 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1417 {"error": "", "response": "", "request-id": ""}
1420 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1421 self
.loop
.run_until_complete(
1422 self
.libjuju
.consume("offer_url", "model_name")
1424 mock_consume
.assert_called_once()
1425 mock_disconnect_model
.assert_called_once()
1426 mock_disconnect_controller
.assert_called_once()
1429 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1430 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1431 class AddK8sTest(LibjujuTestCase
):
1433 super(AddK8sTest
, self
).setUp()
1434 self
.configuration
= kubernetes
.client
.configuration
.Configuration()
1436 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1437 self
.loop
.run_until_complete(
1438 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1440 mock_add_cloud
.assert_called_once()
1441 mock_get_k8s_cloud_credential
.assert_called_once()
1443 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1444 mock_add_cloud
.side_effect
= Exception()
1445 with self
.assertRaises(Exception):
1446 self
.loop
.run_until_complete(
1447 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1449 mock_add_cloud
.assert_called_once()
1450 mock_get_k8s_cloud_credential
.assert_called_once()
1452 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1453 with self
.assertRaises(Exception):
1454 self
.loop
.run_until_complete(
1455 self
.libjuju
.add_k8s("", self
.configuration
, "storage_class")
1457 mock_add_cloud
.assert_not_called()
1459 def test_add_k8s_missing_storage_name(
1460 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1462 with self
.assertRaises(Exception):
1463 self
.loop
.run_until_complete(
1464 self
.libjuju
.add_k8s("cloud", self
.configuration
, "")
1466 mock_add_cloud
.assert_not_called()
1468 def test_add_k8s_missing_configuration_keys(
1469 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1471 with self
.assertRaises(Exception):
1472 self
.loop
.run_until_complete(self
.libjuju
.add_k8s("cloud", None, ""))
1473 mock_add_cloud
.assert_not_called()
1476 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1477 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1478 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1479 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1480 class AddCloudTest(LibjujuTestCase
):
1482 super(AddCloudTest
, self
).setUp()
1483 self
.cloud
= juju
.client
.client
.Cloud()
1484 self
.credential
= juju
.client
.client
.CloudCredential()
1486 def test_add_cloud_with_credential(
1488 mock_add_credential
,
1490 mock_disconnect_controller
,
1491 mock_get_controller
,
1493 mock_get_controller
.return_value
= juju
.controller
.Controller()
1495 cloud
= self
.loop
.run_until_complete(
1496 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1498 self
.assertEqual(cloud
, self
.cloud
)
1499 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1500 mock_add_credential
.assert_called_once_with(
1501 "cloud", credential
=self
.credential
, cloud
="cloud"
1503 mock_disconnect_controller
.assert_called_once()
1505 def test_add_cloud_no_credential(
1507 mock_add_credential
,
1509 mock_disconnect_controller
,
1510 mock_get_controller
,
1512 mock_get_controller
.return_value
= juju
.controller
.Controller()
1514 cloud
= self
.loop
.run_until_complete(
1515 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1517 self
.assertEqual(cloud
, self
.cloud
)
1518 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1519 mock_add_credential
.assert_not_called()
1520 mock_disconnect_controller
.assert_called_once()
1522 def test_add_cloud_exception(
1524 mock_add_credential
,
1526 mock_disconnect_controller
,
1527 mock_get_controller
,
1529 mock_get_controller
.return_value
= juju
.controller
.Controller()
1530 mock_add_cloud
.side_effect
= Exception()
1531 with self
.assertRaises(Exception):
1532 self
.loop
.run_until_complete(
1533 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1536 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1537 mock_add_credential
.assert_not_called()
1538 mock_disconnect_controller
.assert_called_once()
1540 def test_add_credential_exception(
1542 mock_add_credential
,
1544 mock_disconnect_controller
,
1545 mock_get_controller
,
1547 mock_get_controller
.return_value
= juju
.controller
.Controller()
1548 mock_add_credential
.side_effect
= Exception()
1549 with self
.assertRaises(Exception):
1550 self
.loop
.run_until_complete(
1551 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1554 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1555 mock_add_credential
.assert_called_once_with(
1556 "cloud", credential
=self
.credential
, cloud
="cloud"
1558 mock_disconnect_controller
.assert_called_once()
1561 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1562 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1563 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1564 class RemoveCloudTest(LibjujuTestCase
):
1566 super(RemoveCloudTest
, self
).setUp()
1568 def test_remove_cloud(
1571 mock_disconnect_controller
,
1572 mock_get_controller
,
1574 mock_get_controller
.return_value
= juju
.controller
.Controller()
1576 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1577 mock_remove_cloud
.assert_called_once_with("cloud")
1578 mock_disconnect_controller
.assert_called_once()
1580 def test_remove_cloud_exception(
1583 mock_disconnect_controller
,
1584 mock_get_controller
,
1586 mock_get_controller
.return_value
= juju
.controller
.Controller()
1587 mock_remove_cloud
.side_effect
= Exception()
1589 with self
.assertRaises(Exception):
1590 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1591 mock_remove_cloud
.assert_called_once_with("cloud")
1592 mock_disconnect_controller
.assert_called_once()
1595 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1596 class GetK8sCloudCredentials(LibjujuTestCase
):
1598 super(GetK8sCloudCredentials
, self
).setUp()
1600 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1601 def test_not_supported(self
, mock_exception
, mock_configuration
):
1602 mock_configuration
.username
= ""
1603 mock_configuration
.password
= ""
1604 mock_configuration
.ssl_ca_cert
= None
1605 mock_configuration
.cert_file
= None
1606 mock_configuration
.key_file
= None
1607 exception_raised
= False
1609 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1610 except JujuInvalidK8sConfiguration
as e
:
1611 exception_raised
= True
1614 "authentication method not supported",
1616 self
.assertTrue(exception_raised
)
1618 def test_user_pass(self
, mock_configuration
):
1619 mock_configuration
.username
= "admin"
1620 mock_configuration
.password
= "admin"
1621 mock_configuration
.ssl_ca_cert
= None
1622 mock_configuration
.cert_file
= None
1623 mock_configuration
.key_file
= None
1624 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1627 juju
.client
._definitions
.CloudCredential(
1628 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1632 def test_user_no_pass(self
, mock_configuration
):
1633 mock_configuration
.username
= "admin"
1634 mock_configuration
.password
= ""
1635 mock_configuration
.ssl_ca_cert
= None
1636 mock_configuration
.cert_file
= None
1637 mock_configuration
.key_file
= None
1638 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1639 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1642 juju
.client
._definitions
.CloudCredential(
1643 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1646 mock_debug
.assert_called_once_with(
1647 "credential for user admin has empty password"
1650 def test_user_pass_with_cert(self
, mock_configuration
):
1651 mock_configuration
.username
= "admin"
1652 mock_configuration
.password
= "admin"
1653 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1654 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1655 ssl_ca_cert_file
.write("cacert")
1656 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1657 mock_configuration
.cert_file
= None
1658 mock_configuration
.key_file
= None
1659 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1662 juju
.client
._definitions
.CloudCredential(
1664 "username": "admin",
1665 "password": "admin",
1666 "ClientCertificateData": "cacert",
1668 auth_type
="userpasswithcert",
1672 def test_cert(self
, mock_configuration
):
1673 mock_configuration
.username
= ""
1674 mock_configuration
.password
= ""
1675 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1676 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1677 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1678 ssl_ca_cert_file
.write("cacert")
1679 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1680 mock_configuration
.cert_file
= None
1681 mock_configuration
.key_file
= None
1682 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1685 juju
.client
._definitions
.CloudCredential(
1686 attrs
={"ClientCertificateData": "cacert", "Token": "Token"},
1687 auth_type
="certificate",
1691 def test_oauth2(self
, mock_configuration
):
1692 mock_configuration
.username
= ""
1693 mock_configuration
.password
= ""
1694 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1695 key
= tempfile
.NamedTemporaryFile()
1696 with
open(key
.name
, "w") as key_file
:
1697 key_file
.write("key")
1698 mock_configuration
.ssl_ca_cert
= None
1699 mock_configuration
.cert_file
= None
1700 mock_configuration
.key_file
= key
.name
1701 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1704 juju
.client
._definitions
.CloudCredential(
1705 attrs
={"ClientKeyData": "key", "Token": "Token"},
1710 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1711 def test_oauth2_missing_token(self
, mock_exception
, mock_configuration
):
1712 mock_configuration
.username
= ""
1713 mock_configuration
.password
= ""
1714 key
= tempfile
.NamedTemporaryFile()
1715 with
open(key
.name
, "w") as key_file
:
1716 key_file
.write("key")
1717 mock_configuration
.ssl_ca_cert
= None
1718 mock_configuration
.cert_file
= None
1719 mock_configuration
.key_file
= key
.name
1720 exception_raised
= False
1722 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1723 except JujuInvalidK8sConfiguration
as e
:
1724 exception_raised
= True
1727 "missing token for auth type oauth2",
1729 self
.assertTrue(exception_raised
)
1731 def test_unknown_api_key(self
, mock_configuration
):
1732 mock_configuration
.username
= ""
1733 mock_configuration
.password
= ""
1734 mock_configuration
.api_key
= {"authorization": "Bearer Token Wrong"}
1735 mock_configuration
.ssl_ca_cert
= None
1736 mock_configuration
.cert_file
= None
1737 mock_configuration
.key_file
= None
1738 exception_raised
= False
1740 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1741 except JujuInvalidK8sConfiguration
as e
:
1742 exception_raised
= True
1745 "unknown format of api_key",
1747 self
.assertTrue(exception_raised
)
1749 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1750 mock_configuration
.username
= "admin"
1751 mock_configuration
.password
= "pass"
1752 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1753 mock_configuration
.ssl_ca_cert
= None
1754 mock_configuration
.cert_file
= None
1755 mock_configuration
.key_file
= None
1756 exception_raised
= False
1758 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1759 except JujuInvalidK8sConfiguration
as e
:
1760 exception_raised
= True
1763 "Cannot set both token and user/pass",
1765 self
.assertTrue(exception_raised
)