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("n2vc.libjuju.Libjuju.get_controller")
1068 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1069 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1070 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1071 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1072 class DestroyApplicationTest(LibjujuTestCase
):
1074 super(DestroyApplicationTest
, self
).setUp()
1078 mock_get_controller
,
1080 mock_disconnect_controller
,
1081 mock_get_application
,
1082 mock_disconnect_model
,
1084 mock_get_application
.return_value
= FakeApplication()
1085 mock_get_model
.return_value
= None
1086 self
.loop
.run_until_complete(
1087 self
.libjuju
.destroy_application(
1093 mock_get_application
.assert_called()
1094 mock_disconnect_controller
.assert_called_once()
1095 mock_disconnect_model
.assert_called_once()
1097 def test_no_application(
1099 mock_get_controller
,
1101 mock_disconnect_controller
,
1102 mock_get_application
,
1103 mock_disconnect_model
,
1105 mock_get_model
.return_value
= None
1106 mock_get_application
.return_value
= None
1108 self
.loop
.run_until_complete(
1109 self
.libjuju
.destroy_application(
1115 mock_get_application
.assert_called()
1119 mock_get_controller
,
1121 mock_disconnect_controller
,
1122 mock_get_application
,
1123 mock_disconnect_model
,
1125 mock_get_application
.return_value
= FakeApplication
1126 mock_get_model
.return_value
= None
1128 with self
.assertRaises(Exception):
1129 self
.loop
.run_until_complete(
1130 self
.libjuju
.destroy_application(
1136 mock_get_application
.assert_called_once()
1139 # @asynctest.mock.patch("juju.model.Model.get_machines")
1140 # @asynctest.mock.patch("logging.Logger.debug")
1141 # class DestroyMachineTest(LibjujuTestCase):
1143 # super(DestroyMachineTest, self).setUp()
1145 # def test_success_manual_machine(
1146 # self, mock_debug, mock_get_machines,
1148 # mock_get_machines.side_effect = [
1149 # {"machine": FakeManualMachine()},
1150 # {"machine": FakeManualMachine()},
1153 # self.loop.run_until_complete(
1154 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1157 # asynctest.call("Waiting for machine machine is destroyed"),
1158 # asynctest.call("Machine destroyed: machine"),
1160 # mock_debug.assert_has_calls(calls)
1162 # def test_no_machine(
1163 # self, mock_debug, mock_get_machines,
1165 # mock_get_machines.return_value = {}
1166 # self.loop.run_until_complete(
1167 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1169 # mock_debug.assert_called_with("Machine not found: machine")
1172 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1173 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1174 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1175 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1176 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1177 class ConfigureApplicationTest(LibjujuTestCase
):
1179 super(ConfigureApplicationTest
, self
).setUp()
1183 mock_get_application
,
1184 mock_disconnect_controller
,
1185 mock_disconnect_model
,
1187 mock_get_controller
,
1190 mock_get_application
.return_value
= FakeApplication()
1192 self
.loop
.run_until_complete(
1193 self
.libjuju
.configure_application(
1199 mock_get_application
.assert_called_once()
1200 mock_disconnect_controller
.assert_called_once()
1201 mock_disconnect_model
.assert_called_once()
1205 mock_get_application
,
1206 mock_disconnect_controller
,
1207 mock_disconnect_model
,
1209 mock_get_controller
,
1212 mock_get_application
.side_effect
= Exception()
1214 with self
.assertRaises(Exception):
1215 self
.loop
.run_until_complete(
1216 self
.libjuju
.configure_application(
1222 mock_disconnect_controller
.assert_called_once()
1223 mock_disconnect_model
.assert_called_once()
1225 def test_controller_exception(
1227 mock_get_application
,
1228 mock_disconnect_controller
,
1229 mock_disconnect_model
,
1231 mock_get_controller
,
1234 result
= {"error": "not found", "response": "response", "request-id": 1}
1236 mock_get_controller
.side_effect
= JujuAPIError(result
)
1238 with self
.assertRaises(JujuAPIError
):
1239 self
.loop
.run_until_complete(
1240 self
.libjuju
.configure_application(
1246 mock_get_model
.assert_not_called()
1247 mock_disconnect_controller
.assert_not_called()
1248 mock_disconnect_model
.assert_not_called()
1250 def test_get_model_exception(
1252 mock_get_application
,
1253 mock_disconnect_controller
,
1254 mock_disconnect_model
,
1256 mock_get_controller
,
1259 result
= {"error": "not found", "response": "response", "request-id": 1}
1260 mock_get_model
.side_effect
= JujuAPIError(result
)
1262 with self
.assertRaises(JujuAPIError
):
1263 self
.loop
.run_until_complete(
1264 self
.libjuju
.configure_application(
1270 mock_get_model
.assert_called_once()
1271 mock_disconnect_controller
.assert_called_once()
1272 mock_disconnect_model
.assert_not_called()
1275 # TODO _get_api_endpoints_db test case
1276 # TODO _update_api_endpoints_db test case
1277 # TODO healthcheck test case
1280 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1281 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1282 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1283 class ListModelsTest(LibjujuTestCase
):
1285 super(ListModelsTest
, self
).setUp()
1287 def test_containing(
1290 mock_disconnect_controller
,
1291 mock_get_controller
,
1293 mock_get_controller
.return_value
= juju
.controller
.Controller()
1294 mock_list_models
.return_value
= ["existingmodel"]
1295 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1297 mock_disconnect_controller
.assert_called_once()
1298 self
.assertEquals(models
, ["existingmodel"])
1300 def test_not_containing(
1303 mock_disconnect_controller
,
1304 mock_get_controller
,
1306 mock_get_controller
.return_value
= juju
.controller
.Controller()
1307 mock_list_models
.return_value
= ["existingmodel", "model"]
1308 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1310 mock_disconnect_controller
.assert_called_once()
1311 self
.assertEquals(models
, [])
1313 def test_no_contains_arg(
1316 mock_disconnect_controller
,
1317 mock_get_controller
,
1319 mock_get_controller
.return_value
= juju
.controller
.Controller()
1320 mock_list_models
.return_value
= ["existingmodel", "model"]
1321 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1323 mock_disconnect_controller
.assert_called_once()
1324 self
.assertEquals(models
, ["existingmodel", "model"])
1327 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1328 class ModelsExistTest(LibjujuTestCase
):
1330 super(ModelsExistTest
, self
).setUp()
1332 def test_model_names_none(self
, mock_list_models
):
1333 mock_list_models
.return_value
= []
1334 with self
.assertRaises(Exception):
1335 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1337 def test_model_names_empty(self
, mock_list_models
):
1338 mock_list_models
.return_value
= []
1339 with self
.assertRaises(Exception):
1340 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1341 self
.libjuju
.models_exist([])
1344 def test_model_names_not_existing(self
, mock_list_models
):
1345 mock_list_models
.return_value
= ["prometheus", "grafana"]
1346 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1347 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1349 self
.assertFalse(exist
)
1350 self
.assertEqual(non_existing_models
, ["prometheus2"])
1352 def test_model_names_exist(self
, mock_list_models
):
1353 mock_list_models
.return_value
= ["prometheus", "grafana"]
1354 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1355 self
.libjuju
.models_exist(["prometheus", "grafana"])
1357 self
.assertTrue(exist
)
1358 self
.assertEqual(non_existing_models
, [])
1361 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1362 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1363 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1364 class ListOffers(LibjujuTestCase
):
1366 super(ListOffers
, self
).setUp()
1368 def test_disconnect_controller(
1371 mock_disconnect_controller
,
1372 mock_get_controller
,
1374 mock_get_controller
.return_value
= juju
.controller
.Controller()
1375 mock_list_offers
.side_effect
= Exception()
1376 with self
.assertRaises(Exception):
1377 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1378 mock_disconnect_controller
.assert_called_once()
1380 def test_empty_list(
1383 mock_disconnect_controller
,
1384 mock_get_controller
,
1386 mock_get_controller
.return_value
= juju
.controller
.Controller()
1387 mock_list_offers
.return_value
= []
1388 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1389 self
.assertEqual(offers
, [])
1390 mock_disconnect_controller
.assert_called_once()
1392 def test_non_empty_list(
1395 mock_disconnect_controller
,
1396 mock_get_controller
,
1398 mock_get_controller
.return_value
= juju
.controller
.Controller()
1399 mock_list_offers
.return_value
= ["offer"]
1400 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1401 self
.assertEqual(offers
, ["offer"])
1402 mock_disconnect_controller
.assert_called_once()
1405 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1406 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1407 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1408 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1409 @asynctest.mock
.patch("juju.model.Model.consume")
1410 class ConsumeTest(LibjujuTestCase
):
1412 super(ConsumeTest
, self
).setUp()
1417 mock_disconnect_controller
,
1418 mock_disconnect_model
,
1420 mock_get_controller
,
1422 mock_get_controller
.return_value
= juju
.controller
.Controller()
1423 mock_get_model
.return_value
= juju
.model
.Model()
1425 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1426 mock_consume
.assert_called_once()
1427 mock_disconnect_model
.assert_called_once()
1428 mock_disconnect_controller
.assert_called_once()
1430 def test_parsing_error_exception(
1433 mock_disconnect_controller
,
1434 mock_disconnect_model
,
1436 mock_get_controller
,
1438 mock_get_controller
.return_value
= juju
.controller
.Controller()
1439 mock_get_model
.return_value
= juju
.model
.Model()
1440 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1442 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1443 self
.loop
.run_until_complete(
1444 self
.libjuju
.consume("offer_url", "model_name")
1446 mock_consume
.assert_called_once()
1447 mock_disconnect_model
.assert_called_once()
1448 mock_disconnect_controller
.assert_called_once()
1450 def test_juju_error_exception(
1453 mock_disconnect_controller
,
1454 mock_disconnect_model
,
1456 mock_get_controller
,
1458 mock_get_controller
.return_value
= juju
.controller
.Controller()
1459 mock_get_model
.return_value
= juju
.model
.Model()
1460 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1462 with self
.assertRaises(juju
.errors
.JujuError
):
1463 self
.loop
.run_until_complete(
1464 self
.libjuju
.consume("offer_url", "model_name")
1466 mock_consume
.assert_called_once()
1467 mock_disconnect_model
.assert_called_once()
1468 mock_disconnect_controller
.assert_called_once()
1470 def test_juju_api_error_exception(
1473 mock_disconnect_controller
,
1474 mock_disconnect_model
,
1476 mock_get_controller
,
1478 mock_get_controller
.return_value
= juju
.controller
.Controller()
1479 mock_get_model
.return_value
= juju
.model
.Model()
1480 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1481 {"error": "", "response": "", "request-id": ""}
1484 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1485 self
.loop
.run_until_complete(
1486 self
.libjuju
.consume("offer_url", "model_name")
1488 mock_consume
.assert_called_once()
1489 mock_disconnect_model
.assert_called_once()
1490 mock_disconnect_controller
.assert_called_once()
1493 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1494 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1495 class AddK8sTest(LibjujuTestCase
):
1497 super(AddK8sTest
, self
).setUp()
1499 rbac_id
= generate_rbac_id()
1501 client_cert_data
= "cert"
1502 configuration
= kubernetes
.client
.configuration
.Configuration()
1503 storage_class
= "storage_class"
1504 credential_name
= name
1506 self
._add
_k
8s
_args
= {
1510 "client_cert_data": client_cert_data
,
1511 "configuration": configuration
,
1512 "storage_class": storage_class
,
1513 "credential_name": credential_name
,
1516 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1517 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1518 mock_add_cloud
.assert_called_once()
1519 mock_get_k8s_cloud_credential
.assert_called_once()
1521 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1522 mock_add_cloud
.side_effect
= Exception()
1523 with self
.assertRaises(Exception):
1524 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1525 mock_add_cloud
.assert_called_once()
1526 mock_get_k8s_cloud_credential
.assert_called_once()
1528 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1529 self
._add
_k
8s
_args
["name"] = ""
1530 with self
.assertRaises(Exception):
1531 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1532 mock_add_cloud
.assert_not_called()
1534 def test_add_k8s_missing_storage_name(
1535 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1537 self
._add
_k
8s
_args
["storage_class"] = ""
1538 with self
.assertRaises(Exception):
1539 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1540 mock_add_cloud
.assert_not_called()
1542 def test_add_k8s_missing_configuration_keys(
1543 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1545 self
._add
_k
8s
_args
["configuration"] = None
1546 with self
.assertRaises(Exception):
1547 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1548 mock_add_cloud
.assert_not_called()
1551 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1552 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1553 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1554 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1555 class AddCloudTest(LibjujuTestCase
):
1557 super(AddCloudTest
, self
).setUp()
1558 self
.cloud
= juju
.client
.client
.Cloud()
1559 self
.credential
= juju
.client
.client
.CloudCredential()
1561 def test_add_cloud_with_credential(
1563 mock_add_credential
,
1565 mock_disconnect_controller
,
1566 mock_get_controller
,
1568 mock_get_controller
.return_value
= juju
.controller
.Controller()
1570 cloud
= self
.loop
.run_until_complete(
1571 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1573 self
.assertEqual(cloud
, self
.cloud
)
1574 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1575 mock_add_credential
.assert_called_once_with(
1576 "cloud", credential
=self
.credential
, cloud
="cloud"
1578 mock_disconnect_controller
.assert_called_once()
1580 def test_add_cloud_no_credential(
1582 mock_add_credential
,
1584 mock_disconnect_controller
,
1585 mock_get_controller
,
1587 mock_get_controller
.return_value
= juju
.controller
.Controller()
1589 cloud
= self
.loop
.run_until_complete(
1590 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1592 self
.assertEqual(cloud
, self
.cloud
)
1593 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1594 mock_add_credential
.assert_not_called()
1595 mock_disconnect_controller
.assert_called_once()
1597 def test_add_cloud_exception(
1599 mock_add_credential
,
1601 mock_disconnect_controller
,
1602 mock_get_controller
,
1604 mock_get_controller
.return_value
= juju
.controller
.Controller()
1605 mock_add_cloud
.side_effect
= Exception()
1606 with self
.assertRaises(Exception):
1607 self
.loop
.run_until_complete(
1608 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1611 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1612 mock_add_credential
.assert_not_called()
1613 mock_disconnect_controller
.assert_called_once()
1615 def test_add_credential_exception(
1617 mock_add_credential
,
1619 mock_disconnect_controller
,
1620 mock_get_controller
,
1622 mock_get_controller
.return_value
= juju
.controller
.Controller()
1623 mock_add_credential
.side_effect
= Exception()
1624 with self
.assertRaises(Exception):
1625 self
.loop
.run_until_complete(
1626 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1629 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1630 mock_add_credential
.assert_called_once_with(
1631 "cloud", credential
=self
.credential
, cloud
="cloud"
1633 mock_disconnect_controller
.assert_called_once()
1636 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1637 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1638 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1639 class RemoveCloudTest(LibjujuTestCase
):
1641 super(RemoveCloudTest
, self
).setUp()
1643 def test_remove_cloud(
1646 mock_disconnect_controller
,
1647 mock_get_controller
,
1649 mock_get_controller
.return_value
= juju
.controller
.Controller()
1651 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1652 mock_remove_cloud
.assert_called_once_with("cloud")
1653 mock_disconnect_controller
.assert_called_once()
1655 def test_remove_cloud_exception(
1658 mock_disconnect_controller
,
1659 mock_get_controller
,
1661 mock_get_controller
.return_value
= juju
.controller
.Controller()
1662 mock_remove_cloud
.side_effect
= Exception()
1664 with self
.assertRaises(Exception):
1665 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1666 mock_remove_cloud
.assert_called_once_with("cloud")
1667 mock_disconnect_controller
.assert_called_once()
1670 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1671 class GetK8sCloudCredentials(LibjujuTestCase
):
1673 super(GetK8sCloudCredentials
, self
).setUp()
1674 self
.cert_data
= "cert"
1675 self
.token
= "token"
1677 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1678 def test_not_supported(self
, mock_exception
, mock_configuration
):
1679 mock_configuration
.username
= ""
1680 mock_configuration
.password
= ""
1681 mock_configuration
.ssl_ca_cert
= None
1682 mock_configuration
.cert_file
= None
1683 mock_configuration
.key_file
= None
1684 exception_raised
= False
1686 self
.cert_data
= None
1688 _
= self
.libjuju
.get_k8s_cloud_credential(
1693 except JujuInvalidK8sConfiguration
as e
:
1694 exception_raised
= True
1697 "authentication method not supported",
1699 self
.assertTrue(exception_raised
)
1701 def test_user_pass(self
, mock_configuration
):
1702 mock_configuration
.username
= "admin"
1703 mock_configuration
.password
= "admin"
1704 mock_configuration
.ssl_ca_cert
= None
1705 mock_configuration
.cert_file
= None
1706 mock_configuration
.key_file
= None
1708 self
.cert_data
= None
1709 credential
= self
.libjuju
.get_k8s_cloud_credential(
1716 juju
.client
._definitions
.CloudCredential(
1717 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1721 def test_user_pass_with_cert(self
, mock_configuration
):
1722 mock_configuration
.username
= "admin"
1723 mock_configuration
.password
= "admin"
1724 mock_configuration
.ssl_ca_cert
= None
1725 mock_configuration
.cert_file
= None
1726 mock_configuration
.key_file
= None
1728 credential
= self
.libjuju
.get_k8s_cloud_credential(
1735 juju
.client
._definitions
.CloudCredential(
1737 "ClientCertificateData": self
.cert_data
,
1738 "username": "admin",
1739 "password": "admin",
1741 auth_type
="userpasswithcert",
1745 def test_user_no_pass(self
, mock_configuration
):
1746 mock_configuration
.username
= "admin"
1747 mock_configuration
.password
= ""
1748 mock_configuration
.ssl_ca_cert
= None
1749 mock_configuration
.cert_file
= None
1750 mock_configuration
.key_file
= None
1752 self
.cert_data
= None
1753 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1754 credential
= self
.libjuju
.get_k8s_cloud_credential(
1761 juju
.client
._definitions
.CloudCredential(
1762 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1765 mock_debug
.assert_called_once_with(
1766 "credential for user admin has empty password"
1769 def test_cert(self
, mock_configuration
):
1770 mock_configuration
.username
= ""
1771 mock_configuration
.password
= ""
1772 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1773 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1774 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1775 ssl_ca_cert_file
.write("cacert")
1776 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1777 mock_configuration
.cert_file
= None
1778 mock_configuration
.key_file
= None
1779 credential
= self
.libjuju
.get_k8s_cloud_credential(
1786 juju
.client
._definitions
.CloudCredential(
1787 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1788 auth_type
="certificate",
1792 # TODO: Fix this test when oauth authentication is supported
1793 # def test_oauth2(self, mock_configuration):
1794 # mock_configuration.username = ""
1795 # mock_configuration.password = ""
1796 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1797 # key = tempfile.NamedTemporaryFile()
1798 # with open(key.name, "w") as key_file:
1799 # key_file.write("key")
1800 # mock_configuration.ssl_ca_cert = None
1801 # mock_configuration.cert_file = None
1802 # mock_configuration.key_file = key.name
1803 # credential = self.libjuju.get_k8s_cloud_credential(
1804 # mock_configuration,
1810 # juju.client._definitions.CloudCredential(
1811 # attrs={"ClientKeyData": "key", "Token": "Token"},
1812 # auth_type="oauth2",
1816 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1817 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1818 # mock_configuration.username = ""
1819 # mock_configuration.password = ""
1820 # key = tempfile.NamedTemporaryFile()
1821 # with open(key.name, "w") as key_file:
1822 # key_file.write("key")
1823 # mock_configuration.ssl_ca_cert = None
1824 # mock_configuration.cert_file = None
1825 # mock_configuration.key_file = key.name
1826 # exception_raised = False
1828 # _ = self.libjuju.get_k8s_cloud_credential(
1829 # mock_configuration,
1833 # except JujuInvalidK8sConfiguration as e:
1834 # exception_raised = True
1837 # "missing token for auth type oauth2",
1839 # self.assertTrue(exception_raised)
1841 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1842 mock_configuration
.username
= "admin"
1843 mock_configuration
.password
= "pass"
1844 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1845 mock_configuration
.ssl_ca_cert
= None
1846 mock_configuration
.cert_file
= None
1847 mock_configuration
.key_file
= None
1848 exception_raised
= False
1850 _
= self
.libjuju
.get_k8s_cloud_credential(
1855 except JujuInvalidK8sConfiguration
as e
:
1856 exception_raised
= True
1859 "Cannot set both token and user/pass",
1861 self
.assertTrue(exception_raised
)