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
,
28 JujuApplicationNotFound
,
30 JujuApplicationExists
,
31 JujuInvalidK8sConfiguration
,
32 JujuLeaderUnitNotFound
,
34 from n2vc
.k8s_juju_conn
import generate_rbac_id
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",
70 logging
.disable(logging
.CRITICAL
)
71 loop
.run_until_complete(self
.libjuju
.disconnect())
74 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._create_health_check_task")
75 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
76 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
77 class LibjujuInitTestCase(asynctest
.TestCase
):
79 self
.loop
= asyncio
.get_event_loop()
80 self
.n2vc
= FakeN2VC()
81 self
.endpoint
= "192.168.100.100:17070"
82 self
.username
= "admin"
83 self
.password
= "secret"
85 -----BEGIN CERTIFICATE-----
87 -----END CERTIFICATE-----"""
89 def test_endpoint_not_in_db(
91 mock__get_api_endpoints_db
,
92 mock_update_endpoints
,
93 mock_create_health_check_task
,
95 mock__get_api_endpoints_db
.return_value
= ["another_ip"]
98 "192.168.0.155:17070",
107 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
108 mock__get_api_endpoints_db
.assert_called_once()
110 def test_endpoint_in_db(
112 mock__get_api_endpoints_db
,
113 mock_update_endpoints
,
114 mock_create_health_check_task
,
116 mock__get_api_endpoints_db
.return_value
= [self
.endpoint
, "another_ip"]
119 "192.168.0.155:17070",
128 mock_update_endpoints
.assert_not_called()
129 mock__get_api_endpoints_db
.assert_called_once()
131 def test_no_db_endpoints(
133 mock__get_api_endpoints_db
,
134 mock_update_endpoints
,
135 mock_create_health_check_task
,
137 mock__get_api_endpoints_db
.return_value
= None
140 "192.168.0.155:17070",
149 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
150 mock__get_api_endpoints_db
.assert_called_once()
153 @asynctest.mock
.patch("juju.controller.Controller.connect")
154 @asynctest.mock
.patch(
155 "juju.controller.Controller.api_endpoints",
156 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
158 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
159 class GetControllerTest(LibjujuTestCase
):
161 super(GetControllerTest
, self
).setUp()
163 def test_diff_endpoint(
164 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
166 self
.libjuju
.endpoints
= []
167 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
168 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
169 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
171 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
174 mock_disconnect_controller
,
175 mock__update_api_endpoints_db
,
179 self
.libjuju
.endpoints
= []
180 mock__update_api_endpoints_db
.side_effect
= Exception()
182 with self
.assertRaises(JujuControllerFailedConnecting
):
183 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
184 self
.assertIsNone(controller
)
185 mock_disconnect_controller
.assert_called_once()
187 def test_same_endpoint_get_controller(
188 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
190 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
191 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
192 mock__update_api_endpoints_db
.assert_not_called()
193 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
196 class DisconnectTest(LibjujuTestCase
):
198 super(DisconnectTest
, self
).setUp()
200 @asynctest.mock
.patch("juju.model.Model.disconnect")
201 def test_disconnect_model(self
, mock_disconnect
):
202 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
203 mock_disconnect
.assert_called_once()
205 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
206 def test_disconnect_controller(self
, mock_disconnect
):
207 self
.loop
.run_until_complete(
208 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
210 mock_disconnect
.assert_called_once()
213 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
214 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
215 @asynctest.mock
.patch("juju.controller.Controller.add_model")
216 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
218 class AddModelTest(LibjujuTestCase
):
220 super(AddModelTest
, self
).setUp()
222 def test_existing_model(
224 mock_disconnect_model
,
225 mock_disconnect_controller
,
230 mock_model_exists
.return_value
= True
232 # This should not raise an exception
233 self
.loop
.run_until_complete(
234 self
.libjuju
.add_model("existing_model", "cloud")
237 mock_disconnect_controller
.assert_called()
239 # TODO Check two job executing at the same time and one returning without doing anything.
241 def test_non_existing_model(
243 mock_disconnect_model
,
244 mock_disconnect_controller
,
249 mock_model_exists
.return_value
= False
250 mock_get_controller
.return_value
= juju
.controller
.Controller()
252 self
.loop
.run_until_complete(
253 self
.libjuju
.add_model("nonexisting_model", "cloud")
256 mock_add_model
.assert_called_once()
257 mock_disconnect_controller
.assert_called()
258 mock_disconnect_model
.assert_called()
261 @asynctest.mock
.patch("juju.controller.Controller.get_model")
262 class GetModelTest(LibjujuTestCase
):
264 super(GetModelTest
, self
).setUp()
270 mock_get_model
.return_value
= juju
.model
.Model()
271 model
= self
.loop
.run_until_complete(
272 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
274 self
.assertIsInstance(model
, juju
.model
.Model
)
277 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
278 @asynctest.mock
.patch("juju.controller.Controller.list_models")
279 class ModelExistsTest(LibjujuTestCase
):
281 super(ModelExistsTest
, self
).setUp()
283 async def test_existing_model(
288 mock_list_models
.return_value
= ["existing_model"]
290 await self
.libjuju
.model_exists(
291 "existing_model", juju
.controller
.Controller()
295 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
296 async def test_no_controller(
298 mock_disconnect_controller
,
302 mock_list_models
.return_value
= ["existing_model"]
303 mock_get_controller
.return_value
= juju
.controller
.Controller()
304 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
305 mock_disconnect_controller
.assert_called_once()
307 async def test_non_existing_model(
312 mock_list_models
.return_value
= ["existing_model"]
314 await self
.libjuju
.model_exists(
315 "not_existing_model", juju
.controller
.Controller()
320 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
321 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
322 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
323 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
324 @asynctest.mock
.patch("juju.model.Model.get_status")
325 class GetModelStatusTest(LibjujuTestCase
):
327 super(GetModelStatusTest
, self
).setUp()
332 mock_disconnect_controller
,
333 mock_disconnect_model
,
337 mock_get_model
.return_value
= juju
.model
.Model()
338 mock_get_status
.return_value
= {"status"}
340 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
342 mock_get_status
.assert_called_once()
343 mock_disconnect_controller
.assert_called_once()
344 mock_disconnect_model
.assert_called_once()
346 self
.assertEqual(status
, {"status"})
351 mock_disconnect_controller
,
352 mock_disconnect_model
,
356 mock_get_model
.return_value
= juju
.model
.Model()
357 mock_get_status
.side_effect
= Exception()
359 with self
.assertRaises(Exception):
360 status
= self
.loop
.run_until_complete(
361 self
.libjuju
.get_model_status("model")
364 mock_disconnect_controller
.assert_called_once()
365 mock_disconnect_model
.assert_called_once()
367 self
.assertIsNone(status
)
370 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
371 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
372 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
373 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
374 @asynctest.mock
.patch("juju.model.Model.get_machines")
375 @asynctest.mock
.patch("juju.model.Model.add_machine")
376 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
377 class CreateMachineTest(LibjujuTestCase
):
379 super(CreateMachineTest
, self
).setUp()
381 def test_existing_machine(
386 mock_disconnect_controller
,
387 mock_disconnect_model
,
391 mock_get_model
.return_value
= juju
.model
.Model()
392 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
393 machine
, bool_res
= self
.loop
.run_until_complete(
394 self
.libjuju
.create_machine("model", "existing_machine")
397 self
.assertIsInstance(machine
, FakeMachine
)
398 self
.assertFalse(bool_res
)
400 mock_disconnect_controller
.assert_called()
401 mock_disconnect_model
.assert_called()
403 def test_non_existing_machine(
408 mock_disconnect_controller
,
409 mock_disconnect_model
,
415 mock_get_model
.return_value
= juju
.model
.Model()
416 with self
.assertRaises(JujuMachineNotFound
):
417 machine
, bool_res
= self
.loop
.run_until_complete(
418 self
.libjuju
.create_machine("model", "non_existing_machine")
420 self
.assertIsNone(machine
)
421 self
.assertIsNone(bool_res
)
423 mock_disconnect_controller
.assert_called()
424 mock_disconnect_model
.assert_called()
431 mock_disconnect_controller
,
432 mock_disconnect_model
,
436 mock_get_model
.return_value
= juju
.model
.Model()
437 mock_add_machine
.return_value
= FakeMachine()
439 machine
, bool_res
= self
.loop
.run_until_complete(
440 self
.libjuju
.create_machine("model")
443 self
.assertIsInstance(machine
, FakeMachine
)
444 self
.assertTrue(bool_res
)
446 mock_wait_for
.assert_called_once()
447 mock_add_machine
.assert_called_once()
449 mock_disconnect_controller
.assert_called()
450 mock_disconnect_model
.assert_called()
453 # TODO test provision machine
456 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
457 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
458 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
459 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
460 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
461 @asynctest.mock
.patch("juju.model.Model.deploy")
462 class DeployTest(LibjujuTestCase
):
464 super(DeployTest
, self
).setUp()
470 mock_disconnect_controller
,
471 mock_disconnect_model
,
475 mock_get_model
.return_value
= juju
.model
.Model()
476 self
.loop
.run_until_complete(
477 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
479 mock_deploy
.assert_called_once()
480 mock_wait_for_model
.assert_called_once()
481 mock_disconnect_controller
.assert_called_once()
482 mock_disconnect_model
.assert_called_once()
484 def test_deploy_no_wait(
488 mock_disconnect_controller
,
489 mock_disconnect_model
,
493 mock_get_model
.return_value
= juju
.model
.Model()
494 self
.loop
.run_until_complete(
495 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
497 mock_deploy
.assert_called_once()
498 mock_wait_for_model
.assert_not_called()
499 mock_disconnect_controller
.assert_called_once()
500 mock_disconnect_model
.assert_called_once()
502 def test_deploy_exception(
506 mock_disconnect_controller
,
507 mock_disconnect_model
,
511 mock_deploy
.side_effect
= Exception()
512 mock_get_model
.return_value
= juju
.model
.Model()
513 with self
.assertRaises(Exception):
514 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
515 mock_deploy
.assert_called_once()
516 mock_wait_for_model
.assert_not_called()
517 mock_disconnect_controller
.assert_called_once()
518 mock_disconnect_model
.assert_called_once()
521 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
522 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
523 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
524 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
525 @asynctest.mock
.patch(
526 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
528 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
529 @asynctest.mock
.patch("juju.model.Model.deploy")
530 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
531 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
532 class DeployCharmTest(LibjujuTestCase
):
534 super(DeployCharmTest
, self
).setUp()
536 def test_existing_app(
543 mock_disconnect_controller
,
544 mock_disconnect_model
,
548 mock_get_model
.return_value
= juju
.model
.Model()
549 mock_applications
.return_value
= {"existing_app"}
552 with self
.assertRaises(JujuApplicationExists
):
553 application
= self
.loop
.run_until_complete(
554 self
.libjuju
.deploy_charm(
561 self
.assertIsNone(application
)
563 mock_disconnect_controller
.assert_called()
564 mock_disconnect_model
.assert_called()
566 def test_non_existing_machine(
573 mock_disconnect_controller
,
574 mock_disconnect_model
,
578 mock_get_model
.return_value
= juju
.model
.Model()
579 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
581 with self
.assertRaises(JujuMachineNotFound
):
582 application
= self
.loop
.run_until_complete(
583 self
.libjuju
.deploy_charm(
591 self
.assertIsNone(application
)
593 mock_disconnect_controller
.assert_called()
594 mock_disconnect_model
.assert_called()
603 mock_disconnect_controller
,
604 mock_disconnect_model
,
608 mock_get_model
.return_value
= juju
.model
.Model()
609 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
610 mock_create_machine
.return_value
= (FakeMachine(), "other")
611 mock_deploy
.return_value
= FakeApplication()
612 application
= self
.loop
.run_until_complete(
613 self
.libjuju
.deploy_charm(
622 self
.assertIsInstance(application
, FakeApplication
)
624 mock_deploy
.assert_called_once()
625 mock_wait_for
.assert_called_once()
627 mock_create_machine
.assert_called_once()
629 mock_disconnect_controller
.assert_called()
630 mock_disconnect_model
.assert_called()
639 mock_disconnect_controller
,
640 mock_disconnect_model
,
644 mock_get_model
.return_value
= juju
.model
.Model()
645 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
646 mock_deploy
.return_value
= FakeApplication()
647 application
= self
.loop
.run_until_complete(
648 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
651 self
.assertIsInstance(application
, FakeApplication
)
653 mock_deploy
.assert_called_once()
654 mock_wait_for
.assert_called_once()
656 mock_disconnect_controller
.assert_called()
657 mock_disconnect_model
.assert_called()
660 @asynctest.mock
.patch(
661 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
663 class GetApplicationTest(LibjujuTestCase
):
665 super(GetApplicationTest
, self
).setUp()
667 def test_existing_application(
671 mock_applications
.return_value
= {"existing_app": "exists"}
672 model
= juju
.model
.Model()
673 result
= self
.libjuju
._get
_application
(model
, "existing_app")
674 self
.assertEqual(result
, "exists")
676 def test_non_existing_application(
680 mock_applications
.return_value
= {"existing_app": "exists"}
681 model
= juju
.model
.Model()
682 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
683 self
.assertIsNone(result
)
686 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
687 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
688 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
689 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
690 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
691 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
692 @asynctest.mock
.patch("juju.model.Model.get_action_output")
693 @asynctest.mock
.patch("juju.model.Model.get_action_status")
694 class ExecuteActionTest(LibjujuTestCase
):
696 super(ExecuteActionTest
, self
).setUp()
698 def test_no_application(
700 mock_get_action_status
,
701 mock_get_action_output
,
703 mock__get_application
,
704 mock_disconnect_controller
,
705 mock_disconnect_model
,
709 mock__get_application
.return_value
= None
710 mock_get_model
.return_value
= juju
.model
.Model()
713 with self
.assertRaises(JujuApplicationNotFound
):
714 output
, status
= self
.loop
.run_until_complete(
715 self
.libjuju
.execute_action(
721 self
.assertIsNone(output
)
722 self
.assertIsNone(status
)
724 mock_disconnect_controller
.assert_called()
725 mock_disconnect_model
.assert_called()
729 mock_get_action_status
,
730 mock_get_action_output
,
732 mock__get_application
,
733 mock_disconnect_controller
,
734 mock_disconnect_model
,
739 mock_get_model
.return_value
= juju
.model
.Model()
740 mock__get_application
.return_value
= FakeApplication()
743 with self
.assertRaises(JujuActionNotFound
):
744 output
, status
= self
.loop
.run_until_complete(
745 self
.libjuju
.execute_action(
751 self
.assertIsNone(output
)
752 self
.assertIsNone(status
)
754 mock_disconnect_controller
.assert_called()
755 mock_disconnect_model
.assert_called()
757 @asynctest.mock
.patch("asyncio.sleep")
758 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
761 mock_is_leader_from_status
,
763 mock_get_action_status
,
764 mock_get_action_output
,
766 mock__get_application
,
767 mock_disconnect_controller
,
768 mock_disconnect_model
,
772 mock_get_model
.return_value
= juju
.model
.Model()
773 mock__get_application
.return_value
= FakeApplication()
774 mock_is_leader_from_status
.return_value
= False
777 with self
.assertRaises(JujuLeaderUnitNotFound
):
778 output
, status
= self
.loop
.run_until_complete(
779 self
.libjuju
.execute_action(
785 self
.assertIsNone(output
)
786 self
.assertIsNone(status
)
788 mock_disconnect_controller
.assert_called()
789 mock_disconnect_model
.assert_called()
791 def test_succesful_exec(
793 mock_get_action_status
,
794 mock_get_action_output
,
796 mock__get_application
,
797 mock_disconnect_controller
,
798 mock_disconnect_model
,
802 mock_get_model
.return_value
= juju
.model
.Model()
803 mock__get_application
.return_value
= FakeApplication()
804 mock_get_action_output
.return_value
= "output"
805 mock_get_action_status
.return_value
= {"id": "status"}
806 output
, status
= self
.loop
.run_until_complete(
807 self
.libjuju
.execute_action("app", "model", "existing_action")
809 self
.assertEqual(output
, "output")
810 self
.assertEqual(status
, "status")
812 mock_wait_for
.assert_called_once()
814 mock_disconnect_controller
.assert_called()
815 mock_disconnect_model
.assert_called()
818 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
819 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
820 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
821 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
822 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
823 class GetActionTest(LibjujuTestCase
):
825 super(GetActionTest
, self
).setUp()
829 mock_get_application
,
830 mock_disconnect_controller
,
831 mock_disconnect_model
,
835 mock_get_application
.side_effect
= Exception()
837 with self
.assertRaises(Exception):
838 actions
= self
.loop
.run_until_complete(
839 self
.libjuju
.get_actions("app", "model")
842 self
.assertIsNone(actions
)
843 mock_disconnect_controller
.assert_called_once()
844 mock_disconnect_model
.assert_called_once()
848 mock_get_application
,
849 mock_disconnect_controller
,
850 mock_disconnect_model
,
854 mock_get_application
.return_value
= FakeApplication()
856 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
858 self
.assertEqual(actions
, ["existing_action"])
860 mock_get_controller
.assert_called_once()
861 mock_get_model
.assert_called_once()
862 mock_disconnect_controller
.assert_called_once()
863 mock_disconnect_model
.assert_called_once()
866 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
867 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
868 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
869 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
870 @asynctest.mock
.patch("juju.application.Application.get_metrics")
871 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
872 class GetMetricsTest(LibjujuTestCase
):
874 super(GetMetricsTest
, self
).setUp()
876 def test_get_metrics_success(
878 mock_get_application
,
880 mock_disconnect_controller
,
881 mock_disconnect_model
,
885 mock_get_application
.return_value
= FakeApplication()
886 mock_get_model
.return_value
= juju
.model
.Model()
888 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
890 mock_disconnect_controller
.assert_called_once()
891 mock_disconnect_model
.assert_called_once()
893 def test_get_metrics_exception(
895 mock_get_application
,
897 mock_disconnect_controller
,
898 mock_disconnect_model
,
902 mock_get_model
.return_value
= juju
.model
.Model()
903 mock_get_metrics
.side_effect
= Exception()
904 with self
.assertRaises(Exception):
905 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
907 mock_disconnect_controller
.assert_called_once()
908 mock_disconnect_model
.assert_called_once()
910 def test_missing_args_exception(
912 mock_get_application
,
914 mock_disconnect_controller
,
915 mock_disconnect_model
,
919 mock_get_model
.return_value
= juju
.model
.Model()
921 with self
.assertRaises(Exception):
922 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
924 mock_get_controller
.assert_not_called()
925 mock_get_model
.assert_not_called()
926 mock_disconnect_controller
.assert_not_called()
927 mock_disconnect_model
.assert_not_called()
930 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
931 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
932 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
933 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
934 @asynctest.mock
.patch("juju.model.Model.add_relation")
935 class AddRelationTest(LibjujuTestCase
):
937 super(AddRelationTest
, self
).setUp()
939 @asynctest.mock
.patch("logging.Logger.warning")
944 mock_disconnect_controller
,
945 mock_disconnect_model
,
949 # TODO in libjuju.py should this fail only with a log message?
950 result
= {"error": "not found", "response": "response", "request-id": 1}
952 mock_get_model
.return_value
= juju
.model
.Model()
953 mock_add_relation
.side_effect
= JujuAPIError(result
)
955 self
.loop
.run_until_complete(
956 self
.libjuju
.add_relation(
963 mock_warning
.assert_called_with("Relation not found: not found")
964 mock_disconnect_controller
.assert_called_once()
965 mock_disconnect_model
.assert_called_once()
967 @asynctest.mock
.patch("logging.Logger.warning")
968 def test_already_exists(
972 mock_disconnect_controller
,
973 mock_disconnect_model
,
977 # TODO in libjuju.py should this fail silently?
978 result
= {"error": "already exists", "response": "response", "request-id": 1}
980 mock_get_model
.return_value
= juju
.model
.Model()
981 mock_add_relation
.side_effect
= JujuAPIError(result
)
983 self
.loop
.run_until_complete(
984 self
.libjuju
.add_relation(
991 mock_warning
.assert_called_with("Relation already exists: already exists")
992 mock_disconnect_controller
.assert_called_once()
993 mock_disconnect_model
.assert_called_once()
998 mock_disconnect_controller
,
999 mock_disconnect_model
,
1001 mock_get_controller
,
1003 mock_get_model
.return_value
= juju
.model
.Model()
1004 result
= {"error": "", "response": "response", "request-id": 1}
1005 mock_add_relation
.side_effect
= JujuAPIError(result
)
1007 with self
.assertRaises(JujuAPIError
):
1008 self
.loop
.run_until_complete(
1009 self
.libjuju
.add_relation(
1016 mock_disconnect_controller
.assert_called_once()
1017 mock_disconnect_model
.assert_called_once()
1022 mock_disconnect_controller
,
1023 mock_disconnect_model
,
1025 mock_get_controller
,
1027 mock_get_model
.return_value
= juju
.model
.Model()
1029 self
.loop
.run_until_complete(
1030 self
.libjuju
.add_relation(
1037 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1038 mock_disconnect_controller
.assert_called_once()
1039 mock_disconnect_model
.assert_called_once()
1044 mock_disconnect_controller
,
1045 mock_disconnect_model
,
1047 mock_get_controller
,
1049 mock_get_model
.return_value
= juju
.model
.Model()
1051 self
.loop
.run_until_complete(
1052 self
.libjuju
.add_relation(
1059 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1060 mock_disconnect_controller
.assert_called_once()
1061 mock_disconnect_model
.assert_called_once()
1064 # TODO destroy_model testcase
1067 # @asynctest.mock.patch("juju.model.Model.get_machines")
1068 # @asynctest.mock.patch("logging.Logger.debug")
1069 # class DestroyMachineTest(LibjujuTestCase):
1071 # super(DestroyMachineTest, self).setUp()
1073 # def test_success_manual_machine(
1074 # self, mock_debug, mock_get_machines,
1076 # mock_get_machines.side_effect = [
1077 # {"machine": FakeManualMachine()},
1078 # {"machine": FakeManualMachine()},
1081 # self.loop.run_until_complete(
1082 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1085 # asynctest.call("Waiting for machine machine is destroyed"),
1086 # asynctest.call("Machine destroyed: machine"),
1088 # mock_debug.assert_has_calls(calls)
1090 # def test_no_machine(
1091 # self, mock_debug, mock_get_machines,
1093 # mock_get_machines.return_value = {}
1094 # self.loop.run_until_complete(
1095 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1097 # mock_debug.assert_called_with("Machine not found: machine")
1100 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1101 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1102 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1103 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1104 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1105 class ConfigureApplicationTest(LibjujuTestCase
):
1107 super(ConfigureApplicationTest
, self
).setUp()
1111 mock_get_application
,
1112 mock_disconnect_controller
,
1113 mock_disconnect_model
,
1115 mock_get_controller
,
1118 mock_get_application
.return_value
= FakeApplication()
1120 self
.loop
.run_until_complete(
1121 self
.libjuju
.configure_application(
1127 mock_get_application
.assert_called_once()
1128 mock_disconnect_controller
.assert_called_once()
1129 mock_disconnect_model
.assert_called_once()
1133 mock_get_application
,
1134 mock_disconnect_controller
,
1135 mock_disconnect_model
,
1137 mock_get_controller
,
1140 mock_get_application
.side_effect
= Exception()
1142 with self
.assertRaises(Exception):
1143 self
.loop
.run_until_complete(
1144 self
.libjuju
.configure_application(
1150 mock_disconnect_controller
.assert_called_once()
1151 mock_disconnect_model
.assert_called_once()
1153 def test_controller_exception(
1155 mock_get_application
,
1156 mock_disconnect_controller
,
1157 mock_disconnect_model
,
1159 mock_get_controller
,
1162 result
= {"error": "not found", "response": "response", "request-id": 1}
1164 mock_get_controller
.side_effect
= JujuAPIError(result
)
1166 with self
.assertRaises(JujuAPIError
):
1167 self
.loop
.run_until_complete(
1168 self
.libjuju
.configure_application(
1174 mock_get_model
.assert_not_called()
1175 mock_disconnect_controller
.assert_not_called()
1176 mock_disconnect_model
.assert_not_called()
1178 def test_get_model_exception(
1180 mock_get_application
,
1181 mock_disconnect_controller
,
1182 mock_disconnect_model
,
1184 mock_get_controller
,
1187 result
= {"error": "not found", "response": "response", "request-id": 1}
1188 mock_get_model
.side_effect
= JujuAPIError(result
)
1190 with self
.assertRaises(JujuAPIError
):
1191 self
.loop
.run_until_complete(
1192 self
.libjuju
.configure_application(
1198 mock_get_model
.assert_called_once()
1199 mock_disconnect_controller
.assert_called_once()
1200 mock_disconnect_model
.assert_not_called()
1203 # TODO _get_api_endpoints_db test case
1204 # TODO _update_api_endpoints_db test case
1205 # TODO healthcheck test case
1208 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1209 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1210 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1211 class ListModelsTest(LibjujuTestCase
):
1213 super(ListModelsTest
, self
).setUp()
1215 def test_containing(
1218 mock_disconnect_controller
,
1219 mock_get_controller
,
1221 mock_get_controller
.return_value
= juju
.controller
.Controller()
1222 mock_list_models
.return_value
= ["existingmodel"]
1223 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1225 mock_disconnect_controller
.assert_called_once()
1226 self
.assertEquals(models
, ["existingmodel"])
1228 def test_not_containing(
1231 mock_disconnect_controller
,
1232 mock_get_controller
,
1234 mock_get_controller
.return_value
= juju
.controller
.Controller()
1235 mock_list_models
.return_value
= ["existingmodel", "model"]
1236 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1238 mock_disconnect_controller
.assert_called_once()
1239 self
.assertEquals(models
, [])
1241 def test_no_contains_arg(
1244 mock_disconnect_controller
,
1245 mock_get_controller
,
1247 mock_get_controller
.return_value
= juju
.controller
.Controller()
1248 mock_list_models
.return_value
= ["existingmodel", "model"]
1249 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1251 mock_disconnect_controller
.assert_called_once()
1252 self
.assertEquals(models
, ["existingmodel", "model"])
1255 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1256 class ModelsExistTest(LibjujuTestCase
):
1258 super(ModelsExistTest
, self
).setUp()
1260 def test_model_names_none(self
, mock_list_models
):
1261 mock_list_models
.return_value
= []
1262 with self
.assertRaises(Exception):
1263 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1265 def test_model_names_empty(self
, mock_list_models
):
1266 mock_list_models
.return_value
= []
1267 with self
.assertRaises(Exception):
1268 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1269 self
.libjuju
.models_exist([])
1272 def test_model_names_not_existing(self
, mock_list_models
):
1273 mock_list_models
.return_value
= ["prometheus", "grafana"]
1274 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1275 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1277 self
.assertFalse(exist
)
1278 self
.assertEqual(non_existing_models
, ["prometheus2"])
1280 def test_model_names_exist(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(["prometheus", "grafana"])
1285 self
.assertTrue(exist
)
1286 self
.assertEqual(non_existing_models
, [])
1289 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1290 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1291 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1292 class ListOffers(LibjujuTestCase
):
1294 super(ListOffers
, self
).setUp()
1296 def test_disconnect_controller(
1299 mock_disconnect_controller
,
1300 mock_get_controller
,
1302 mock_get_controller
.return_value
= juju
.controller
.Controller()
1303 mock_list_offers
.side_effect
= Exception()
1304 with self
.assertRaises(Exception):
1305 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1306 mock_disconnect_controller
.assert_called_once()
1308 def test_empty_list(
1311 mock_disconnect_controller
,
1312 mock_get_controller
,
1314 mock_get_controller
.return_value
= juju
.controller
.Controller()
1315 mock_list_offers
.return_value
= []
1316 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1317 self
.assertEqual(offers
, [])
1318 mock_disconnect_controller
.assert_called_once()
1320 def test_non_empty_list(
1323 mock_disconnect_controller
,
1324 mock_get_controller
,
1326 mock_get_controller
.return_value
= juju
.controller
.Controller()
1327 mock_list_offers
.return_value
= ["offer"]
1328 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1329 self
.assertEqual(offers
, ["offer"])
1330 mock_disconnect_controller
.assert_called_once()
1333 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1334 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1335 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1336 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1337 @asynctest.mock
.patch("juju.model.Model.consume")
1338 class ConsumeTest(LibjujuTestCase
):
1340 super(ConsumeTest
, self
).setUp()
1345 mock_disconnect_controller
,
1346 mock_disconnect_model
,
1348 mock_get_controller
,
1350 mock_get_controller
.return_value
= juju
.controller
.Controller()
1351 mock_get_model
.return_value
= juju
.model
.Model()
1353 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1354 mock_consume
.assert_called_once()
1355 mock_disconnect_model
.assert_called_once()
1356 mock_disconnect_controller
.assert_called_once()
1358 def test_parsing_error_exception(
1361 mock_disconnect_controller
,
1362 mock_disconnect_model
,
1364 mock_get_controller
,
1366 mock_get_controller
.return_value
= juju
.controller
.Controller()
1367 mock_get_model
.return_value
= juju
.model
.Model()
1368 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1370 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1371 self
.loop
.run_until_complete(
1372 self
.libjuju
.consume("offer_url", "model_name")
1374 mock_consume
.assert_called_once()
1375 mock_disconnect_model
.assert_called_once()
1376 mock_disconnect_controller
.assert_called_once()
1378 def test_juju_error_exception(
1381 mock_disconnect_controller
,
1382 mock_disconnect_model
,
1384 mock_get_controller
,
1386 mock_get_controller
.return_value
= juju
.controller
.Controller()
1387 mock_get_model
.return_value
= juju
.model
.Model()
1388 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1390 with self
.assertRaises(juju
.errors
.JujuError
):
1391 self
.loop
.run_until_complete(
1392 self
.libjuju
.consume("offer_url", "model_name")
1394 mock_consume
.assert_called_once()
1395 mock_disconnect_model
.assert_called_once()
1396 mock_disconnect_controller
.assert_called_once()
1398 def test_juju_api_error_exception(
1401 mock_disconnect_controller
,
1402 mock_disconnect_model
,
1404 mock_get_controller
,
1406 mock_get_controller
.return_value
= juju
.controller
.Controller()
1407 mock_get_model
.return_value
= juju
.model
.Model()
1408 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1409 {"error": "", "response": "", "request-id": ""}
1412 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1413 self
.loop
.run_until_complete(
1414 self
.libjuju
.consume("offer_url", "model_name")
1416 mock_consume
.assert_called_once()
1417 mock_disconnect_model
.assert_called_once()
1418 mock_disconnect_controller
.assert_called_once()
1421 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1422 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1423 class AddK8sTest(LibjujuTestCase
):
1425 super(AddK8sTest
, self
).setUp()
1427 rbac_id
= generate_rbac_id()
1429 client_cert_data
= "cert"
1430 configuration
= kubernetes
.client
.configuration
.Configuration()
1431 storage_class
= "storage_class"
1432 credential_name
= name
1434 self
._add
_k
8s
_args
= {
1438 "client_cert_data": client_cert_data
,
1439 "configuration": configuration
,
1440 "storage_class": storage_class
,
1441 "credential_name": credential_name
,
1444 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1445 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1446 mock_add_cloud
.assert_called_once()
1447 mock_get_k8s_cloud_credential
.assert_called_once()
1449 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1450 mock_add_cloud
.side_effect
= Exception()
1451 with self
.assertRaises(Exception):
1452 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1453 mock_add_cloud
.assert_called_once()
1454 mock_get_k8s_cloud_credential
.assert_called_once()
1456 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1457 self
._add
_k
8s
_args
["name"] = ""
1458 with self
.assertRaises(Exception):
1459 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1460 mock_add_cloud
.assert_not_called()
1462 def test_add_k8s_missing_storage_name(
1463 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1465 self
._add
_k
8s
_args
["storage_class"] = ""
1466 with self
.assertRaises(Exception):
1467 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1468 mock_add_cloud
.assert_not_called()
1470 def test_add_k8s_missing_configuration_keys(
1471 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1473 self
._add
_k
8s
_args
["configuration"] = None
1474 with self
.assertRaises(Exception):
1475 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1476 mock_add_cloud
.assert_not_called()
1479 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1480 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1481 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1482 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1483 class AddCloudTest(LibjujuTestCase
):
1485 super(AddCloudTest
, self
).setUp()
1486 self
.cloud
= juju
.client
.client
.Cloud()
1487 self
.credential
= juju
.client
.client
.CloudCredential()
1489 def test_add_cloud_with_credential(
1491 mock_add_credential
,
1493 mock_disconnect_controller
,
1494 mock_get_controller
,
1496 mock_get_controller
.return_value
= juju
.controller
.Controller()
1498 cloud
= self
.loop
.run_until_complete(
1499 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1501 self
.assertEqual(cloud
, self
.cloud
)
1502 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1503 mock_add_credential
.assert_called_once_with(
1504 "cloud", credential
=self
.credential
, cloud
="cloud"
1506 mock_disconnect_controller
.assert_called_once()
1508 def test_add_cloud_no_credential(
1510 mock_add_credential
,
1512 mock_disconnect_controller
,
1513 mock_get_controller
,
1515 mock_get_controller
.return_value
= juju
.controller
.Controller()
1517 cloud
= self
.loop
.run_until_complete(
1518 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1520 self
.assertEqual(cloud
, self
.cloud
)
1521 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1522 mock_add_credential
.assert_not_called()
1523 mock_disconnect_controller
.assert_called_once()
1525 def test_add_cloud_exception(
1527 mock_add_credential
,
1529 mock_disconnect_controller
,
1530 mock_get_controller
,
1532 mock_get_controller
.return_value
= juju
.controller
.Controller()
1533 mock_add_cloud
.side_effect
= Exception()
1534 with self
.assertRaises(Exception):
1535 self
.loop
.run_until_complete(
1536 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1539 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1540 mock_add_credential
.assert_not_called()
1541 mock_disconnect_controller
.assert_called_once()
1543 def test_add_credential_exception(
1545 mock_add_credential
,
1547 mock_disconnect_controller
,
1548 mock_get_controller
,
1550 mock_get_controller
.return_value
= juju
.controller
.Controller()
1551 mock_add_credential
.side_effect
= Exception()
1552 with self
.assertRaises(Exception):
1553 self
.loop
.run_until_complete(
1554 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1557 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1558 mock_add_credential
.assert_called_once_with(
1559 "cloud", credential
=self
.credential
, cloud
="cloud"
1561 mock_disconnect_controller
.assert_called_once()
1564 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1565 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1566 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1567 class RemoveCloudTest(LibjujuTestCase
):
1569 super(RemoveCloudTest
, self
).setUp()
1571 def test_remove_cloud(
1574 mock_disconnect_controller
,
1575 mock_get_controller
,
1577 mock_get_controller
.return_value
= juju
.controller
.Controller()
1579 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1580 mock_remove_cloud
.assert_called_once_with("cloud")
1581 mock_disconnect_controller
.assert_called_once()
1583 def test_remove_cloud_exception(
1586 mock_disconnect_controller
,
1587 mock_get_controller
,
1589 mock_get_controller
.return_value
= juju
.controller
.Controller()
1590 mock_remove_cloud
.side_effect
= Exception()
1592 with self
.assertRaises(Exception):
1593 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1594 mock_remove_cloud
.assert_called_once_with("cloud")
1595 mock_disconnect_controller
.assert_called_once()
1598 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1599 class GetK8sCloudCredentials(LibjujuTestCase
):
1601 super(GetK8sCloudCredentials
, self
).setUp()
1602 self
.cert_data
= "cert"
1603 self
.token
= "token"
1605 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1606 def test_not_supported(self
, mock_exception
, mock_configuration
):
1607 mock_configuration
.username
= ""
1608 mock_configuration
.password
= ""
1609 mock_configuration
.ssl_ca_cert
= None
1610 mock_configuration
.cert_file
= None
1611 mock_configuration
.key_file
= None
1612 exception_raised
= False
1614 self
.cert_data
= None
1616 _
= self
.libjuju
.get_k8s_cloud_credential(
1621 except JujuInvalidK8sConfiguration
as e
:
1622 exception_raised
= True
1625 "authentication method not supported",
1627 self
.assertTrue(exception_raised
)
1629 def test_user_pass(self
, mock_configuration
):
1630 mock_configuration
.username
= "admin"
1631 mock_configuration
.password
= "admin"
1632 mock_configuration
.ssl_ca_cert
= None
1633 mock_configuration
.cert_file
= None
1634 mock_configuration
.key_file
= None
1636 self
.cert_data
= None
1637 credential
= self
.libjuju
.get_k8s_cloud_credential(
1644 juju
.client
._definitions
.CloudCredential(
1645 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1649 def test_user_pass_with_cert(self
, mock_configuration
):
1650 mock_configuration
.username
= "admin"
1651 mock_configuration
.password
= "admin"
1652 mock_configuration
.ssl_ca_cert
= None
1653 mock_configuration
.cert_file
= None
1654 mock_configuration
.key_file
= None
1656 credential
= self
.libjuju
.get_k8s_cloud_credential(
1663 juju
.client
._definitions
.CloudCredential(
1665 "ClientCertificateData": self
.cert_data
,
1666 "username": "admin",
1667 "password": "admin",
1669 auth_type
="userpasswithcert",
1673 def test_user_no_pass(self
, mock_configuration
):
1674 mock_configuration
.username
= "admin"
1675 mock_configuration
.password
= ""
1676 mock_configuration
.ssl_ca_cert
= None
1677 mock_configuration
.cert_file
= None
1678 mock_configuration
.key_file
= None
1680 self
.cert_data
= None
1681 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1682 credential
= self
.libjuju
.get_k8s_cloud_credential(
1689 juju
.client
._definitions
.CloudCredential(
1690 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1693 mock_debug
.assert_called_once_with(
1694 "credential for user admin has empty password"
1697 def test_cert(self
, mock_configuration
):
1698 mock_configuration
.username
= ""
1699 mock_configuration
.password
= ""
1700 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1701 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1702 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1703 ssl_ca_cert_file
.write("cacert")
1704 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1705 mock_configuration
.cert_file
= None
1706 mock_configuration
.key_file
= None
1707 credential
= self
.libjuju
.get_k8s_cloud_credential(
1714 juju
.client
._definitions
.CloudCredential(
1715 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1716 auth_type
="certificate",
1720 # TODO: Fix this test when oauth authentication is supported
1721 # def test_oauth2(self, mock_configuration):
1722 # mock_configuration.username = ""
1723 # mock_configuration.password = ""
1724 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1725 # key = tempfile.NamedTemporaryFile()
1726 # with open(key.name, "w") as key_file:
1727 # key_file.write("key")
1728 # mock_configuration.ssl_ca_cert = None
1729 # mock_configuration.cert_file = None
1730 # mock_configuration.key_file = key.name
1731 # credential = self.libjuju.get_k8s_cloud_credential(
1732 # mock_configuration,
1738 # juju.client._definitions.CloudCredential(
1739 # attrs={"ClientKeyData": "key", "Token": "Token"},
1740 # auth_type="oauth2",
1744 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1745 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1746 # mock_configuration.username = ""
1747 # mock_configuration.password = ""
1748 # key = tempfile.NamedTemporaryFile()
1749 # with open(key.name, "w") as key_file:
1750 # key_file.write("key")
1751 # mock_configuration.ssl_ca_cert = None
1752 # mock_configuration.cert_file = None
1753 # mock_configuration.key_file = key.name
1754 # exception_raised = False
1756 # _ = self.libjuju.get_k8s_cloud_credential(
1757 # mock_configuration,
1761 # except JujuInvalidK8sConfiguration as e:
1762 # exception_raised = True
1765 # "missing token for auth type oauth2",
1767 # self.assertTrue(exception_raised)
1769 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1770 mock_configuration
.username
= "admin"
1771 mock_configuration
.password
= "pass"
1772 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1773 mock_configuration
.ssl_ca_cert
= None
1774 mock_configuration
.cert_file
= None
1775 mock_configuration
.key_file
= None
1776 exception_raised
= False
1778 _
= self
.libjuju
.get_k8s_cloud_credential(
1783 except JujuInvalidK8sConfiguration
as e
:
1784 exception_raised
= True
1787 "Cannot set both token and user/pass",
1789 self
.assertTrue(exception_raised
)