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
,
36 class LibjujuTestCase(asynctest
.TestCase
):
37 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
38 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
39 @asynctest.mock
.patch("juju.controller.Controller.connection")
40 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
43 mock__get_api_endpoints_db
=None,
46 mock_update_endpoints
=None,
48 loop
= asyncio
.get_event_loop()
50 mock__get_api_endpoints_db
.return_value
= ["127.0.0.1:17070"]
51 endpoints
= "127.0.0.1:17070"
55 -----BEGIN CERTIFICATE-----
57 -----END CERTIFICATE-----"""
58 self
.libjuju
= Libjuju(
60 "192.168.0.155:17070",
68 apt_mirror
="192.168.0.100",
69 enable_os_upgrade
=True,
71 logging
.disable(logging
.CRITICAL
)
72 loop
.run_until_complete(self
.libjuju
.disconnect())
75 @asynctest.mock
.patch("juju.controller.Controller.connect")
76 @asynctest.mock
.patch(
77 "juju.controller.Controller.api_endpoints",
78 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
80 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
81 class GetControllerTest(LibjujuTestCase
):
83 super(GetControllerTest
, self
).setUp()
85 def test_diff_endpoint(
86 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
88 self
.libjuju
.endpoints
= []
89 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
90 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
91 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
93 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
96 mock_disconnect_controller
,
97 mock__update_api_endpoints_db
,
101 self
.libjuju
.endpoints
= []
102 mock__update_api_endpoints_db
.side_effect
= Exception()
104 with self
.assertRaises(JujuControllerFailedConnecting
):
105 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
106 self
.assertIsNone(controller
)
107 mock_disconnect_controller
.assert_called_once()
109 def test_same_endpoint_get_controller(
110 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
112 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
113 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
114 mock__update_api_endpoints_db
.assert_not_called()
115 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
118 class DisconnectTest(LibjujuTestCase
):
120 super(DisconnectTest
, self
).setUp()
122 @asynctest.mock
.patch("juju.model.Model.disconnect")
123 def test_disconnect_model(self
, mock_disconnect
):
124 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
125 mock_disconnect
.assert_called_once()
127 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
128 def test_disconnect_controller(self
, mock_disconnect
):
129 self
.loop
.run_until_complete(
130 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
132 mock_disconnect
.assert_called_once()
135 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
136 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
137 @asynctest.mock
.patch("juju.controller.Controller.add_model")
138 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
139 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
140 class AddModelTest(LibjujuTestCase
):
142 super(AddModelTest
, self
).setUp()
144 def test_existing_model(
146 mock_disconnect_model
,
147 mock_disconnect_controller
,
152 mock_model_exists
.return_value
= True
154 with self
.assertRaises(JujuModelAlreadyExists
):
155 self
.loop
.run_until_complete(
156 self
.libjuju
.add_model("existing_model", "cloud")
159 mock_disconnect_controller
.assert_called()
161 # TODO Check two job executing at the same time and one returning without doing anything.
163 def test_non_existing_model(
165 mock_disconnect_model
,
166 mock_disconnect_controller
,
171 mock_model_exists
.return_value
= False
172 mock_get_controller
.return_value
= juju
.controller
.Controller()
174 self
.loop
.run_until_complete(
175 self
.libjuju
.add_model("nonexisting_model", "cloud")
178 mock_add_model
.assert_called_once()
179 mock_disconnect_controller
.assert_called()
180 mock_disconnect_model
.assert_called()
183 @asynctest.mock
.patch("juju.controller.Controller.get_model")
184 class GetModelTest(LibjujuTestCase
):
186 super(GetModelTest
, self
).setUp()
189 self
, mock_get_model
,
191 mock_get_model
.return_value
= juju
.model
.Model()
192 model
= self
.loop
.run_until_complete(
193 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
195 self
.assertIsInstance(model
, juju
.model
.Model
)
198 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
199 @asynctest.mock
.patch("juju.controller.Controller.list_models")
200 class ModelExistsTest(LibjujuTestCase
):
202 super(ModelExistsTest
, self
).setUp()
204 async def test_existing_model(
205 self
, mock_list_models
, mock_get_controller
,
207 mock_list_models
.return_value
= ["existing_model"]
209 await self
.libjuju
.model_exists(
210 "existing_model", juju
.controller
.Controller()
214 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
215 async def test_no_controller(
216 self
, mock_disconnect_controller
, mock_list_models
, mock_get_controller
,
218 mock_list_models
.return_value
= ["existing_model"]
219 mock_get_controller
.return_value
= juju
.controller
.Controller()
220 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
221 mock_disconnect_controller
.assert_called_once()
223 async def test_non_existing_model(
224 self
, mock_list_models
, mock_get_controller
,
226 mock_list_models
.return_value
= ["existing_model"]
228 await self
.libjuju
.model_exists(
229 "not_existing_model", juju
.controller
.Controller()
234 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
235 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
236 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
237 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
238 @asynctest.mock
.patch("juju.model.Model.get_status")
239 class GetModelStatusTest(LibjujuTestCase
):
241 super(GetModelStatusTest
, self
).setUp()
246 mock_disconnect_controller
,
247 mock_disconnect_model
,
251 mock_get_model
.return_value
= juju
.model
.Model()
252 mock_get_status
.return_value
= {"status"}
254 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
256 mock_get_status
.assert_called_once()
257 mock_disconnect_controller
.assert_called_once()
258 mock_disconnect_model
.assert_called_once()
260 self
.assertEqual(status
, {"status"})
265 mock_disconnect_controller
,
266 mock_disconnect_model
,
270 mock_get_model
.return_value
= juju
.model
.Model()
271 mock_get_status
.side_effect
= Exception()
273 with self
.assertRaises(Exception):
274 status
= self
.loop
.run_until_complete(
275 self
.libjuju
.get_model_status("model")
278 mock_disconnect_controller
.assert_called_once()
279 mock_disconnect_model
.assert_called_once()
281 self
.assertIsNone(status
)
284 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
285 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
286 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
287 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
288 @asynctest.mock
.patch("juju.model.Model.get_machines")
289 @asynctest.mock
.patch("juju.model.Model.add_machine")
290 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
291 class CreateMachineTest(LibjujuTestCase
):
293 super(CreateMachineTest
, self
).setUp()
295 def test_existing_machine(
300 mock_disconnect_controller
,
301 mock_disconnect_model
,
305 mock_get_model
.return_value
= juju
.model
.Model()
306 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
307 machine
, bool_res
= self
.loop
.run_until_complete(
308 self
.libjuju
.create_machine("model", "existing_machine")
311 self
.assertIsInstance(machine
, FakeMachine
)
312 self
.assertFalse(bool_res
)
314 mock_disconnect_controller
.assert_called()
315 mock_disconnect_model
.assert_called()
317 def test_non_existing_machine(
322 mock_disconnect_controller
,
323 mock_disconnect_model
,
329 mock_get_model
.return_value
= juju
.model
.Model()
330 with self
.assertRaises(JujuMachineNotFound
):
331 machine
, bool_res
= self
.loop
.run_until_complete(
332 self
.libjuju
.create_machine("model", "non_existing_machine")
334 self
.assertIsNone(machine
)
335 self
.assertIsNone(bool_res
)
337 mock_disconnect_controller
.assert_called()
338 mock_disconnect_model
.assert_called()
345 mock_disconnect_controller
,
346 mock_disconnect_model
,
350 mock_get_model
.return_value
= juju
.model
.Model()
351 mock_add_machine
.return_value
= FakeMachine()
353 machine
, bool_res
= self
.loop
.run_until_complete(
354 self
.libjuju
.create_machine("model")
357 self
.assertIsInstance(machine
, FakeMachine
)
358 self
.assertTrue(bool_res
)
360 mock_wait_for
.assert_called_once()
361 mock_add_machine
.assert_called_once()
363 mock_disconnect_controller
.assert_called()
364 mock_disconnect_model
.assert_called()
367 # TODO test provision machine
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(
375 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
377 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
378 @asynctest.mock
.patch("juju.model.Model.deploy")
379 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
380 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
381 class DeployCharmTest(LibjujuTestCase
):
383 super(DeployCharmTest
, self
).setUp()
385 def test_existing_app(
392 mock_disconnect_controller
,
393 mock_disconnect_model
,
397 mock_get_model
.return_value
= juju
.model
.Model()
398 mock_applications
.return_value
= {"existing_app"}
401 with self
.assertRaises(JujuApplicationExists
):
402 application
= self
.loop
.run_until_complete(
403 self
.libjuju
.deploy_charm("existing_app", "path", "model", "machine",)
405 self
.assertIsNone(application
)
407 mock_disconnect_controller
.assert_called()
408 mock_disconnect_model
.assert_called()
410 def test_non_existing_machine(
417 mock_disconnect_controller
,
418 mock_disconnect_model
,
422 mock_get_model
.return_value
= juju
.model
.Model()
423 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
425 with self
.assertRaises(JujuMachineNotFound
):
426 application
= self
.loop
.run_until_complete(
427 self
.libjuju
.deploy_charm("app", "path", "model", "machine",)
430 self
.assertIsNone(application
)
432 mock_disconnect_controller
.assert_called()
433 mock_disconnect_model
.assert_called()
442 mock_disconnect_controller
,
443 mock_disconnect_model
,
447 mock_get_model
.return_value
= juju
.model
.Model()
448 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
449 mock_create_machine
.return_value
= (FakeMachine(), "other")
450 mock_deploy
.return_value
= FakeApplication()
451 application
= self
.loop
.run_until_complete(
452 self
.libjuju
.deploy_charm(
453 "app", "path", "model", "existing_machine", num_units
=2,
457 self
.assertIsInstance(application
, FakeApplication
)
459 mock_deploy
.assert_called_once()
460 mock_wait_for
.assert_called_once()
462 mock_create_machine
.assert_called_once()
464 mock_disconnect_controller
.assert_called()
465 mock_disconnect_model
.assert_called()
474 mock_disconnect_controller
,
475 mock_disconnect_model
,
479 mock_get_model
.return_value
= juju
.model
.Model()
480 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
481 mock_deploy
.return_value
= FakeApplication()
482 application
= self
.loop
.run_until_complete(
483 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
486 self
.assertIsInstance(application
, FakeApplication
)
488 mock_deploy
.assert_called_once()
489 mock_wait_for
.assert_called_once()
491 mock_disconnect_controller
.assert_called()
492 mock_disconnect_model
.assert_called()
495 @asynctest.mock
.patch(
496 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
498 class GetApplicationTest(LibjujuTestCase
):
500 super(GetApplicationTest
, self
).setUp()
502 def test_existing_application(
503 self
, mock_applications
,
505 mock_applications
.return_value
= {"existing_app": "exists"}
506 model
= juju
.model
.Model()
507 result
= self
.libjuju
._get
_application
(model
, "existing_app")
508 self
.assertEqual(result
, "exists")
510 def test_non_existing_application(
511 self
, mock_applications
,
513 mock_applications
.return_value
= {"existing_app": "exists"}
514 model
= juju
.model
.Model()
515 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
516 self
.assertIsNone(result
)
519 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
520 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
521 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
522 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
523 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
524 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
525 @asynctest.mock
.patch("juju.model.Model.get_action_output")
526 @asynctest.mock
.patch("juju.model.Model.get_action_status")
527 class ExecuteActionTest(LibjujuTestCase
):
529 super(ExecuteActionTest
, self
).setUp()
531 def test_no_application(
533 mock_get_action_status
,
534 mock_get_action_output
,
536 mock__get_application
,
537 mock_disconnect_controller
,
538 mock_disconnect_model
,
542 mock__get_application
.return_value
= None
543 mock_get_model
.return_value
= juju
.model
.Model()
546 with self
.assertRaises(JujuApplicationNotFound
):
547 output
, status
= self
.loop
.run_until_complete(
548 self
.libjuju
.execute_action("app", "model", "action",)
550 self
.assertIsNone(output
)
551 self
.assertIsNone(status
)
553 mock_disconnect_controller
.assert_called()
554 mock_disconnect_model
.assert_called()
558 mock_get_action_status
,
559 mock_get_action_output
,
561 mock__get_application
,
562 mock_disconnect_controller
,
563 mock_disconnect_model
,
568 mock_get_model
.return_value
= juju
.model
.Model()
569 mock__get_application
.return_value
= FakeApplication()
572 with self
.assertRaises(JujuActionNotFound
):
573 output
, status
= self
.loop
.run_until_complete(
574 self
.libjuju
.execute_action("app", "model", "action",)
576 self
.assertIsNone(output
)
577 self
.assertIsNone(status
)
579 mock_disconnect_controller
.assert_called()
580 mock_disconnect_model
.assert_called()
582 # TODO no leader unit found exception
584 def test_succesful_exec(
586 mock_get_action_status
,
587 mock_get_action_output
,
589 mock__get_application
,
590 mock_disconnect_controller
,
591 mock_disconnect_model
,
595 mock_get_model
.return_value
= juju
.model
.Model()
596 mock__get_application
.return_value
= FakeApplication()
597 mock_get_action_output
.return_value
= "output"
598 mock_get_action_status
.return_value
= {"id": "status"}
599 output
, status
= self
.loop
.run_until_complete(
600 self
.libjuju
.execute_action("app", "model", "existing_action")
602 self
.assertEqual(output
, "output")
603 self
.assertEqual(status
, "status")
605 mock_wait_for
.assert_called_once()
607 mock_disconnect_controller
.assert_called()
608 mock_disconnect_model
.assert_called()
611 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
612 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
613 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
614 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
615 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
616 class GetActionTest(LibjujuTestCase
):
618 super(GetActionTest
, self
).setUp()
622 mock_get_application
,
623 mock_disconnect_controller
,
624 mock_disconnect_model
,
628 mock_get_application
.side_effect
= Exception()
630 with self
.assertRaises(Exception):
631 actions
= self
.loop
.run_until_complete(
632 self
.libjuju
.get_actions("app", "model")
635 self
.assertIsNone(actions
)
636 mock_disconnect_controller
.assert_called_once()
637 mock_disconnect_model
.assert_called_once()
641 mock_get_application
,
642 mock_disconnect_controller
,
643 mock_disconnect_model
,
647 mock_get_application
.return_value
= FakeApplication()
649 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
651 self
.assertEqual(actions
, ["existing_action"])
653 mock_get_controller
.assert_called_once()
654 mock_get_model
.assert_called_once()
655 mock_disconnect_controller
.assert_called_once()
656 mock_disconnect_model
.assert_called_once()
659 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
660 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
661 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
662 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
663 @asynctest.mock
.patch("juju.application.Application.get_metrics")
664 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
665 class GetMetricsTest(LibjujuTestCase
):
667 super(GetMetricsTest
, self
).setUp()
669 def test_get_metrics_success(
671 mock_get_application
,
673 mock_disconnect_controller
,
674 mock_disconnect_model
,
678 mock_get_application
.return_value
= FakeApplication()
679 mock_get_model
.return_value
= juju
.model
.Model()
681 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
683 mock_disconnect_controller
.assert_called_once()
684 mock_disconnect_model
.assert_called_once()
686 def test_get_metrics_exception(
688 mock_get_application
,
690 mock_disconnect_controller
,
691 mock_disconnect_model
,
695 mock_get_model
.return_value
= juju
.model
.Model()
696 mock_get_metrics
.side_effect
= Exception()
697 with self
.assertRaises(Exception):
698 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
700 mock_disconnect_controller
.assert_called_once()
701 mock_disconnect_model
.assert_called_once()
703 def test_missing_args_exception(
705 mock_get_application
,
707 mock_disconnect_controller
,
708 mock_disconnect_model
,
712 mock_get_model
.return_value
= juju
.model
.Model()
714 with self
.assertRaises(Exception):
715 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
717 mock_get_controller
.assert_not_called()
718 mock_get_model
.assert_not_called()
719 mock_disconnect_controller
.assert_not_called()
720 mock_disconnect_model
.assert_not_called()
723 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
724 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
725 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
726 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
727 @asynctest.mock
.patch("juju.model.Model.add_relation")
728 class AddRelationTest(LibjujuTestCase
):
730 super(AddRelationTest
, self
).setUp()
732 @asynctest.mock
.patch("logging.Logger.warning")
737 mock_disconnect_controller
,
738 mock_disconnect_model
,
742 # TODO in libjuju.py should this fail only with a log message?
743 result
= {"error": "not found", "response": "response", "request-id": 1}
745 mock_get_model
.return_value
= juju
.model
.Model()
746 mock_add_relation
.side_effect
= JujuAPIError(result
)
748 self
.loop
.run_until_complete(
749 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
752 mock_warning
.assert_called_with("Relation not found: not found")
753 mock_disconnect_controller
.assert_called_once()
754 mock_disconnect_model
.assert_called_once()
756 @asynctest.mock
.patch("logging.Logger.warning")
757 def test_already_exists(
761 mock_disconnect_controller
,
762 mock_disconnect_model
,
766 # TODO in libjuju.py should this fail silently?
767 result
= {"error": "already exists", "response": "response", "request-id": 1}
769 mock_get_model
.return_value
= juju
.model
.Model()
770 mock_add_relation
.side_effect
= JujuAPIError(result
)
772 self
.loop
.run_until_complete(
773 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
776 mock_warning
.assert_called_with("Relation already exists: already exists")
777 mock_disconnect_controller
.assert_called_once()
778 mock_disconnect_model
.assert_called_once()
783 mock_disconnect_controller
,
784 mock_disconnect_model
,
788 mock_get_model
.return_value
= juju
.model
.Model()
789 result
= {"error": "", "response": "response", "request-id": 1}
790 mock_add_relation
.side_effect
= JujuAPIError(result
)
792 with self
.assertRaises(JujuAPIError
):
793 self
.loop
.run_until_complete(
794 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
797 mock_disconnect_controller
.assert_called_once()
798 mock_disconnect_model
.assert_called_once()
803 mock_disconnect_controller
,
804 mock_disconnect_model
,
808 mock_get_model
.return_value
= juju
.model
.Model()
810 self
.loop
.run_until_complete(
811 self
.libjuju
.add_relation("model", "app1:relation1", "app2:relation2",)
814 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
815 mock_disconnect_controller
.assert_called_once()
816 mock_disconnect_model
.assert_called_once()
821 mock_disconnect_controller
,
822 mock_disconnect_model
,
826 mock_get_model
.return_value
= juju
.model
.Model()
828 self
.loop
.run_until_complete(
829 self
.libjuju
.add_relation("model", "app1:relation1", "saas_name",)
832 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
833 mock_disconnect_controller
.assert_called_once()
834 mock_disconnect_model
.assert_called_once()
837 # TODO destroy_model testcase
840 @asynctest.mock
.patch("juju.model.Model.get_machines")
841 @asynctest.mock
.patch("logging.Logger.debug")
842 class DestroyMachineTest(LibjujuTestCase
):
844 super(DestroyMachineTest
, self
).setUp()
847 self
, mock_debug
, mock_get_machines
,
849 mock_get_machines
.side_effect
= [
850 {"machine": FakeMachine()},
851 {"machine": FakeMachine()},
854 self
.loop
.run_until_complete(
855 self
.libjuju
.destroy_machine(juju
.model
.Model(), "machine", 2,)
858 asynctest
.call("Waiting for machine machine is destroyed"),
859 asynctest
.call("Machine destroyed: machine"),
861 mock_debug
.assert_has_calls(calls
)
864 self
, mock_debug
, mock_get_machines
,
866 mock_get_machines
.return_value
= {}
867 self
.loop
.run_until_complete(
868 self
.libjuju
.destroy_machine(juju
.model
.Model(), "machine", 2,)
870 mock_debug
.assert_called_with("Machine not found: machine")
873 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
874 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
875 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
876 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
877 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
878 class ConfigureApplicationTest(LibjujuTestCase
):
880 super(ConfigureApplicationTest
, self
).setUp()
884 mock_get_application
,
885 mock_disconnect_controller
,
886 mock_disconnect_model
,
891 mock_get_application
.return_value
= FakeApplication()
893 self
.loop
.run_until_complete(
894 self
.libjuju
.configure_application("model", "app", {"config"},)
896 mock_get_application
.assert_called_once()
897 mock_disconnect_controller
.assert_called_once()
898 mock_disconnect_model
.assert_called_once()
902 mock_get_application
,
903 mock_disconnect_controller
,
904 mock_disconnect_model
,
909 mock_get_application
.side_effect
= Exception()
911 with self
.assertRaises(Exception):
912 self
.loop
.run_until_complete(
913 self
.libjuju
.configure_application("model", "app", {"config"},)
915 mock_disconnect_controller
.assert_called_once()
916 mock_disconnect_model
.assert_called_once()
919 # TODO _get_api_endpoints_db test case
920 # TODO _update_api_endpoints_db test case
921 # TODO healthcheck test case
924 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
925 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
926 @asynctest.mock
.patch("juju.controller.Controller.list_models")
927 class ListModelsTest(LibjujuTestCase
):
929 super(ListModelsTest
, self
).setUp()
932 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
934 mock_get_controller
.return_value
= juju
.controller
.Controller()
935 mock_list_models
.return_value
= ["existingmodel"]
936 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
938 mock_disconnect_controller
.assert_called_once()
939 self
.assertEquals(models
, ["existingmodel"])
941 def test_not_containing(
942 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
944 mock_get_controller
.return_value
= juju
.controller
.Controller()
945 mock_list_models
.return_value
= ["existingmodel", "model"]
946 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
948 mock_disconnect_controller
.assert_called_once()
949 self
.assertEquals(models
, [])
951 def test_no_contains_arg(
952 self
, mock_list_models
, mock_disconnect_controller
, mock_get_controller
,
954 mock_get_controller
.return_value
= juju
.controller
.Controller()
955 mock_list_models
.return_value
= ["existingmodel", "model"]
956 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
958 mock_disconnect_controller
.assert_called_once()
959 self
.assertEquals(models
, ["existingmodel", "model"])
962 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
963 class ModelsExistTest(LibjujuTestCase
):
965 super(ModelsExistTest
, self
).setUp()
967 def test_model_names_none(self
, mock_list_models
):
968 mock_list_models
.return_value
= []
969 with self
.assertRaises(Exception):
970 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
972 def test_model_names_empty(self
, mock_list_models
):
973 mock_list_models
.return_value
= []
974 with self
.assertRaises(Exception):
975 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
976 self
.libjuju
.models_exist([])
979 def test_model_names_not_existing(self
, mock_list_models
):
980 mock_list_models
.return_value
= ["prometheus", "grafana"]
981 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
982 self
.libjuju
.models_exist(["prometheus2", "grafana"])
984 self
.assertFalse(exist
)
985 self
.assertEqual(non_existing_models
, ["prometheus2"])
987 def test_model_names_exist(self
, mock_list_models
):
988 mock_list_models
.return_value
= ["prometheus", "grafana"]
989 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
990 self
.libjuju
.models_exist(["prometheus", "grafana"])
992 self
.assertTrue(exist
)
993 self
.assertEqual(non_existing_models
, [])
996 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
997 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
998 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
999 class ListOffers(LibjujuTestCase
):
1001 super(ListOffers
, self
).setUp()
1003 def test_disconnect_controller(
1004 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1006 mock_get_controller
.return_value
= juju
.controller
.Controller()
1007 mock_list_offers
.side_effect
= Exception()
1008 with self
.assertRaises(Exception):
1009 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1010 mock_disconnect_controller
.assert_called_once()
1012 def test_empty_list(
1013 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1015 mock_get_controller
.return_value
= juju
.controller
.Controller()
1016 mock_list_offers
.return_value
= []
1017 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1018 self
.assertEqual(offers
, [])
1019 mock_disconnect_controller
.assert_called_once()
1021 def test_non_empty_list(
1022 self
, mock_list_offers
, mock_disconnect_controller
, mock_get_controller
,
1024 mock_get_controller
.return_value
= juju
.controller
.Controller()
1025 mock_list_offers
.return_value
= ["offer"]
1026 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1027 self
.assertEqual(offers
, ["offer"])
1028 mock_disconnect_controller
.assert_called_once()
1031 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1032 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1033 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1034 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1035 @asynctest.mock
.patch("juju.model.Model.consume")
1036 class ConsumeTest(LibjujuTestCase
):
1038 super(ConsumeTest
, self
).setUp()
1043 mock_disconnect_controller
,
1044 mock_disconnect_model
,
1046 mock_get_controller
,
1048 mock_get_controller
.return_value
= juju
.controller
.Controller()
1049 mock_get_model
.return_value
= juju
.model
.Model()
1051 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1052 mock_consume
.assert_called_once()
1053 mock_disconnect_model
.assert_called_once()
1054 mock_disconnect_controller
.assert_called_once()
1056 def test_parsing_error_exception(
1059 mock_disconnect_controller
,
1060 mock_disconnect_model
,
1062 mock_get_controller
,
1064 mock_get_controller
.return_value
= juju
.controller
.Controller()
1065 mock_get_model
.return_value
= juju
.model
.Model()
1066 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1068 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1069 self
.loop
.run_until_complete(
1070 self
.libjuju
.consume("offer_url", "model_name")
1072 mock_consume
.assert_called_once()
1073 mock_disconnect_model
.assert_called_once()
1074 mock_disconnect_controller
.assert_called_once()
1076 def test_juju_error_exception(
1079 mock_disconnect_controller
,
1080 mock_disconnect_model
,
1082 mock_get_controller
,
1084 mock_get_controller
.return_value
= juju
.controller
.Controller()
1085 mock_get_model
.return_value
= juju
.model
.Model()
1086 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1088 with self
.assertRaises(juju
.errors
.JujuError
):
1089 self
.loop
.run_until_complete(
1090 self
.libjuju
.consume("offer_url", "model_name")
1092 mock_consume
.assert_called_once()
1093 mock_disconnect_model
.assert_called_once()
1094 mock_disconnect_controller
.assert_called_once()
1096 def test_juju_api_error_exception(
1099 mock_disconnect_controller
,
1100 mock_disconnect_model
,
1102 mock_get_controller
,
1104 mock_get_controller
.return_value
= juju
.controller
.Controller()
1105 mock_get_model
.return_value
= juju
.model
.Model()
1106 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1107 {"error": "", "response": "", "request-id": ""}
1110 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1111 self
.loop
.run_until_complete(
1112 self
.libjuju
.consume("offer_url", "model_name")
1114 mock_consume
.assert_called_once()
1115 mock_disconnect_model
.assert_called_once()
1116 mock_disconnect_controller
.assert_called_once()
1119 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1120 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1121 class AddK8sTest(LibjujuTestCase
):
1123 super(AddK8sTest
, self
).setUp()
1124 self
.configuration
= kubernetes
.client
.configuration
.Configuration()
1126 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1127 self
.loop
.run_until_complete(
1128 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1130 mock_add_cloud
.assert_called_once()
1131 mock_get_k8s_cloud_credential
.assert_called_once()
1133 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1134 mock_add_cloud
.side_effect
= Exception()
1135 with self
.assertRaises(Exception):
1136 self
.loop
.run_until_complete(
1137 self
.libjuju
.add_k8s("cloud", self
.configuration
, "storage_class")
1139 mock_add_cloud
.assert_called_once()
1140 mock_get_k8s_cloud_credential
.assert_called_once()
1142 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1143 with self
.assertRaises(Exception):
1144 self
.loop
.run_until_complete(
1145 self
.libjuju
.add_k8s("", self
.configuration
, "storage_class")
1147 mock_add_cloud
.assert_not_called()
1149 def test_add_k8s_missing_storage_name(
1150 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1152 with self
.assertRaises(Exception):
1153 self
.loop
.run_until_complete(
1154 self
.libjuju
.add_k8s("cloud", self
.configuration
, "")
1156 mock_add_cloud
.assert_not_called()
1158 def test_add_k8s_missing_configuration_keys(
1159 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1161 with self
.assertRaises(Exception):
1162 self
.loop
.run_until_complete(self
.libjuju
.add_k8s("cloud", None, ""))
1163 mock_add_cloud
.assert_not_called()
1166 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1167 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1168 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1169 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1170 class AddCloudTest(LibjujuTestCase
):
1172 super(AddCloudTest
, self
).setUp()
1173 self
.cloud
= juju
.client
.client
.Cloud()
1174 self
.credential
= juju
.client
.client
.CloudCredential()
1176 def test_add_cloud_with_credential(
1178 mock_add_credential
,
1180 mock_disconnect_controller
,
1181 mock_get_controller
,
1183 mock_get_controller
.return_value
= juju
.controller
.Controller()
1185 cloud
= self
.loop
.run_until_complete(
1186 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1188 self
.assertEqual(cloud
, self
.cloud
)
1189 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1190 mock_add_credential
.assert_called_once_with(
1191 "cloud", credential
=self
.credential
, cloud
="cloud"
1193 mock_disconnect_controller
.assert_called_once()
1195 def test_add_cloud_no_credential(
1197 mock_add_credential
,
1199 mock_disconnect_controller
,
1200 mock_get_controller
,
1202 mock_get_controller
.return_value
= juju
.controller
.Controller()
1204 cloud
= self
.loop
.run_until_complete(
1205 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1207 self
.assertEqual(cloud
, self
.cloud
)
1208 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1209 mock_add_credential
.assert_not_called()
1210 mock_disconnect_controller
.assert_called_once()
1212 def test_add_cloud_exception(
1214 mock_add_credential
,
1216 mock_disconnect_controller
,
1217 mock_get_controller
,
1219 mock_get_controller
.return_value
= juju
.controller
.Controller()
1220 mock_add_cloud
.side_effect
= Exception()
1221 with self
.assertRaises(Exception):
1222 self
.loop
.run_until_complete(
1223 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1226 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1227 mock_add_credential
.assert_not_called()
1228 mock_disconnect_controller
.assert_called_once()
1230 def test_add_credential_exception(
1232 mock_add_credential
,
1234 mock_disconnect_controller
,
1235 mock_get_controller
,
1237 mock_get_controller
.return_value
= juju
.controller
.Controller()
1238 mock_add_credential
.side_effect
= Exception()
1239 with self
.assertRaises(Exception):
1240 self
.loop
.run_until_complete(
1241 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1244 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1245 mock_add_credential
.assert_called_once_with(
1246 "cloud", credential
=self
.credential
, cloud
="cloud"
1248 mock_disconnect_controller
.assert_called_once()
1251 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1252 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1253 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1254 class RemoveCloudTest(LibjujuTestCase
):
1256 super(RemoveCloudTest
, self
).setUp()
1258 def test_remove_cloud(
1259 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1261 mock_get_controller
.return_value
= juju
.controller
.Controller()
1263 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1264 mock_remove_cloud
.assert_called_once_with("cloud")
1265 mock_disconnect_controller
.assert_called_once()
1267 def test_remove_cloud_exception(
1268 self
, mock_remove_cloud
, mock_disconnect_controller
, mock_get_controller
,
1270 mock_get_controller
.return_value
= juju
.controller
.Controller()
1271 mock_remove_cloud
.side_effect
= Exception()
1273 with self
.assertRaises(Exception):
1274 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1275 mock_remove_cloud
.assert_called_once_with("cloud")
1276 mock_disconnect_controller
.assert_called_once()
1279 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1280 class GetK8sCloudCredentials(LibjujuTestCase
):
1282 super(GetK8sCloudCredentials
, self
).setUp()
1284 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1285 def test_not_supported(self
, mock_exception
, mock_configuration
):
1286 mock_configuration
.username
= ""
1287 mock_configuration
.password
= ""
1288 mock_configuration
.ssl_ca_cert
= None
1289 mock_configuration
.cert_file
= None
1290 mock_configuration
.key_file
= None
1291 exception_raised
= False
1293 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1294 except JujuInvalidK8sConfiguration
as e
:
1295 exception_raised
= True
1297 e
.message
, "authentication method not supported",
1299 self
.assertTrue(exception_raised
)
1301 def test_user_pass(self
, mock_configuration
):
1302 mock_configuration
.username
= "admin"
1303 mock_configuration
.password
= "admin"
1304 mock_configuration
.ssl_ca_cert
= None
1305 mock_configuration
.cert_file
= None
1306 mock_configuration
.key_file
= None
1307 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1310 juju
.client
._definitions
.CloudCredential(
1311 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1315 def test_user_no_pass(self
, mock_configuration
):
1316 mock_configuration
.username
= "admin"
1317 mock_configuration
.password
= ""
1318 mock_configuration
.ssl_ca_cert
= None
1319 mock_configuration
.cert_file
= None
1320 mock_configuration
.key_file
= None
1321 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1322 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1325 juju
.client
._definitions
.CloudCredential(
1326 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1329 mock_debug
.assert_called_once_with(
1330 "credential for user admin has empty password"
1333 def test_user_pass_with_cert(self
, mock_configuration
):
1334 mock_configuration
.username
= "admin"
1335 mock_configuration
.password
= "admin"
1336 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1337 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1338 ssl_ca_cert_file
.write("cacert")
1339 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1340 mock_configuration
.cert_file
= None
1341 mock_configuration
.key_file
= None
1342 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1345 juju
.client
._definitions
.CloudCredential(
1347 "username": "admin",
1348 "password": "admin",
1349 "ClientCertificateData": "cacert",
1351 auth_type
="userpasswithcert",
1355 def test_cert(self
, mock_configuration
):
1356 mock_configuration
.username
= ""
1357 mock_configuration
.password
= ""
1358 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1359 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1360 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1361 ssl_ca_cert_file
.write("cacert")
1362 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1363 mock_configuration
.cert_file
= None
1364 mock_configuration
.key_file
= None
1365 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1368 juju
.client
._definitions
.CloudCredential(
1369 attrs
={"ClientCertificateData": "cacert", "Token": "Token"},
1370 auth_type
="certificate",
1374 def test_oauth2(self
, mock_configuration
):
1375 mock_configuration
.username
= ""
1376 mock_configuration
.password
= ""
1377 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1378 key
= tempfile
.NamedTemporaryFile()
1379 with
open(key
.name
, "w") as key_file
:
1380 key_file
.write("key")
1381 mock_configuration
.ssl_ca_cert
= None
1382 mock_configuration
.cert_file
= None
1383 mock_configuration
.key_file
= key
.name
1384 credential
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1387 juju
.client
._definitions
.CloudCredential(
1388 attrs
={"ClientKeyData": "key", "Token": "Token"}, auth_type
="oauth2",
1392 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1393 def test_oauth2_missing_token(self
, mock_exception
, mock_configuration
):
1394 mock_configuration
.username
= ""
1395 mock_configuration
.password
= ""
1396 key
= tempfile
.NamedTemporaryFile()
1397 with
open(key
.name
, "w") as key_file
:
1398 key_file
.write("key")
1399 mock_configuration
.ssl_ca_cert
= None
1400 mock_configuration
.cert_file
= None
1401 mock_configuration
.key_file
= key
.name
1402 exception_raised
= False
1404 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1405 except JujuInvalidK8sConfiguration
as e
:
1406 exception_raised
= True
1408 e
.message
, "missing token for auth type oauth2",
1410 self
.assertTrue(exception_raised
)
1412 def test_unknown_api_key(self
, mock_configuration
):
1413 mock_configuration
.username
= ""
1414 mock_configuration
.password
= ""
1415 mock_configuration
.api_key
= {"authorization": "Bearer Token Wrong"}
1416 mock_configuration
.ssl_ca_cert
= None
1417 mock_configuration
.cert_file
= None
1418 mock_configuration
.key_file
= None
1419 exception_raised
= False
1421 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1422 except JujuInvalidK8sConfiguration
as e
:
1423 exception_raised
= True
1425 e
.message
, "unknown format of api_key",
1427 self
.assertTrue(exception_raised
)
1429 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1430 mock_configuration
.username
= "admin"
1431 mock_configuration
.password
= "pass"
1432 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1433 mock_configuration
.ssl_ca_cert
= None
1434 mock_configuration
.cert_file
= None
1435 mock_configuration
.key_file
= None
1436 exception_raised
= False
1438 _
= self
.libjuju
.get_k8s_cloud_credential(mock_configuration
)
1439 except JujuInvalidK8sConfiguration
as e
:
1440 exception_raised
= True
1442 e
.message
, "Cannot set both token and user/pass",
1444 self
.assertTrue(exception_raised
)