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("juju.controller.Controller.connect")
77 @asynctest.mock
.patch(
78 "juju.controller.Controller.api_endpoints",
79 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
81 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
82 class GetControllerTest(LibjujuTestCase
):
84 super(GetControllerTest
, self
).setUp()
86 def test_diff_endpoint(
87 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
89 self
.libjuju
.endpoints
= []
90 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
91 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
92 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
94 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
97 mock_disconnect_controller
,
98 mock__update_api_endpoints_db
,
102 self
.libjuju
.endpoints
= []
103 mock__update_api_endpoints_db
.side_effect
= Exception()
105 with self
.assertRaises(JujuControllerFailedConnecting
):
106 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
107 self
.assertIsNone(controller
)
108 mock_disconnect_controller
.assert_called_once()
110 def test_same_endpoint_get_controller(
111 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
113 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
114 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
115 mock__update_api_endpoints_db
.assert_not_called()
116 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
119 class DisconnectTest(LibjujuTestCase
):
121 super(DisconnectTest
, self
).setUp()
123 @asynctest.mock
.patch("juju.model.Model.disconnect")
124 def test_disconnect_model(self
, mock_disconnect
):
125 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
126 mock_disconnect
.assert_called_once()
128 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
129 def test_disconnect_controller(self
, mock_disconnect
):
130 self
.loop
.run_until_complete(
131 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
133 mock_disconnect
.assert_called_once()
136 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
137 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
138 @asynctest.mock
.patch("juju.controller.Controller.add_model")
139 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
140 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
141 class AddModelTest(LibjujuTestCase
):
143 super(AddModelTest
, self
).setUp()
145 def test_existing_model(
147 mock_disconnect_model
,
148 mock_disconnect_controller
,
153 mock_model_exists
.return_value
= True
155 with self
.assertRaises(JujuModelAlreadyExists
):
156 self
.loop
.run_until_complete(
157 self
.libjuju
.add_model("existing_model", "cloud")
160 mock_disconnect_controller
.assert_called()
162 # TODO Check two job executing at the same time and one returning without doing anything.
164 def test_non_existing_model(
166 mock_disconnect_model
,
167 mock_disconnect_controller
,
172 mock_model_exists
.return_value
= False
173 mock_get_controller
.return_value
= juju
.controller
.Controller()
175 self
.loop
.run_until_complete(
176 self
.libjuju
.add_model("nonexisting_model", "cloud")
179 mock_add_model
.assert_called_once()
180 mock_disconnect_controller
.assert_called()
181 mock_disconnect_model
.assert_called()
184 @asynctest.mock
.patch("juju.controller.Controller.get_model")
185 class GetModelTest(LibjujuTestCase
):
187 super(GetModelTest
, self
).setUp()
190 self
, mock_get_model
,
192 mock_get_model
.return_value
= juju
.model
.Model()
193 model
= self
.loop
.run_until_complete(
194 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
196 self
.assertIsInstance(model
, juju
.model
.Model
)
199 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
200 @asynctest.mock
.patch("juju.controller.Controller.list_models")
201 class ModelExistsTest(LibjujuTestCase
):
203 super(ModelExistsTest
, self
).setUp()
205 async def test_existing_model(
206 self
, mock_list_models
, mock_get_controller
,
208 mock_list_models
.return_value
= ["existing_model"]
210 await self
.libjuju
.model_exists(
211 "existing_model", juju
.controller
.Controller()
215 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
216 async def test_no_controller(
217 self
, mock_disconnect_controller
, mock_list_models
, mock_get_controller
,
219 mock_list_models
.return_value
= ["existing_model"]
220 mock_get_controller
.return_value
= juju
.controller
.Controller()
221 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
222 mock_disconnect_controller
.assert_called_once()
224 async def test_non_existing_model(
225 self
, mock_list_models
, mock_get_controller
,
227 mock_list_models
.return_value
= ["existing_model"]
229 await self
.libjuju
.model_exists(
230 "not_existing_model", juju
.controller
.Controller()
235 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
236 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
237 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
238 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
239 @asynctest.mock
.patch("juju.model.Model.get_status")
240 class GetModelStatusTest(LibjujuTestCase
):
242 super(GetModelStatusTest
, self
).setUp()
247 mock_disconnect_controller
,
248 mock_disconnect_model
,
252 mock_get_model
.return_value
= juju
.model
.Model()
253 mock_get_status
.return_value
= {"status"}
255 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
257 mock_get_status
.assert_called_once()
258 mock_disconnect_controller
.assert_called_once()
259 mock_disconnect_model
.assert_called_once()
261 self
.assertEqual(status
, {"status"})
266 mock_disconnect_controller
,
267 mock_disconnect_model
,
271 mock_get_model
.return_value
= juju
.model
.Model()
272 mock_get_status
.side_effect
= Exception()
274 with self
.assertRaises(Exception):
275 status
= self
.loop
.run_until_complete(
276 self
.libjuju
.get_model_status("model")
279 mock_disconnect_controller
.assert_called_once()
280 mock_disconnect_model
.assert_called_once()
282 self
.assertIsNone(status
)
285 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
286 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
287 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
288 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
289 @asynctest.mock
.patch("juju.model.Model.get_machines")
290 @asynctest.mock
.patch("juju.model.Model.add_machine")
291 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
292 class CreateMachineTest(LibjujuTestCase
):
294 super(CreateMachineTest
, self
).setUp()
296 def test_existing_machine(
301 mock_disconnect_controller
,
302 mock_disconnect_model
,
306 mock_get_model
.return_value
= juju
.model
.Model()
307 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
308 machine
, bool_res
= self
.loop
.run_until_complete(
309 self
.libjuju
.create_machine("model", "existing_machine")
312 self
.assertIsInstance(machine
, FakeMachine
)
313 self
.assertFalse(bool_res
)
315 mock_disconnect_controller
.assert_called()
316 mock_disconnect_model
.assert_called()
318 def test_non_existing_machine(
323 mock_disconnect_controller
,
324 mock_disconnect_model
,
330 mock_get_model
.return_value
= juju
.model
.Model()
331 with self
.assertRaises(JujuMachineNotFound
):
332 machine
, bool_res
= self
.loop
.run_until_complete(
333 self
.libjuju
.create_machine("model", "non_existing_machine")
335 self
.assertIsNone(machine
)
336 self
.assertIsNone(bool_res
)
338 mock_disconnect_controller
.assert_called()
339 mock_disconnect_model
.assert_called()
346 mock_disconnect_controller
,
347 mock_disconnect_model
,
351 mock_get_model
.return_value
= juju
.model
.Model()
352 mock_add_machine
.return_value
= FakeMachine()
354 machine
, bool_res
= self
.loop
.run_until_complete(
355 self
.libjuju
.create_machine("model")
358 self
.assertIsInstance(machine
, FakeMachine
)
359 self
.assertTrue(bool_res
)
361 mock_wait_for
.assert_called_once()
362 mock_add_machine
.assert_called_once()
364 mock_disconnect_controller
.assert_called()
365 mock_disconnect_model
.assert_called()
368 # TODO test provision machine
371 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
372 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
373 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
374 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
375 @asynctest.mock
.patch(
376 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
378 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
379 @asynctest.mock
.patch("juju.model.Model.deploy")
380 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
381 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
382 class DeployCharmTest(LibjujuTestCase
):
384 super(DeployCharmTest
, self
).setUp()
386 def test_existing_app(
393 mock_disconnect_controller
,
394 mock_disconnect_model
,
398 mock_get_model
.return_value
= juju
.model
.Model()
399 mock_applications
.return_value
= {"existing_app"}
402 with self
.assertRaises(JujuApplicationExists
):
403 application
= self
.loop
.run_until_complete(
404 self
.libjuju
.deploy_charm("existing_app", "path", "model", "machine",)
406 self
.assertIsNone(application
)
408 mock_disconnect_controller
.assert_called()
409 mock_disconnect_model
.assert_called()
411 def test_non_existing_machine(
418 mock_disconnect_controller
,
419 mock_disconnect_model
,
423 mock_get_model
.return_value
= juju
.model
.Model()
424 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
426 with self
.assertRaises(JujuMachineNotFound
):
427 application
= self
.loop
.run_until_complete(
428 self
.libjuju
.deploy_charm("app", "path", "model", "machine",)
431 self
.assertIsNone(application
)
433 mock_disconnect_controller
.assert_called()
434 mock_disconnect_model
.assert_called()
443 mock_disconnect_controller
,
444 mock_disconnect_model
,
448 mock_get_model
.return_value
= juju
.model
.Model()
449 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
450 mock_create_machine
.return_value
= (FakeMachine(), "other")
451 mock_deploy
.return_value
= FakeApplication()
452 application
= self
.loop
.run_until_complete(
453 self
.libjuju
.deploy_charm(
454 "app", "path", "model", "existing_machine", num_units
=2,
458 self
.assertIsInstance(application
, FakeApplication
)
460 mock_deploy
.assert_called_once()
461 mock_wait_for
.assert_called_once()
463 mock_create_machine
.assert_called_once()
465 mock_disconnect_controller
.assert_called()
466 mock_disconnect_model
.assert_called()
475 mock_disconnect_controller
,
476 mock_disconnect_model
,
480 mock_get_model
.return_value
= juju
.model
.Model()
481 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
482 mock_deploy
.return_value
= FakeApplication()
483 application
= self
.loop
.run_until_complete(
484 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
487 self
.assertIsInstance(application
, FakeApplication
)
489 mock_deploy
.assert_called_once()
490 mock_wait_for
.assert_called_once()
492 mock_disconnect_controller
.assert_called()
493 mock_disconnect_model
.assert_called()
496 @asynctest.mock
.patch(
497 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
499 class GetApplicationTest(LibjujuTestCase
):
501 super(GetApplicationTest
, self
).setUp()
503 def test_existing_application(
504 self
, mock_applications
,
506 mock_applications
.return_value
= {"existing_app": "exists"}
507 model
= juju
.model
.Model()
508 result
= self
.libjuju
._get
_application
(model
, "existing_app")
509 self
.assertEqual(result
, "exists")
511 def test_non_existing_application(
512 self
, mock_applications
,
514 mock_applications
.return_value
= {"existing_app": "exists"}
515 model
= juju
.model
.Model()
516 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
517 self
.assertIsNone(result
)
520 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
521 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
522 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
523 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
524 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
525 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
526 @asynctest.mock
.patch("juju.model.Model.get_action_output")
527 @asynctest.mock
.patch("juju.model.Model.get_action_status")
528 class ExecuteActionTest(LibjujuTestCase
):
530 super(ExecuteActionTest
, self
).setUp()
532 def test_no_application(
534 mock_get_action_status
,
535 mock_get_action_output
,
537 mock__get_application
,
538 mock_disconnect_controller
,
539 mock_disconnect_model
,
543 mock__get_application
.return_value
= None
544 mock_get_model
.return_value
= juju
.model
.Model()
547 with self
.assertRaises(JujuApplicationNotFound
):
548 output
, status
= self
.loop
.run_until_complete(
549 self
.libjuju
.execute_action("app", "model", "action",)
551 self
.assertIsNone(output
)
552 self
.assertIsNone(status
)
554 mock_disconnect_controller
.assert_called()
555 mock_disconnect_model
.assert_called()
559 mock_get_action_status
,
560 mock_get_action_output
,
562 mock__get_application
,
563 mock_disconnect_controller
,
564 mock_disconnect_model
,
569 mock_get_model
.return_value
= juju
.model
.Model()
570 mock__get_application
.return_value
= FakeApplication()
573 with self
.assertRaises(JujuActionNotFound
):
574 output
, status
= self
.loop
.run_until_complete(
575 self
.libjuju
.execute_action("app", "model", "action",)
577 self
.assertIsNone(output
)
578 self
.assertIsNone(status
)
580 mock_disconnect_controller
.assert_called()
581 mock_disconnect_model
.assert_called()
583 @asynctest.mock
.patch("asyncio.sleep")
584 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
587 mock_is_leader_from_status
,
589 mock_get_action_status
,
590 mock_get_action_output
,
592 mock__get_application
,
593 mock_disconnect_controller
,
594 mock_disconnect_model
,
598 mock_get_model
.return_value
= juju
.model
.Model()
599 mock__get_application
.return_value
= FakeApplication()
600 mock_is_leader_from_status
.return_value
= False
603 with self
.assertRaises(JujuLeaderUnitNotFound
):
604 output
, status
= self
.loop
.run_until_complete(
605 self
.libjuju
.execute_action("app", "model", "action",)
607 self
.assertIsNone(output
)
608 self
.assertIsNone(status
)
610 mock_disconnect_controller
.assert_called()
611 mock_disconnect_model
.assert_called()
613 def test_succesful_exec(
615 mock_get_action_status
,
616 mock_get_action_output
,
618 mock__get_application
,
619 mock_disconnect_controller
,
620 mock_disconnect_model
,
624 mock_get_model
.return_value
= juju
.model
.Model()
625 mock__get_application
.return_value
= FakeApplication()
626 mock_get_action_output
.return_value
= "output"
627 mock_get_action_status
.return_value
= {"id": "status"}
628 output
, status
= self
.loop
.run_until_complete(
629 self
.libjuju
.execute_action("app", "model", "existing_action")
631 self
.assertEqual(output
, "output")
632 self
.assertEqual(status
, "status")
634 mock_wait_for
.assert_called_once()
636 mock_disconnect_controller
.assert_called()
637 mock_disconnect_model
.assert_called()
640 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
641 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
642 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
643 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
644 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
645 class GetActionTest(LibjujuTestCase
):
647 super(GetActionTest
, self
).setUp()
651 mock_get_application
,
652 mock_disconnect_controller
,
653 mock_disconnect_model
,
657 mock_get_application
.side_effect
= Exception()
659 with self
.assertRaises(Exception):
660 actions
= self
.loop
.run_until_complete(
661 self
.libjuju
.get_actions("app", "model")
664 self
.assertIsNone(actions
)
665 mock_disconnect_controller
.assert_called_once()
666 mock_disconnect_model
.assert_called_once()
670 mock_get_application
,
671 mock_disconnect_controller
,
672 mock_disconnect_model
,
676 mock_get_application
.return_value
= FakeApplication()
678 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
680 self
.assertEqual(actions
, ["existing_action"])
682 mock_get_controller
.assert_called_once()
683 mock_get_model
.assert_called_once()
684 mock_disconnect_controller
.assert_called_once()
685 mock_disconnect_model
.assert_called_once()
688 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
689 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
690 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
691 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
692 @asynctest.mock
.patch("juju.application.Application.get_metrics")
693 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
694 class GetMetricsTest(LibjujuTestCase
):
696 super(GetMetricsTest
, self
).setUp()
698 def test_get_metrics_success(
700 mock_get_application
,
702 mock_disconnect_controller
,
703 mock_disconnect_model
,
707 mock_get_application
.return_value
= FakeApplication()
708 mock_get_model
.return_value
= juju
.model
.Model()
710 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
712 mock_disconnect_controller
.assert_called_once()
713 mock_disconnect_model
.assert_called_once()
715 def test_get_metrics_exception(
717 mock_get_application
,
719 mock_disconnect_controller
,
720 mock_disconnect_model
,
724 mock_get_model
.return_value
= juju
.model
.Model()
725 mock_get_metrics
.side_effect
= Exception()
726 with self
.assertRaises(Exception):
727 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
729 mock_disconnect_controller
.assert_called_once()
730 mock_disconnect_model
.assert_called_once()
732 def test_missing_args_exception(
734 mock_get_application
,
736 mock_disconnect_controller
,
737 mock_disconnect_model
,
741 mock_get_model
.return_value
= juju
.model
.Model()
743 with self
.assertRaises(Exception):
744 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
746 mock_get_controller
.assert_not_called()
747 mock_get_model
.assert_not_called()
748 mock_disconnect_controller
.assert_not_called()
749 mock_disconnect_model
.assert_not_called()
752 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
753 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
754 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
755 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
756 @asynctest.mock
.patch("juju.model.Model.add_relation")
757 class AddRelationTest(LibjujuTestCase
):
759 super(AddRelationTest
, self
).setUp()
761 @asynctest.mock
.patch("logging.Logger.warning")
766 mock_disconnect_controller
,
767 mock_disconnect_model
,
771 # TODO in libjuju.py should this fail only with a log message?
772 result
= {"error": "not found", "response": "response", "request-id": 1}
774 mock_get_model
.return_value
= juju
.model
.Model()
775 mock_add_relation
.side_effect
= JujuAPIError(result
)
777 self
.loop
.run_until_complete(
778 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
781 mock_warning
.assert_called_with("Relation not found: not found")
782 mock_disconnect_controller
.assert_called_once()
783 mock_disconnect_model
.assert_called_once()
785 @asynctest.mock
.patch("logging.Logger.warning")
786 def test_already_exists(
790 mock_disconnect_controller
,
791 mock_disconnect_model
,
795 # TODO in libjuju.py should this fail silently?
796 result
= {"error": "already exists", "response": "response", "request-id": 1}
798 mock_get_model
.return_value
= juju
.model
.Model()
799 mock_add_relation
.side_effect
= JujuAPIError(result
)
801 self
.loop
.run_until_complete(
802 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
805 mock_warning
.assert_called_with("Relation already exists: already exists")
806 mock_disconnect_controller
.assert_called_once()
807 mock_disconnect_model
.assert_called_once()
812 mock_disconnect_controller
,
813 mock_disconnect_model
,
817 mock_get_model
.return_value
= juju
.model
.Model()
818 result
= {"error": "", "response": "response", "request-id": 1}
819 mock_add_relation
.side_effect
= JujuAPIError(result
)
821 with self
.assertRaises(JujuAPIError
):
822 self
.loop
.run_until_complete(
823 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
826 mock_disconnect_controller
.assert_called_once()
827 mock_disconnect_model
.assert_called_once()
832 mock_disconnect_controller
,
833 mock_disconnect_model
,
837 mock_get_model
.return_value
= juju
.model
.Model()
839 self
.loop
.run_until_complete(
840 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
843 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
844 mock_disconnect_controller
.assert_called_once()
845 mock_disconnect_model
.assert_called_once()
850 mock_disconnect_controller
,
851 mock_disconnect_model
,
855 mock_get_model
.return_value
= juju
.model
.Model()
857 self
.loop
.run_until_complete(
858 self
.libjuju
.add_relation("model", "app1:relation1", "saas_name",)
861 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
862 mock_disconnect_controller
.assert_called_once()
863 mock_disconnect_model
.assert_called_once()
866 # TODO destroy_model testcase
869 # @asynctest.mock.patch("juju.model.Model.get_machines")
870 # @asynctest.mock.patch("logging.Logger.debug")
871 # class DestroyMachineTest(LibjujuTestCase):
873 # super(DestroyMachineTest, self).setUp()
875 # def test_success_manual_machine(
876 # self, mock_debug, mock_get_machines,
878 # mock_get_machines.side_effect = [
879 # {"machine": FakeManualMachine()},
880 # {"machine": FakeManualMachine()},
883 # self.loop.run_until_complete(
884 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
887 # asynctest.call("Waiting for machine machine is destroyed"),
888 # asynctest.call("Machine destroyed: machine"),
890 # mock_debug.assert_has_calls(calls)
892 # def test_no_machine(
893 # self, mock_debug, mock_get_machines,
895 # mock_get_machines.return_value = {}
896 # self.loop.run_until_complete(
897 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
899 # mock_debug.assert_called_with("Machine not found: machine")
902 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
903 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
904 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
905 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
906 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
907 class ConfigureApplicationTest(LibjujuTestCase
):
909 super(ConfigureApplicationTest
, self
).setUp()
913 mock_get_application
,
914 mock_disconnect_controller
,
915 mock_disconnect_model
,
920 mock_get_application
.return_value
= FakeApplication()
922 self
.loop
.run_until_complete(
923 self
.libjuju
.configure_application("model", "app", {"config"},)
925 mock_get_application
.assert_called_once()
926 mock_disconnect_controller
.assert_called_once()
927 mock_disconnect_model
.assert_called_once()
931 mock_get_application
,
932 mock_disconnect_controller
,
933 mock_disconnect_model
,
938 mock_get_application
.side_effect
= Exception()
940 with self
.assertRaises(Exception):
941 self
.loop
.run_until_complete(
942 self
.libjuju
.configure_application("model", "app", {"config"},)
944 mock_disconnect_controller
.assert_called_once()
945 mock_disconnect_model
.assert_called_once()
948 # TODO _get_api_endpoints_db test case
949 # TODO _update_api_endpoints_db test case
950 # TODO healthcheck test case
953 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
954 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
955 @asynctest.mock
.patch("juju.controller.Controller.list_models")
956 class ListModelsTest(LibjujuTestCase
):
958 super(ListModelsTest
, self
).setUp()
961 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
963 mock_get_controller
.return_value
= juju
.controller
.Controller()
964 mock_list_models
.return_value
= ["existingmodel"]
965 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
967 mock_disconnect_controller
.assert_called_once()
968 self
.assertEquals(models
, ["existingmodel"])
970 def test_not_containing(
971 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
973 mock_get_controller
.return_value
= juju
.controller
.Controller()
974 mock_list_models
.return_value
= ["existingmodel", "model"]
975 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
977 mock_disconnect_controller
.assert_called_once()
978 self
.assertEquals(models
, [])
980 def test_no_contains_arg(
981 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
983 mock_get_controller
.return_value
= juju
.controller
.Controller()
984 mock_list_models
.return_value
= ["existingmodel", "model"]
985 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
987 mock_disconnect_controller
.assert_called_once()
988 self
.assertEquals(models
, ["existingmodel", "model"])
991 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
992 class ModelsExistTest(LibjujuTestCase
):
994 super(ModelsExistTest
, self
).setUp()
996 def test_model_names_none(self
, mock_list_models
):
997 mock_list_models
.return_value
= []
998 with self
.assertRaises(Exception):
999 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1001 def test_model_names_empty(self
, mock_list_models
):
1002 mock_list_models
.return_value
= []
1003 with self
.assertRaises(Exception):
1004 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1005 self
.libjuju
.models_exist([])
1008 def test_model_names_not_existing(self
, mock_list_models
):
1009 mock_list_models
.return_value
= ["prometheus", "grafana"]
1010 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1011 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1013 self
.assertFalse(exist
)
1014 self
.assertEqual(non_existing_models
, ["prometheus2"])
1016 def test_model_names_exist(self
, mock_list_models
):
1017 mock_list_models
.return_value
= ["prometheus", "grafana"]
1018 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1019 self
.libjuju
.models_exist(["prometheus", "grafana"])
1021 self
.assertTrue(exist
)
1022 self
.assertEqual(non_existing_models
, [])
1025 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1026 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1027 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1028 class ListOffers(LibjujuTestCase
):
1030 super(ListOffers
, self
).setUp()
1032 def test_disconnect_controller(
1033 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1035 mock_get_controller
.return_value
= juju
.controller
.Controller()
1036 mock_list_offers
.side_effect
= Exception()
1037 with self
.assertRaises(Exception):
1038 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1039 mock_disconnect_controller
.assert_called_once()
1041 def test_empty_list(
1042 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1044 mock_get_controller
.return_value
= juju
.controller
.Controller()
1045 mock_list_offers
.return_value
= []
1046 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1047 self
.assertEqual(offers
, [])
1048 mock_disconnect_controller
.assert_called_once()
1050 def test_non_empty_list(
1051 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1053 mock_get_controller
.return_value
= juju
.controller
.Controller()
1054 mock_list_offers
.return_value
= ["offer"]
1055 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1056 self
.assertEqual(offers
, ["offer"])
1057 mock_disconnect_controller
.assert_called_once()
1060 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1061 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1062 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1063 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1064 @asynctest.mock
.patch("juju.model.Model.consume")
1065 class ConsumeTest(LibjujuTestCase
):
1067 super(ConsumeTest
, self
).setUp()
1072 mock_disconnect_controller
,
1073 mock_disconnect_model
,
1075 mock_get_controller
,
1077 mock_get_controller
.return_value
= juju
.controller
.Controller()
1078 mock_get_model
.return_value
= juju
.model
.Model()
1080 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1081 mock_consume
.assert_called_once()
1082 mock_disconnect_model
.assert_called_once()
1083 mock_disconnect_controller
.assert_called_once()
1085 def test_parsing_error_exception(
1088 mock_disconnect_controller
,
1089 mock_disconnect_model
,
1091 mock_get_controller
,
1093 mock_get_controller
.return_value
= juju
.controller
.Controller()
1094 mock_get_model
.return_value
= juju
.model
.Model()
1095 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1097 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1098 self
.loop
.run_until_complete(
1099 self
.libjuju
.consume("offer_url", "model_name")
1101 mock_consume
.assert_called_once()
1102 mock_disconnect_model
.assert_called_once()
1103 mock_disconnect_controller
.assert_called_once()
1105 def test_juju_error_exception(
1108 mock_disconnect_controller
,
1109 mock_disconnect_model
,
1111 mock_get_controller
,
1113 mock_get_controller
.return_value
= juju
.controller
.Controller()
1114 mock_get_model
.return_value
= juju
.model
.Model()
1115 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1117 with self
.assertRaises(juju
.errors
.JujuError
):
1118 self
.loop
.run_until_complete(
1119 self
.libjuju
.consume("offer_url", "model_name")
1121 mock_consume
.assert_called_once()
1122 mock_disconnect_model
.assert_called_once()
1123 mock_disconnect_controller
.assert_called_once()
1125 def test_juju_api_error_exception(
1128 mock_disconnect_controller
,
1129 mock_disconnect_model
,
1131 mock_get_controller
,
1133 mock_get_controller
.return_value
= juju
.controller
.Controller()
1134 mock_get_model
.return_value
= juju
.model
.Model()
1135 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1136 {"error": "", "response": "", "request-id": ""}
1139 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1140 self
.loop
.run_until_complete(
1141 self
.libjuju
.consume("offer_url", "model_name")
1143 mock_consume
.assert_called_once()
1144 mock_disconnect_model
.assert_called_once()
1145 mock_disconnect_controller
.assert_called_once()
1148 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1149 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1150 class AddK8sTest(LibjujuTestCase
):
1152 super(AddK8sTest
, self
).setUp()
1153 self
.configuration
= kubernetes
.client
.configuration
.Configuration()
1155 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1156 self
.loop
.run_until_complete(
1157 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1159 mock_add_cloud
.assert_called_once()
1160 mock_get_k8s_cloud_credential
.assert_called_once()
1162 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1163 mock_add_cloud
.side_effect
= Exception()
1164 with self
.assertRaises(Exception):
1165 self
.loop
.run_until_complete(
1166 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1168 mock_add_cloud
.assert_called_once()
1169 mock_get_k8s_cloud_credential
.assert_called_once()
1171 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1172 with self
.assertRaises(Exception):
1173 self
.loop
.run_until_complete(
1174 self
.libjuju
.add_k8s("", self
.configuration
, "storage_class")
1176 mock_add_cloud
.assert_not_called()
1178 def test_add_k8s_missing_storage_name(
1179 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1181 with self
.assertRaises(Exception):
1182 self
.loop
.run_until_complete(
1183 self
.libjuju
.add_k8s("cloud", self
.configuration
, "")
1185 mock_add_cloud
.assert_not_called()
1187 def test_add_k8s_missing_configuration_keys(
1188 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1190 with self
.assertRaises(Exception):
1191 self
.loop
.run_until_complete(self
.libjuju
.add_k8s("cloud", None, ""))
1192 mock_add_cloud
.assert_not_called()
1195 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1196 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1197 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1198 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1199 class AddCloudTest(LibjujuTestCase
):
1201 super(AddCloudTest
, self
).setUp()
1202 self
.cloud
= juju
.client
.client
.Cloud()
1203 self
.credential
= juju
.client
.client
.CloudCredential()
1205 def test_add_cloud_with_credential(
1207 mock_add_credential
,
1209 mock_disconnect_controller
,
1210 mock_get_controller
,
1212 mock_get_controller
.return_value
= juju
.controller
.Controller()
1214 cloud
= self
.loop
.run_until_complete(
1215 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1217 self
.assertEqual(cloud
, self
.cloud
)
1218 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1219 mock_add_credential
.assert_called_once_with(
1220 "cloud", credential
=self
.credential
, cloud
="cloud"
1222 mock_disconnect_controller
.assert_called_once()
1224 def test_add_cloud_no_credential(
1226 mock_add_credential
,
1228 mock_disconnect_controller
,
1229 mock_get_controller
,
1231 mock_get_controller
.return_value
= juju
.controller
.Controller()
1233 cloud
= self
.loop
.run_until_complete(
1234 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1236 self
.assertEqual(cloud
, self
.cloud
)
1237 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1238 mock_add_credential
.assert_not_called()
1239 mock_disconnect_controller
.assert_called_once()
1241 def test_add_cloud_exception(
1243 mock_add_credential
,
1245 mock_disconnect_controller
,
1246 mock_get_controller
,
1248 mock_get_controller
.return_value
= juju
.controller
.Controller()
1249 mock_add_cloud
.side_effect
= Exception()
1250 with self
.assertRaises(Exception):
1251 self
.loop
.run_until_complete(
1252 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1255 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1256 mock_add_credential
.assert_not_called()
1257 mock_disconnect_controller
.assert_called_once()
1259 def test_add_credential_exception(
1261 mock_add_credential
,
1263 mock_disconnect_controller
,
1264 mock_get_controller
,
1266 mock_get_controller
.return_value
= juju
.controller
.Controller()
1267 mock_add_credential
.side_effect
= Exception()
1268 with self
.assertRaises(Exception):
1269 self
.loop
.run_until_complete(
1270 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1273 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1274 mock_add_credential
.assert_called_once_with(
1275 "cloud", credential
=self
.credential
, cloud
="cloud"
1277 mock_disconnect_controller
.assert_called_once()
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.remove_cloud")
1283 class RemoveCloudTest(LibjujuTestCase
):
1285 super(RemoveCloudTest
, self
).setUp()
1287 def test_remove_cloud(
1288 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1290 mock_get_controller
.return_value
= juju
.controller
.Controller()
1292 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1293 mock_remove_cloud
.assert_called_once_with("cloud")
1294 mock_disconnect_controller
.assert_called_once()
1296 def test_remove_cloud_exception(
1297 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1299 mock_get_controller
.return_value
= juju
.controller
.Controller()
1300 mock_remove_cloud
.side_effect
= Exception()
1302 with self
.assertRaises(Exception):
1303 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1304 mock_remove_cloud
.assert_called_once_with("cloud")
1305 mock_disconnect_controller
.assert_called_once()
1308 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1309 class GetK8sCloudCredentials(LibjujuTestCase
):
1311 super(GetK8sCloudCredentials
, self
).setUp()
1313 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1314 def test_not_supported(self
, mock_exception
, mock_configuration
):
1315 mock_configuration
.username
= ""
1316 mock_configuration
.password
= ""
1317 mock_configuration
.ssl_ca_cert
= None
1318 mock_configuration
.cert_file
= None
1319 mock_configuration
.key_file
= None
1320 exception_raised
= False
1322 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1323 except JujuInvalidK8sConfiguration
as e
:
1324 exception_raised
= True
1326 e
.message
, "authentication method not supported",
1328 self
.assertTrue(exception_raised
)
1330 def test_user_pass(self
, mock_configuration
):
1331 mock_configuration
.username
= "admin"
1332 mock_configuration
.password
= "admin"
1333 mock_configuration
.ssl_ca_cert
= None
1334 mock_configuration
.cert_file
= None
1335 mock_configuration
.key_file
= None
1336 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1339 juju
.client
._definitions
.CloudCredential(
1340 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1344 def test_user_no_pass(self
, mock_configuration
):
1345 mock_configuration
.username
= "admin"
1346 mock_configuration
.password
= ""
1347 mock_configuration
.ssl_ca_cert
= None
1348 mock_configuration
.cert_file
= None
1349 mock_configuration
.key_file
= None
1350 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1351 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1354 juju
.client
._definitions
.CloudCredential(
1355 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1358 mock_debug
.assert_called_once_with(
1359 "credential for user admin has empty password"
1362 def test_user_pass_with_cert(self
, mock_configuration
):
1363 mock_configuration
.username
= "admin"
1364 mock_configuration
.password
= "admin"
1365 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1366 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1367 ssl_ca_cert_file
.write("cacert")
1368 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1369 mock_configuration
.cert_file
= None
1370 mock_configuration
.key_file
= None
1371 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1374 juju
.client
._definitions
.CloudCredential(
1376 "username": "admin",
1377 "password": "admin",
1378 "ClientCertificateData": "cacert",
1380 auth_type
="userpasswithcert",
1384 def test_cert(self
, mock_configuration
):
1385 mock_configuration
.username
= ""
1386 mock_configuration
.password
= ""
1387 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1388 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1389 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1390 ssl_ca_cert_file
.write("cacert")
1391 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1392 mock_configuration
.cert_file
= None
1393 mock_configuration
.key_file
= None
1394 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1397 juju
.client
._definitions
.CloudCredential(
1398 attrs
={"ClientCertificateData": "cacert", "Token": "Token"},
1399 auth_type
="certificate",
1403 def test_oauth2(self
, mock_configuration
):
1404 mock_configuration
.username
= ""
1405 mock_configuration
.password
= ""
1406 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1407 key
= tempfile
.NamedTemporaryFile()
1408 with
open(key
.name
, "w") as key_file
:
1409 key_file
.write("key")
1410 mock_configuration
.ssl_ca_cert
= None
1411 mock_configuration
.cert_file
= None
1412 mock_configuration
.key_file
= key
.name
1413 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1416 juju
.client
._definitions
.CloudCredential(
1417 attrs
={"ClientKeyData": "key", "Token": "Token"}, auth_type
="oauth2",
1421 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1422 def test_oauth2_missing_token(self
, mock_exception
, mock_configuration
):
1423 mock_configuration
.username
= ""
1424 mock_configuration
.password
= ""
1425 key
= tempfile
.NamedTemporaryFile()
1426 with
open(key
.name
, "w") as key_file
:
1427 key_file
.write("key")
1428 mock_configuration
.ssl_ca_cert
= None
1429 mock_configuration
.cert_file
= None
1430 mock_configuration
.key_file
= key
.name
1431 exception_raised
= False
1433 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1434 except JujuInvalidK8sConfiguration
as e
:
1435 exception_raised
= True
1437 e
.message
, "missing token for auth type oauth2",
1439 self
.assertTrue(exception_raised
)
1441 def test_unknown_api_key(self
, mock_configuration
):
1442 mock_configuration
.username
= ""
1443 mock_configuration
.password
= ""
1444 mock_configuration
.api_key
= {"authorization": "Bearer Token Wrong"}
1445 mock_configuration
.ssl_ca_cert
= None
1446 mock_configuration
.cert_file
= None
1447 mock_configuration
.key_file
= None
1448 exception_raised
= False
1450 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1451 except JujuInvalidK8sConfiguration
as e
:
1452 exception_raised
= True
1454 e
.message
, "unknown format of api_key",
1456 self
.assertTrue(exception_raised
)
1458 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1459 mock_configuration
.username
= "admin"
1460 mock_configuration
.password
= "pass"
1461 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1462 mock_configuration
.ssl_ca_cert
= None
1463 mock_configuration
.cert_file
= None
1464 mock_configuration
.key_file
= None
1465 exception_raised
= False
1467 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1468 except JujuInvalidK8sConfiguration
as e
:
1469 exception_raised
= True
1471 e
.message
, "Cannot set both token and user/pass",
1473 self
.assertTrue(exception_raised
)