1 # Copyright 2020 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
18 from unittest
import mock
21 from juju
.errors
import JujuAPIError
23 from .utils
import FakeN2VC
, FakeMachine
, FakeApplication
24 from n2vc
.libjuju
import Libjuju
25 from n2vc
.exceptions
import (
26 JujuControllerFailedConnecting
,
27 JujuModelAlreadyExists
,
29 JujuApplicationNotFound
,
31 JujuApplicationExists
,
32 JujuInvalidK8sConfiguration
,
33 JujuLeaderUnitNotFound
,
37 class LibjujuTestCase(asynctest
.TestCase
):
38 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
39 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
40 @asynctest.mock
.patch("juju.controller.Controller.connection")
41 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
44 mock__get_api_endpoints_db
=None,
47 mock_update_endpoints
=None,
49 loop
= asyncio
.get_event_loop()
51 mock__get_api_endpoints_db
.return_value
= ["127.0.0.1:17070"]
52 endpoints
= "127.0.0.1:17070"
56 -----BEGIN CERTIFICATE-----
58 -----END CERTIFICATE-----"""
59 self
.libjuju
= Libjuju(
61 "192.168.0.155:17070",
69 apt_mirror
="192.168.0.100",
70 enable_os_upgrade
=True,
72 logging
.disable(logging
.CRITICAL
)
73 loop
.run_until_complete(self
.libjuju
.disconnect())
76 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._create_health_check_task")
77 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
78 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
79 class LibjujuInitTestCase(asynctest
.TestCase
):
81 self
.loop
= asyncio
.get_event_loop()
82 self
.n2vc
= FakeN2VC()
83 self
.endpoint
= "192.168.100.100:17070"
84 self
.username
= "admin"
85 self
.password
= "secret"
87 -----BEGIN CERTIFICATE-----
89 -----END CERTIFICATE-----"""
91 def test_endpoint_not_in_db(
93 mock__get_api_endpoints_db
,
94 mock_update_endpoints
,
95 mock_create_health_check_task
,
97 mock__get_api_endpoints_db
.return_value
= ["another_ip"]
100 "192.168.0.155:17070",
108 apt_mirror
="192.168.0.100",
109 enable_os_upgrade
=True,
111 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
112 mock__get_api_endpoints_db
.assert_called_once()
114 def test_endpoint_in_db(
116 mock__get_api_endpoints_db
,
117 mock_update_endpoints
,
118 mock_create_health_check_task
,
120 mock__get_api_endpoints_db
.return_value
= [self
.endpoint
, "another_ip"]
123 "192.168.0.155:17070",
131 apt_mirror
="192.168.0.100",
132 enable_os_upgrade
=True,
134 mock_update_endpoints
.assert_not_called()
135 mock__get_api_endpoints_db
.assert_called_once()
137 def test_no_db_endpoints(
139 mock__get_api_endpoints_db
,
140 mock_update_endpoints
,
141 mock_create_health_check_task
,
143 mock__get_api_endpoints_db
.return_value
= None
146 "192.168.0.155:17070",
154 apt_mirror
="192.168.0.100",
155 enable_os_upgrade
=True,
157 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
158 mock__get_api_endpoints_db
.assert_called_once()
161 @asynctest.mock
.patch("juju.controller.Controller.connect")
162 @asynctest.mock
.patch(
163 "juju.controller.Controller.api_endpoints",
164 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
166 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
167 class GetControllerTest(LibjujuTestCase
):
169 super(GetControllerTest
, self
).setUp()
171 def test_diff_endpoint(
172 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
174 self
.libjuju
.endpoints
= []
175 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
176 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
177 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
179 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
182 mock_disconnect_controller
,
183 mock__update_api_endpoints_db
,
187 self
.libjuju
.endpoints
= []
188 mock__update_api_endpoints_db
.side_effect
= Exception()
190 with self
.assertRaises(JujuControllerFailedConnecting
):
191 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
192 self
.assertIsNone(controller
)
193 mock_disconnect_controller
.assert_called_once()
195 def test_same_endpoint_get_controller(
196 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
198 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
199 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
200 mock__update_api_endpoints_db
.assert_not_called()
201 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
204 class DisconnectTest(LibjujuTestCase
):
206 super(DisconnectTest
, self
).setUp()
208 @asynctest.mock
.patch("juju.model.Model.disconnect")
209 def test_disconnect_model(self
, mock_disconnect
):
210 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
211 mock_disconnect
.assert_called_once()
213 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
214 def test_disconnect_controller(self
, mock_disconnect
):
215 self
.loop
.run_until_complete(
216 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
218 mock_disconnect
.assert_called_once()
221 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
222 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
223 @asynctest.mock
.patch("juju.controller.Controller.add_model")
224 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
225 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
226 class AddModelTest(LibjujuTestCase
):
228 super(AddModelTest
, self
).setUp()
230 def test_existing_model(
232 mock_disconnect_model
,
233 mock_disconnect_controller
,
238 mock_model_exists
.return_value
= True
240 with self
.assertRaises(JujuModelAlreadyExists
):
241 self
.loop
.run_until_complete(
242 self
.libjuju
.add_model("existing_model", "cloud")
245 mock_disconnect_controller
.assert_called()
247 # TODO Check two job executing at the same time and one returning without doing anything.
249 def test_non_existing_model(
251 mock_disconnect_model
,
252 mock_disconnect_controller
,
257 mock_model_exists
.return_value
= False
258 mock_get_controller
.return_value
= juju
.controller
.Controller()
260 self
.loop
.run_until_complete(
261 self
.libjuju
.add_model("nonexisting_model", "cloud")
264 mock_add_model
.assert_called_once()
265 mock_disconnect_controller
.assert_called()
266 mock_disconnect_model
.assert_called()
269 @asynctest.mock
.patch("juju.controller.Controller.get_model")
270 class GetModelTest(LibjujuTestCase
):
272 super(GetModelTest
, self
).setUp()
275 self
, mock_get_model
,
277 mock_get_model
.return_value
= juju
.model
.Model()
278 model
= self
.loop
.run_until_complete(
279 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
281 self
.assertIsInstance(model
, juju
.model
.Model
)
284 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
285 @asynctest.mock
.patch("juju.controller.Controller.list_models")
286 class ModelExistsTest(LibjujuTestCase
):
288 super(ModelExistsTest
, self
).setUp()
290 async def test_existing_model(
291 self
, mock_list_models
, mock_get_controller
,
293 mock_list_models
.return_value
= ["existing_model"]
295 await self
.libjuju
.model_exists(
296 "existing_model", juju
.controller
.Controller()
300 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
301 async def test_no_controller(
302 self
, mock_disconnect_controller
, mock_list_models
, mock_get_controller
,
304 mock_list_models
.return_value
= ["existing_model"]
305 mock_get_controller
.return_value
= juju
.controller
.Controller()
306 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
307 mock_disconnect_controller
.assert_called_once()
309 async def test_non_existing_model(
310 self
, mock_list_models
, mock_get_controller
,
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(
461 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
463 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
464 @asynctest.mock
.patch("juju.model.Model.deploy")
465 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
466 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
467 class DeployCharmTest(LibjujuTestCase
):
469 super(DeployCharmTest
, self
).setUp()
471 def test_existing_app(
478 mock_disconnect_controller
,
479 mock_disconnect_model
,
483 mock_get_model
.return_value
= juju
.model
.Model()
484 mock_applications
.return_value
= {"existing_app"}
487 with self
.assertRaises(JujuApplicationExists
):
488 application
= self
.loop
.run_until_complete(
489 self
.libjuju
.deploy_charm("existing_app", "path", "model", "machine",)
491 self
.assertIsNone(application
)
493 mock_disconnect_controller
.assert_called()
494 mock_disconnect_model
.assert_called()
496 def test_non_existing_machine(
503 mock_disconnect_controller
,
504 mock_disconnect_model
,
508 mock_get_model
.return_value
= juju
.model
.Model()
509 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
511 with self
.assertRaises(JujuMachineNotFound
):
512 application
= self
.loop
.run_until_complete(
513 self
.libjuju
.deploy_charm("app", "path", "model", "machine",)
516 self
.assertIsNone(application
)
518 mock_disconnect_controller
.assert_called()
519 mock_disconnect_model
.assert_called()
528 mock_disconnect_controller
,
529 mock_disconnect_model
,
533 mock_get_model
.return_value
= juju
.model
.Model()
534 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
535 mock_create_machine
.return_value
= (FakeMachine(), "other")
536 mock_deploy
.return_value
= FakeApplication()
537 application
= self
.loop
.run_until_complete(
538 self
.libjuju
.deploy_charm(
539 "app", "path", "model", "existing_machine", num_units
=2,
543 self
.assertIsInstance(application
, FakeApplication
)
545 mock_deploy
.assert_called_once()
546 mock_wait_for
.assert_called_once()
548 mock_create_machine
.assert_called_once()
550 mock_disconnect_controller
.assert_called()
551 mock_disconnect_model
.assert_called()
560 mock_disconnect_controller
,
561 mock_disconnect_model
,
565 mock_get_model
.return_value
= juju
.model
.Model()
566 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
567 mock_deploy
.return_value
= FakeApplication()
568 application
= self
.loop
.run_until_complete(
569 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
572 self
.assertIsInstance(application
, FakeApplication
)
574 mock_deploy
.assert_called_once()
575 mock_wait_for
.assert_called_once()
577 mock_disconnect_controller
.assert_called()
578 mock_disconnect_model
.assert_called()
581 @asynctest.mock
.patch(
582 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
584 class GetApplicationTest(LibjujuTestCase
):
586 super(GetApplicationTest
, self
).setUp()
588 def test_existing_application(
589 self
, mock_applications
,
591 mock_applications
.return_value
= {"existing_app": "exists"}
592 model
= juju
.model
.Model()
593 result
= self
.libjuju
._get
_application
(model
, "existing_app")
594 self
.assertEqual(result
, "exists")
596 def test_non_existing_application(
597 self
, mock_applications
,
599 mock_applications
.return_value
= {"existing_app": "exists"}
600 model
= juju
.model
.Model()
601 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
602 self
.assertIsNone(result
)
605 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
606 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
607 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
608 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
609 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
610 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
611 @asynctest.mock
.patch("juju.model.Model.get_action_output")
612 @asynctest.mock
.patch("juju.model.Model.get_action_status")
613 class ExecuteActionTest(LibjujuTestCase
):
615 super(ExecuteActionTest
, self
).setUp()
617 def test_no_application(
619 mock_get_action_status
,
620 mock_get_action_output
,
622 mock__get_application
,
623 mock_disconnect_controller
,
624 mock_disconnect_model
,
628 mock__get_application
.return_value
= None
629 mock_get_model
.return_value
= juju
.model
.Model()
632 with self
.assertRaises(JujuApplicationNotFound
):
633 output
, status
= self
.loop
.run_until_complete(
634 self
.libjuju
.execute_action("app", "model", "action",)
636 self
.assertIsNone(output
)
637 self
.assertIsNone(status
)
639 mock_disconnect_controller
.assert_called()
640 mock_disconnect_model
.assert_called()
644 mock_get_action_status
,
645 mock_get_action_output
,
647 mock__get_application
,
648 mock_disconnect_controller
,
649 mock_disconnect_model
,
654 mock_get_model
.return_value
= juju
.model
.Model()
655 mock__get_application
.return_value
= FakeApplication()
658 with self
.assertRaises(JujuActionNotFound
):
659 output
, status
= self
.loop
.run_until_complete(
660 self
.libjuju
.execute_action("app", "model", "action",)
662 self
.assertIsNone(output
)
663 self
.assertIsNone(status
)
665 mock_disconnect_controller
.assert_called()
666 mock_disconnect_model
.assert_called()
668 @asynctest.mock
.patch("asyncio.sleep")
669 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
672 mock_is_leader_from_status
,
674 mock_get_action_status
,
675 mock_get_action_output
,
677 mock__get_application
,
678 mock_disconnect_controller
,
679 mock_disconnect_model
,
683 mock_get_model
.return_value
= juju
.model
.Model()
684 mock__get_application
.return_value
= FakeApplication()
685 mock_is_leader_from_status
.return_value
= False
688 with self
.assertRaises(JujuLeaderUnitNotFound
):
689 output
, status
= self
.loop
.run_until_complete(
690 self
.libjuju
.execute_action("app", "model", "action",)
692 self
.assertIsNone(output
)
693 self
.assertIsNone(status
)
695 mock_disconnect_controller
.assert_called()
696 mock_disconnect_model
.assert_called()
698 def test_succesful_exec(
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_model
.return_value
= juju
.model
.Model()
710 mock__get_application
.return_value
= FakeApplication()
711 mock_get_action_output
.return_value
= "output"
712 mock_get_action_status
.return_value
= {"id": "status"}
713 output
, status
= self
.loop
.run_until_complete(
714 self
.libjuju
.execute_action("app", "model", "existing_action")
716 self
.assertEqual(output
, "output")
717 self
.assertEqual(status
, "status")
719 mock_wait_for
.assert_called_once()
721 mock_disconnect_controller
.assert_called()
722 mock_disconnect_model
.assert_called()
725 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
726 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
727 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
728 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
729 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
730 class GetActionTest(LibjujuTestCase
):
732 super(GetActionTest
, self
).setUp()
736 mock_get_application
,
737 mock_disconnect_controller
,
738 mock_disconnect_model
,
742 mock_get_application
.side_effect
= Exception()
744 with self
.assertRaises(Exception):
745 actions
= self
.loop
.run_until_complete(
746 self
.libjuju
.get_actions("app", "model")
749 self
.assertIsNone(actions
)
750 mock_disconnect_controller
.assert_called_once()
751 mock_disconnect_model
.assert_called_once()
755 mock_get_application
,
756 mock_disconnect_controller
,
757 mock_disconnect_model
,
761 mock_get_application
.return_value
= FakeApplication()
763 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
765 self
.assertEqual(actions
, ["existing_action"])
767 mock_get_controller
.assert_called_once()
768 mock_get_model
.assert_called_once()
769 mock_disconnect_controller
.assert_called_once()
770 mock_disconnect_model
.assert_called_once()
773 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
774 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
775 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
776 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
777 @asynctest.mock
.patch("juju.application.Application.get_metrics")
778 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
779 class GetMetricsTest(LibjujuTestCase
):
781 super(GetMetricsTest
, self
).setUp()
783 def test_get_metrics_success(
785 mock_get_application
,
787 mock_disconnect_controller
,
788 mock_disconnect_model
,
792 mock_get_application
.return_value
= FakeApplication()
793 mock_get_model
.return_value
= juju
.model
.Model()
795 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
797 mock_disconnect_controller
.assert_called_once()
798 mock_disconnect_model
.assert_called_once()
800 def test_get_metrics_exception(
802 mock_get_application
,
804 mock_disconnect_controller
,
805 mock_disconnect_model
,
809 mock_get_model
.return_value
= juju
.model
.Model()
810 mock_get_metrics
.side_effect
= Exception()
811 with self
.assertRaises(Exception):
812 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
814 mock_disconnect_controller
.assert_called_once()
815 mock_disconnect_model
.assert_called_once()
817 def test_missing_args_exception(
819 mock_get_application
,
821 mock_disconnect_controller
,
822 mock_disconnect_model
,
826 mock_get_model
.return_value
= juju
.model
.Model()
828 with self
.assertRaises(Exception):
829 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
831 mock_get_controller
.assert_not_called()
832 mock_get_model
.assert_not_called()
833 mock_disconnect_controller
.assert_not_called()
834 mock_disconnect_model
.assert_not_called()
837 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
838 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
839 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
840 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
841 @asynctest.mock
.patch("juju.model.Model.add_relation")
842 class AddRelationTest(LibjujuTestCase
):
844 super(AddRelationTest
, self
).setUp()
846 @asynctest.mock
.patch("logging.Logger.warning")
851 mock_disconnect_controller
,
852 mock_disconnect_model
,
856 # TODO in libjuju.py should this fail only with a log message?
857 result
= {"error": "not found", "response": "response", "request-id": 1}
859 mock_get_model
.return_value
= juju
.model
.Model()
860 mock_add_relation
.side_effect
= JujuAPIError(result
)
862 self
.loop
.run_until_complete(
863 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
866 mock_warning
.assert_called_with("Relation not found: not found")
867 mock_disconnect_controller
.assert_called_once()
868 mock_disconnect_model
.assert_called_once()
870 @asynctest.mock
.patch("logging.Logger.warning")
871 def test_already_exists(
875 mock_disconnect_controller
,
876 mock_disconnect_model
,
880 # TODO in libjuju.py should this fail silently?
881 result
= {"error": "already exists", "response": "response", "request-id": 1}
883 mock_get_model
.return_value
= juju
.model
.Model()
884 mock_add_relation
.side_effect
= JujuAPIError(result
)
886 self
.loop
.run_until_complete(
887 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
890 mock_warning
.assert_called_with("Relation already exists: already exists")
891 mock_disconnect_controller
.assert_called_once()
892 mock_disconnect_model
.assert_called_once()
897 mock_disconnect_controller
,
898 mock_disconnect_model
,
902 mock_get_model
.return_value
= juju
.model
.Model()
903 result
= {"error": "", "response": "response", "request-id": 1}
904 mock_add_relation
.side_effect
= JujuAPIError(result
)
906 with self
.assertRaises(JujuAPIError
):
907 self
.loop
.run_until_complete(
908 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
911 mock_disconnect_controller
.assert_called_once()
912 mock_disconnect_model
.assert_called_once()
917 mock_disconnect_controller
,
918 mock_disconnect_model
,
922 mock_get_model
.return_value
= juju
.model
.Model()
924 self
.loop
.run_until_complete(
925 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
928 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
929 mock_disconnect_controller
.assert_called_once()
930 mock_disconnect_model
.assert_called_once()
935 mock_disconnect_controller
,
936 mock_disconnect_model
,
940 mock_get_model
.return_value
= juju
.model
.Model()
942 self
.loop
.run_until_complete(
943 self
.libjuju
.add_relation("model", "app1:relation1", "saas_name",)
946 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
947 mock_disconnect_controller
.assert_called_once()
948 mock_disconnect_model
.assert_called_once()
951 # TODO destroy_model testcase
954 # @asynctest.mock.patch("juju.model.Model.get_machines")
955 # @asynctest.mock.patch("logging.Logger.debug")
956 # class DestroyMachineTest(LibjujuTestCase):
958 # super(DestroyMachineTest, self).setUp()
960 # def test_success_manual_machine(
961 # self, mock_debug, mock_get_machines,
963 # mock_get_machines.side_effect = [
964 # {"machine": FakeManualMachine()},
965 # {"machine": FakeManualMachine()},
968 # self.loop.run_until_complete(
969 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
972 # asynctest.call("Waiting for machine machine is destroyed"),
973 # asynctest.call("Machine destroyed: machine"),
975 # mock_debug.assert_has_calls(calls)
977 # def test_no_machine(
978 # self, mock_debug, mock_get_machines,
980 # mock_get_machines.return_value = {}
981 # self.loop.run_until_complete(
982 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
984 # mock_debug.assert_called_with("Machine not found: machine")
987 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
988 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
989 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
990 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
991 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
992 class ConfigureApplicationTest(LibjujuTestCase
):
994 super(ConfigureApplicationTest
, self
).setUp()
998 mock_get_application
,
999 mock_disconnect_controller
,
1000 mock_disconnect_model
,
1002 mock_get_controller
,
1005 mock_get_application
.return_value
= FakeApplication()
1007 self
.loop
.run_until_complete(
1008 self
.libjuju
.configure_application("model", "app", {"config"},)
1010 mock_get_application
.assert_called_once()
1011 mock_disconnect_controller
.assert_called_once()
1012 mock_disconnect_model
.assert_called_once()
1016 mock_get_application
,
1017 mock_disconnect_controller
,
1018 mock_disconnect_model
,
1020 mock_get_controller
,
1023 mock_get_application
.side_effect
= Exception()
1025 with self
.assertRaises(Exception):
1026 self
.loop
.run_until_complete(
1027 self
.libjuju
.configure_application("model", "app", {"config"},)
1029 mock_disconnect_controller
.assert_called_once()
1030 mock_disconnect_model
.assert_called_once()
1033 # TODO _get_api_endpoints_db test case
1034 # TODO _update_api_endpoints_db test case
1035 # TODO healthcheck test case
1038 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1039 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1040 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1041 class ListModelsTest(LibjujuTestCase
):
1043 super(ListModelsTest
, self
).setUp()
1045 def test_containing(
1046 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
1048 mock_get_controller
.return_value
= juju
.controller
.Controller()
1049 mock_list_models
.return_value
= ["existingmodel"]
1050 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1052 mock_disconnect_controller
.assert_called_once()
1053 self
.assertEquals(models
, ["existingmodel"])
1055 def test_not_containing(
1056 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
1058 mock_get_controller
.return_value
= juju
.controller
.Controller()
1059 mock_list_models
.return_value
= ["existingmodel", "model"]
1060 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1062 mock_disconnect_controller
.assert_called_once()
1063 self
.assertEquals(models
, [])
1065 def test_no_contains_arg(
1066 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
1068 mock_get_controller
.return_value
= juju
.controller
.Controller()
1069 mock_list_models
.return_value
= ["existingmodel", "model"]
1070 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1072 mock_disconnect_controller
.assert_called_once()
1073 self
.assertEquals(models
, ["existingmodel", "model"])
1076 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1077 class ModelsExistTest(LibjujuTestCase
):
1079 super(ModelsExistTest
, self
).setUp()
1081 def test_model_names_none(self
, mock_list_models
):
1082 mock_list_models
.return_value
= []
1083 with self
.assertRaises(Exception):
1084 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1086 def test_model_names_empty(self
, mock_list_models
):
1087 mock_list_models
.return_value
= []
1088 with self
.assertRaises(Exception):
1089 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1090 self
.libjuju
.models_exist([])
1093 def test_model_names_not_existing(self
, mock_list_models
):
1094 mock_list_models
.return_value
= ["prometheus", "grafana"]
1095 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1096 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1098 self
.assertFalse(exist
)
1099 self
.assertEqual(non_existing_models
, ["prometheus2"])
1101 def test_model_names_exist(self
, mock_list_models
):
1102 mock_list_models
.return_value
= ["prometheus", "grafana"]
1103 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1104 self
.libjuju
.models_exist(["prometheus", "grafana"])
1106 self
.assertTrue(exist
)
1107 self
.assertEqual(non_existing_models
, [])
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1112 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1113 class ListOffers(LibjujuTestCase
):
1115 super(ListOffers
, self
).setUp()
1117 def test_disconnect_controller(
1118 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1120 mock_get_controller
.return_value
= juju
.controller
.Controller()
1121 mock_list_offers
.side_effect
= Exception()
1122 with self
.assertRaises(Exception):
1123 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1124 mock_disconnect_controller
.assert_called_once()
1126 def test_empty_list(
1127 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1129 mock_get_controller
.return_value
= juju
.controller
.Controller()
1130 mock_list_offers
.return_value
= []
1131 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1132 self
.assertEqual(offers
, [])
1133 mock_disconnect_controller
.assert_called_once()
1135 def test_non_empty_list(
1136 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1138 mock_get_controller
.return_value
= juju
.controller
.Controller()
1139 mock_list_offers
.return_value
= ["offer"]
1140 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1141 self
.assertEqual(offers
, ["offer"])
1142 mock_disconnect_controller
.assert_called_once()
1145 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1146 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1147 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1148 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1149 @asynctest.mock
.patch("juju.model.Model.consume")
1150 class ConsumeTest(LibjujuTestCase
):
1152 super(ConsumeTest
, self
).setUp()
1157 mock_disconnect_controller
,
1158 mock_disconnect_model
,
1160 mock_get_controller
,
1162 mock_get_controller
.return_value
= juju
.controller
.Controller()
1163 mock_get_model
.return_value
= juju
.model
.Model()
1165 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1166 mock_consume
.assert_called_once()
1167 mock_disconnect_model
.assert_called_once()
1168 mock_disconnect_controller
.assert_called_once()
1170 def test_parsing_error_exception(
1173 mock_disconnect_controller
,
1174 mock_disconnect_model
,
1176 mock_get_controller
,
1178 mock_get_controller
.return_value
= juju
.controller
.Controller()
1179 mock_get_model
.return_value
= juju
.model
.Model()
1180 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1182 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1183 self
.loop
.run_until_complete(
1184 self
.libjuju
.consume("offer_url", "model_name")
1186 mock_consume
.assert_called_once()
1187 mock_disconnect_model
.assert_called_once()
1188 mock_disconnect_controller
.assert_called_once()
1190 def test_juju_error_exception(
1193 mock_disconnect_controller
,
1194 mock_disconnect_model
,
1196 mock_get_controller
,
1198 mock_get_controller
.return_value
= juju
.controller
.Controller()
1199 mock_get_model
.return_value
= juju
.model
.Model()
1200 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1202 with self
.assertRaises(juju
.errors
.JujuError
):
1203 self
.loop
.run_until_complete(
1204 self
.libjuju
.consume("offer_url", "model_name")
1206 mock_consume
.assert_called_once()
1207 mock_disconnect_model
.assert_called_once()
1208 mock_disconnect_controller
.assert_called_once()
1210 def test_juju_api_error_exception(
1213 mock_disconnect_controller
,
1214 mock_disconnect_model
,
1216 mock_get_controller
,
1218 mock_get_controller
.return_value
= juju
.controller
.Controller()
1219 mock_get_model
.return_value
= juju
.model
.Model()
1220 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1221 {"error": "", "response": "", "request-id": ""}
1224 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1225 self
.loop
.run_until_complete(
1226 self
.libjuju
.consume("offer_url", "model_name")
1228 mock_consume
.assert_called_once()
1229 mock_disconnect_model
.assert_called_once()
1230 mock_disconnect_controller
.assert_called_once()
1233 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1234 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1235 class AddK8sTest(LibjujuTestCase
):
1237 super(AddK8sTest
, self
).setUp()
1238 self
.configuration
= kubernetes
.client
.configuration
.Configuration()
1240 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1241 self
.loop
.run_until_complete(
1242 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1244 mock_add_cloud
.assert_called_once()
1245 mock_get_k8s_cloud_credential
.assert_called_once()
1247 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1248 mock_add_cloud
.side_effect
= Exception()
1249 with self
.assertRaises(Exception):
1250 self
.loop
.run_until_complete(
1251 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1253 mock_add_cloud
.assert_called_once()
1254 mock_get_k8s_cloud_credential
.assert_called_once()
1256 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1257 with self
.assertRaises(Exception):
1258 self
.loop
.run_until_complete(
1259 self
.libjuju
.add_k8s("", self
.configuration
, "storage_class")
1261 mock_add_cloud
.assert_not_called()
1263 def test_add_k8s_missing_storage_name(
1264 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1266 with self
.assertRaises(Exception):
1267 self
.loop
.run_until_complete(
1268 self
.libjuju
.add_k8s("cloud", self
.configuration
, "")
1270 mock_add_cloud
.assert_not_called()
1272 def test_add_k8s_missing_configuration_keys(
1273 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1275 with self
.assertRaises(Exception):
1276 self
.loop
.run_until_complete(self
.libjuju
.add_k8s("cloud", None, ""))
1277 mock_add_cloud
.assert_not_called()
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.add_cloud")
1283 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1284 class AddCloudTest(LibjujuTestCase
):
1286 super(AddCloudTest
, self
).setUp()
1287 self
.cloud
= juju
.client
.client
.Cloud()
1288 self
.credential
= juju
.client
.client
.CloudCredential()
1290 def test_add_cloud_with_credential(
1292 mock_add_credential
,
1294 mock_disconnect_controller
,
1295 mock_get_controller
,
1297 mock_get_controller
.return_value
= juju
.controller
.Controller()
1299 cloud
= self
.loop
.run_until_complete(
1300 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1302 self
.assertEqual(cloud
, self
.cloud
)
1303 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1304 mock_add_credential
.assert_called_once_with(
1305 "cloud", credential
=self
.credential
, cloud
="cloud"
1307 mock_disconnect_controller
.assert_called_once()
1309 def test_add_cloud_no_credential(
1311 mock_add_credential
,
1313 mock_disconnect_controller
,
1314 mock_get_controller
,
1316 mock_get_controller
.return_value
= juju
.controller
.Controller()
1318 cloud
= self
.loop
.run_until_complete(
1319 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1321 self
.assertEqual(cloud
, self
.cloud
)
1322 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1323 mock_add_credential
.assert_not_called()
1324 mock_disconnect_controller
.assert_called_once()
1326 def test_add_cloud_exception(
1328 mock_add_credential
,
1330 mock_disconnect_controller
,
1331 mock_get_controller
,
1333 mock_get_controller
.return_value
= juju
.controller
.Controller()
1334 mock_add_cloud
.side_effect
= Exception()
1335 with self
.assertRaises(Exception):
1336 self
.loop
.run_until_complete(
1337 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1340 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1341 mock_add_credential
.assert_not_called()
1342 mock_disconnect_controller
.assert_called_once()
1344 def test_add_credential_exception(
1346 mock_add_credential
,
1348 mock_disconnect_controller
,
1349 mock_get_controller
,
1351 mock_get_controller
.return_value
= juju
.controller
.Controller()
1352 mock_add_credential
.side_effect
= Exception()
1353 with self
.assertRaises(Exception):
1354 self
.loop
.run_until_complete(
1355 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1358 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1359 mock_add_credential
.assert_called_once_with(
1360 "cloud", credential
=self
.credential
, cloud
="cloud"
1362 mock_disconnect_controller
.assert_called_once()
1365 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1366 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1367 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1368 class RemoveCloudTest(LibjujuTestCase
):
1370 super(RemoveCloudTest
, self
).setUp()
1372 def test_remove_cloud(
1373 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1375 mock_get_controller
.return_value
= juju
.controller
.Controller()
1377 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1378 mock_remove_cloud
.assert_called_once_with("cloud")
1379 mock_disconnect_controller
.assert_called_once()
1381 def test_remove_cloud_exception(
1382 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1384 mock_get_controller
.return_value
= juju
.controller
.Controller()
1385 mock_remove_cloud
.side_effect
= Exception()
1387 with self
.assertRaises(Exception):
1388 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1389 mock_remove_cloud
.assert_called_once_with("cloud")
1390 mock_disconnect_controller
.assert_called_once()
1393 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1394 class GetK8sCloudCredentials(LibjujuTestCase
):
1396 super(GetK8sCloudCredentials
, self
).setUp()
1398 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1399 def test_not_supported(self
, mock_exception
, mock_configuration
):
1400 mock_configuration
.username
= ""
1401 mock_configuration
.password
= ""
1402 mock_configuration
.ssl_ca_cert
= None
1403 mock_configuration
.cert_file
= None
1404 mock_configuration
.key_file
= None
1405 exception_raised
= False
1407 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1408 except JujuInvalidK8sConfiguration
as e
:
1409 exception_raised
= True
1411 e
.message
, "authentication method not supported",
1413 self
.assertTrue(exception_raised
)
1415 def test_user_pass(self
, mock_configuration
):
1416 mock_configuration
.username
= "admin"
1417 mock_configuration
.password
= "admin"
1418 mock_configuration
.ssl_ca_cert
= None
1419 mock_configuration
.cert_file
= None
1420 mock_configuration
.key_file
= None
1421 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1424 juju
.client
._definitions
.CloudCredential(
1425 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1429 def test_user_no_pass(self
, mock_configuration
):
1430 mock_configuration
.username
= "admin"
1431 mock_configuration
.password
= ""
1432 mock_configuration
.ssl_ca_cert
= None
1433 mock_configuration
.cert_file
= None
1434 mock_configuration
.key_file
= None
1435 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1436 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1439 juju
.client
._definitions
.CloudCredential(
1440 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1443 mock_debug
.assert_called_once_with(
1444 "credential for user admin has empty password"
1447 def test_user_pass_with_cert(self
, mock_configuration
):
1448 mock_configuration
.username
= "admin"
1449 mock_configuration
.password
= "admin"
1450 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1451 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1452 ssl_ca_cert_file
.write("cacert")
1453 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1454 mock_configuration
.cert_file
= None
1455 mock_configuration
.key_file
= None
1456 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1459 juju
.client
._definitions
.CloudCredential(
1461 "username": "admin",
1462 "password": "admin",
1463 "ClientCertificateData": "cacert",
1465 auth_type
="userpasswithcert",
1469 def test_cert(self
, mock_configuration
):
1470 mock_configuration
.username
= ""
1471 mock_configuration
.password
= ""
1472 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1473 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1474 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1475 ssl_ca_cert_file
.write("cacert")
1476 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1477 mock_configuration
.cert_file
= None
1478 mock_configuration
.key_file
= None
1479 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1482 juju
.client
._definitions
.CloudCredential(
1483 attrs
={"ClientCertificateData": "cacert", "Token": "Token"},
1484 auth_type
="certificate",
1488 def test_oauth2(self
, mock_configuration
):
1489 mock_configuration
.username
= ""
1490 mock_configuration
.password
= ""
1491 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1492 key
= tempfile
.NamedTemporaryFile()
1493 with
open(key
.name
, "w") as key_file
:
1494 key_file
.write("key")
1495 mock_configuration
.ssl_ca_cert
= None
1496 mock_configuration
.cert_file
= None
1497 mock_configuration
.key_file
= key
.name
1498 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1501 juju
.client
._definitions
.CloudCredential(
1502 attrs
={"ClientKeyData": "key", "Token": "Token"}, auth_type
="oauth2",
1506 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1507 def test_oauth2_missing_token(self
, mock_exception
, mock_configuration
):
1508 mock_configuration
.username
= ""
1509 mock_configuration
.password
= ""
1510 key
= tempfile
.NamedTemporaryFile()
1511 with
open(key
.name
, "w") as key_file
:
1512 key_file
.write("key")
1513 mock_configuration
.ssl_ca_cert
= None
1514 mock_configuration
.cert_file
= None
1515 mock_configuration
.key_file
= key
.name
1516 exception_raised
= False
1518 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1519 except JujuInvalidK8sConfiguration
as e
:
1520 exception_raised
= True
1522 e
.message
, "missing token for auth type oauth2",
1524 self
.assertTrue(exception_raised
)
1526 def test_unknown_api_key(self
, mock_configuration
):
1527 mock_configuration
.username
= ""
1528 mock_configuration
.password
= ""
1529 mock_configuration
.api_key
= {"authorization": "Bearer Token Wrong"}
1530 mock_configuration
.ssl_ca_cert
= None
1531 mock_configuration
.cert_file
= None
1532 mock_configuration
.key_file
= None
1533 exception_raised
= False
1535 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1536 except JujuInvalidK8sConfiguration
as e
:
1537 exception_raised
= True
1539 e
.message
, "unknown format of api_key",
1541 self
.assertTrue(exception_raised
)
1543 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1544 mock_configuration
.username
= "admin"
1545 mock_configuration
.password
= "pass"
1546 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1547 mock_configuration
.ssl_ca_cert
= None
1548 mock_configuration
.cert_file
= None
1549 mock_configuration
.key_file
= None
1550 exception_raised
= False
1552 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1553 except JujuInvalidK8sConfiguration
as e
:
1554 exception_raised
= True
1556 e
.message
, "Cannot set both token and user/pass",
1558 self
.assertTrue(exception_raised
)