1 # Copyright 2020 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
18 from unittest
import mock
21 from juju
.errors
import JujuAPIError
23 from .utils
import FakeN2VC
, FakeMachine
, FakeApplication
24 from n2vc
.libjuju
import Libjuju
25 from n2vc
.exceptions
import (
26 JujuControllerFailedConnecting
,
28 JujuApplicationNotFound
,
30 JujuApplicationExists
,
31 JujuInvalidK8sConfiguration
,
32 JujuLeaderUnitNotFound
,
35 from n2vc
.k8s_juju_conn
import generate_rbac_id
38 class LibjujuTestCase(asynctest
.TestCase
):
39 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
40 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
41 @asynctest.mock
.patch("juju.controller.Controller.connection")
42 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
45 mock__get_api_endpoints_db
=None,
48 mock_update_endpoints
=None,
50 loop
= asyncio
.get_event_loop()
52 mock__get_api_endpoints_db
.return_value
= ["127.0.0.1:17070"]
53 endpoints
= "127.0.0.1:17070"
57 -----BEGIN CERTIFICATE-----
59 -----END CERTIFICATE-----"""
60 self
.libjuju
= Libjuju(
62 "192.168.0.155:17070",
71 logging
.disable(logging
.CRITICAL
)
72 loop
.run_until_complete(self
.libjuju
.disconnect())
75 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._create_health_check_task")
76 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
77 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
78 class LibjujuInitTestCase(asynctest
.TestCase
):
80 self
.loop
= asyncio
.get_event_loop()
81 self
.n2vc
= FakeN2VC()
82 self
.endpoint
= "192.168.100.100:17070"
83 self
.username
= "admin"
84 self
.password
= "secret"
86 -----BEGIN CERTIFICATE-----
88 -----END CERTIFICATE-----"""
90 def test_endpoint_not_in_db(
92 mock__get_api_endpoints_db
,
93 mock_update_endpoints
,
94 mock_create_health_check_task
,
96 mock__get_api_endpoints_db
.return_value
= ["another_ip"]
99 "192.168.0.155:17070",
108 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
109 mock__get_api_endpoints_db
.assert_called_once()
111 def test_endpoint_in_db(
113 mock__get_api_endpoints_db
,
114 mock_update_endpoints
,
115 mock_create_health_check_task
,
117 mock__get_api_endpoints_db
.return_value
= [self
.endpoint
, "another_ip"]
120 "192.168.0.155:17070",
129 mock_update_endpoints
.assert_not_called()
130 mock__get_api_endpoints_db
.assert_called_once()
132 def test_no_db_endpoints(
134 mock__get_api_endpoints_db
,
135 mock_update_endpoints
,
136 mock_create_health_check_task
,
138 mock__get_api_endpoints_db
.return_value
= None
141 "192.168.0.155:17070",
150 mock_update_endpoints
.assert_called_once_with([self
.endpoint
])
151 mock__get_api_endpoints_db
.assert_called_once()
154 @asynctest.mock
.patch("juju.controller.Controller.connect")
155 @asynctest.mock
.patch(
156 "juju.controller.Controller.api_endpoints",
157 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
159 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
160 class GetControllerTest(LibjujuTestCase
):
162 super(GetControllerTest
, self
).setUp()
164 def test_diff_endpoint(
165 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
167 self
.libjuju
.endpoints
= []
168 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
169 mock__update_api_endpoints_db
.assert_called_once_with(["127.0.0.1:17070"])
170 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
172 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
175 mock_disconnect_controller
,
176 mock__update_api_endpoints_db
,
180 self
.libjuju
.endpoints
= []
181 mock__update_api_endpoints_db
.side_effect
= Exception()
183 with self
.assertRaises(JujuControllerFailedConnecting
):
184 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
185 self
.assertIsNone(controller
)
186 mock_disconnect_controller
.assert_called_once()
188 def test_same_endpoint_get_controller(
189 self
, mock__update_api_endpoints_db
, mock_api_endpoints
, mock_connect
191 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
192 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
193 mock__update_api_endpoints_db
.assert_not_called()
194 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
197 class DisconnectTest(LibjujuTestCase
):
199 super(DisconnectTest
, self
).setUp()
201 @asynctest.mock
.patch("juju.model.Model.disconnect")
202 def test_disconnect_model(self
, mock_disconnect
):
203 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
204 mock_disconnect
.assert_called_once()
206 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
207 def test_disconnect_controller(self
, mock_disconnect
):
208 self
.loop
.run_until_complete(
209 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
211 mock_disconnect
.assert_called_once()
214 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
215 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
216 @asynctest.mock
.patch("juju.controller.Controller.add_model")
217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
218 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
219 class AddModelTest(LibjujuTestCase
):
221 super(AddModelTest
, self
).setUp()
223 def test_existing_model(
225 mock_disconnect_model
,
226 mock_disconnect_controller
,
231 mock_model_exists
.return_value
= True
233 # This should not raise an exception
234 self
.loop
.run_until_complete(
235 self
.libjuju
.add_model("existing_model", "cloud")
238 mock_disconnect_controller
.assert_called()
240 # TODO Check two job executing at the same time and one returning without doing anything.
242 def test_non_existing_model(
244 mock_disconnect_model
,
245 mock_disconnect_controller
,
250 mock_model_exists
.return_value
= False
251 mock_get_controller
.return_value
= juju
.controller
.Controller()
253 self
.loop
.run_until_complete(
254 self
.libjuju
.add_model("nonexisting_model", "cloud")
257 mock_add_model
.assert_called_once()
258 mock_disconnect_controller
.assert_called()
259 mock_disconnect_model
.assert_called()
262 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
263 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
264 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
265 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
266 @asynctest.mock
.patch(
267 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
269 @asynctest.mock
.patch("juju.model.Model.get_action_status")
270 @asynctest.mock
.patch("juju.model.Model.get_action_output")
271 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_actions")
272 class GetExecutedActionsTest(LibjujuTestCase
):
274 super(GetExecutedActionsTest
, self
).setUp()
279 mock_get_action_output
,
280 mock_get_action_status
,
282 mock_disconnect_controller
,
283 mock_disconnect_model
,
287 mock_get_model
.return_value
= None
288 with self
.assertRaises(JujuError
):
289 self
.loop
.run_until_complete(self
.libjuju
.get_executed_actions("model"))
291 mock_get_controller
.assert_called_once()
292 mock_disconnect_controller
.assert_called_once()
293 mock_get_model
.assert_called_once()
294 mock_disconnect_model
.assert_not_called()
299 mock_get_action_output
,
300 mock_get_action_status
,
302 mock_disconnect_controller
,
303 mock_disconnect_model
,
307 mock_get_model
.return_value
= juju
.model
.Model()
308 mock_applications
.return_value
= {"existing_app"}
309 mock_get_actions
.return_value
= {"action_name": "description"}
310 mock_get_action_status
.return_value
= {"id": "status"}
311 mock_get_action_output
.return_value
= {"output": "completed"}
313 executed_actions
= self
.loop
.run_until_complete(
314 self
.libjuju
.get_executed_actions("model")
316 expected_result
= [{'id': 'id', 'action': 'action_name',
317 'status': 'status', 'output': 'completed'}]
318 self
.assertListEqual(expected_result
, executed_actions
)
319 self
.assertIsInstance(executed_actions
, list)
321 mock_get_controller
.assert_called_once()
322 mock_get_model
.assert_called_once()
323 mock_disconnect_controller
.assert_called_once()
324 mock_disconnect_model
.assert_called_once()
327 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
328 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
329 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
330 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
331 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
332 class GetApplicationConfigsTest(LibjujuTestCase
):
334 super(GetApplicationConfigsTest
, self
).setUp()
338 mock_get_application
,
339 mock_disconnect_controller
,
340 mock_disconnect_model
,
344 mock_get_model
.return_value
= None
345 with self
.assertRaises(JujuError
):
346 self
.loop
.run_until_complete(
347 self
.libjuju
.get_application_configs("model", "app"))
349 mock_get_controller
.assert_called_once()
350 mock_disconnect_controller
.assert_called_once()
351 mock_get_model
.assert_called_once()
352 mock_disconnect_model
.assert_not_called()
356 mock_get_application
,
357 mock_disconnect_controller
,
358 mock_disconnect_model
,
362 mock_get_application
.return_value
= FakeApplication()
363 application_configs
= self
.loop
.run_until_complete(self
.libjuju
364 .get_application_configs("model", "app"))
366 self
.assertEqual(application_configs
, ["app_config"])
368 mock_get_controller
.assert_called_once()
369 mock_get_model
.assert_called_once()
370 mock_disconnect_controller
.assert_called_once()
371 mock_disconnect_model
.assert_called_once()
374 @asynctest.mock
.patch("juju.controller.Controller.get_model")
375 class GetModelTest(LibjujuTestCase
):
377 super(GetModelTest
, self
).setUp()
383 mock_get_model
.return_value
= juju
.model
.Model()
384 model
= self
.loop
.run_until_complete(
385 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
387 self
.assertIsInstance(model
, juju
.model
.Model
)
390 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
391 @asynctest.mock
.patch("juju.controller.Controller.list_models")
392 class ModelExistsTest(LibjujuTestCase
):
394 super(ModelExistsTest
, self
).setUp()
396 async def test_existing_model(
401 mock_list_models
.return_value
= ["existing_model"]
403 await self
.libjuju
.model_exists(
404 "existing_model", juju
.controller
.Controller()
408 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
409 async def test_no_controller(
411 mock_disconnect_controller
,
415 mock_list_models
.return_value
= ["existing_model"]
416 mock_get_controller
.return_value
= juju
.controller
.Controller()
417 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
418 mock_disconnect_controller
.assert_called_once()
420 async def test_non_existing_model(
425 mock_list_models
.return_value
= ["existing_model"]
427 await self
.libjuju
.model_exists(
428 "not_existing_model", juju
.controller
.Controller()
433 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
434 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
435 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
436 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
437 @asynctest.mock
.patch("juju.model.Model.get_status")
438 class GetModelStatusTest(LibjujuTestCase
):
440 super(GetModelStatusTest
, self
).setUp()
445 mock_disconnect_controller
,
446 mock_disconnect_model
,
450 mock_get_model
.return_value
= juju
.model
.Model()
451 mock_get_status
.return_value
= {"status"}
453 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
455 mock_get_status
.assert_called_once()
456 mock_disconnect_controller
.assert_called_once()
457 mock_disconnect_model
.assert_called_once()
459 self
.assertEqual(status
, {"status"})
464 mock_disconnect_controller
,
465 mock_disconnect_model
,
469 mock_get_model
.return_value
= juju
.model
.Model()
470 mock_get_status
.side_effect
= Exception()
472 with self
.assertRaises(Exception):
473 status
= self
.loop
.run_until_complete(
474 self
.libjuju
.get_model_status("model")
477 mock_disconnect_controller
.assert_called_once()
478 mock_disconnect_model
.assert_called_once()
480 self
.assertIsNone(status
)
483 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
484 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
485 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
486 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
487 @asynctest.mock
.patch("juju.model.Model.get_machines")
488 @asynctest.mock
.patch("juju.model.Model.add_machine")
489 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
490 class CreateMachineTest(LibjujuTestCase
):
492 super(CreateMachineTest
, self
).setUp()
494 def test_existing_machine(
499 mock_disconnect_controller
,
500 mock_disconnect_model
,
504 mock_get_model
.return_value
= juju
.model
.Model()
505 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
506 machine
, bool_res
= self
.loop
.run_until_complete(
507 self
.libjuju
.create_machine("model", "existing_machine")
510 self
.assertIsInstance(machine
, FakeMachine
)
511 self
.assertFalse(bool_res
)
513 mock_disconnect_controller
.assert_called()
514 mock_disconnect_model
.assert_called()
516 def test_non_existing_machine(
521 mock_disconnect_controller
,
522 mock_disconnect_model
,
528 mock_get_model
.return_value
= juju
.model
.Model()
529 with self
.assertRaises(JujuMachineNotFound
):
530 machine
, bool_res
= self
.loop
.run_until_complete(
531 self
.libjuju
.create_machine("model", "non_existing_machine")
533 self
.assertIsNone(machine
)
534 self
.assertIsNone(bool_res
)
536 mock_disconnect_controller
.assert_called()
537 mock_disconnect_model
.assert_called()
544 mock_disconnect_controller
,
545 mock_disconnect_model
,
549 mock_get_model
.return_value
= juju
.model
.Model()
550 mock_add_machine
.return_value
= FakeMachine()
552 machine
, bool_res
= self
.loop
.run_until_complete(
553 self
.libjuju
.create_machine("model")
556 self
.assertIsInstance(machine
, FakeMachine
)
557 self
.assertTrue(bool_res
)
559 mock_wait_for
.assert_called_once()
560 mock_add_machine
.assert_called_once()
562 mock_disconnect_controller
.assert_called()
563 mock_disconnect_model
.assert_called()
566 # TODO test provision machine
569 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
570 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
571 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
572 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
573 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
574 @asynctest.mock
.patch("juju.model.Model.deploy")
575 class DeployTest(LibjujuTestCase
):
577 super(DeployTest
, self
).setUp()
583 mock_disconnect_controller
,
584 mock_disconnect_model
,
588 mock_get_model
.return_value
= juju
.model
.Model()
589 self
.loop
.run_until_complete(
590 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
592 mock_deploy
.assert_called_once()
593 mock_wait_for_model
.assert_called_once()
594 mock_disconnect_controller
.assert_called_once()
595 mock_disconnect_model
.assert_called_once()
597 def test_deploy_no_wait(
601 mock_disconnect_controller
,
602 mock_disconnect_model
,
606 mock_get_model
.return_value
= juju
.model
.Model()
607 self
.loop
.run_until_complete(
608 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
610 mock_deploy
.assert_called_once()
611 mock_wait_for_model
.assert_not_called()
612 mock_disconnect_controller
.assert_called_once()
613 mock_disconnect_model
.assert_called_once()
615 def test_deploy_exception(
619 mock_disconnect_controller
,
620 mock_disconnect_model
,
624 mock_deploy
.side_effect
= Exception()
625 mock_get_model
.return_value
= juju
.model
.Model()
626 with self
.assertRaises(Exception):
627 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
628 mock_deploy
.assert_called_once()
629 mock_wait_for_model
.assert_not_called()
630 mock_disconnect_controller
.assert_called_once()
631 mock_disconnect_model
.assert_called_once()
634 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
635 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
636 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
637 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
638 @asynctest.mock
.patch(
639 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
641 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
642 @asynctest.mock
.patch("juju.model.Model.deploy")
643 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
644 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
645 class DeployCharmTest(LibjujuTestCase
):
647 super(DeployCharmTest
, self
).setUp()
649 def test_existing_app(
656 mock_disconnect_controller
,
657 mock_disconnect_model
,
661 mock_get_model
.return_value
= juju
.model
.Model()
662 mock_applications
.return_value
= {"existing_app"}
665 with self
.assertRaises(JujuApplicationExists
):
666 application
= self
.loop
.run_until_complete(
667 self
.libjuju
.deploy_charm(
674 self
.assertIsNone(application
)
676 mock_disconnect_controller
.assert_called()
677 mock_disconnect_model
.assert_called()
679 def test_non_existing_machine(
686 mock_disconnect_controller
,
687 mock_disconnect_model
,
691 mock_get_model
.return_value
= juju
.model
.Model()
692 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
694 with self
.assertRaises(JujuMachineNotFound
):
695 application
= self
.loop
.run_until_complete(
696 self
.libjuju
.deploy_charm(
704 self
.assertIsNone(application
)
706 mock_disconnect_controller
.assert_called()
707 mock_disconnect_model
.assert_called()
716 mock_disconnect_controller
,
717 mock_disconnect_model
,
721 mock_get_model
.return_value
= juju
.model
.Model()
722 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
723 mock_create_machine
.return_value
= (FakeMachine(), "other")
724 mock_deploy
.return_value
= FakeApplication()
725 application
= self
.loop
.run_until_complete(
726 self
.libjuju
.deploy_charm(
735 self
.assertIsInstance(application
, FakeApplication
)
737 mock_deploy
.assert_called_once()
738 mock_wait_for
.assert_called_once()
740 mock_create_machine
.assert_called_once()
742 mock_disconnect_controller
.assert_called()
743 mock_disconnect_model
.assert_called()
752 mock_disconnect_controller
,
753 mock_disconnect_model
,
757 mock_get_model
.return_value
= juju
.model
.Model()
758 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
759 mock_deploy
.return_value
= FakeApplication()
760 application
= self
.loop
.run_until_complete(
761 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
764 self
.assertIsInstance(application
, FakeApplication
)
766 mock_deploy
.assert_called_once()
767 mock_wait_for
.assert_called_once()
769 mock_disconnect_controller
.assert_called()
770 mock_disconnect_model
.assert_called()
773 @asynctest.mock
.patch(
774 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
776 class GetApplicationTest(LibjujuTestCase
):
778 super(GetApplicationTest
, self
).setUp()
780 def test_existing_application(
784 mock_applications
.return_value
= {"existing_app": "exists"}
785 model
= juju
.model
.Model()
786 result
= self
.libjuju
._get
_application
(model
, "existing_app")
787 self
.assertEqual(result
, "exists")
789 def test_non_existing_application(
793 mock_applications
.return_value
= {"existing_app": "exists"}
794 model
= juju
.model
.Model()
795 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
796 self
.assertIsNone(result
)
799 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
800 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
801 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
802 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
803 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
804 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
805 @asynctest.mock
.patch("juju.model.Model.get_action_output")
806 @asynctest.mock
.patch("juju.model.Model.get_action_status")
807 class ExecuteActionTest(LibjujuTestCase
):
809 super(ExecuteActionTest
, self
).setUp()
811 def test_no_application(
813 mock_get_action_status
,
814 mock_get_action_output
,
816 mock__get_application
,
817 mock_disconnect_controller
,
818 mock_disconnect_model
,
822 mock__get_application
.return_value
= None
823 mock_get_model
.return_value
= juju
.model
.Model()
826 with self
.assertRaises(JujuApplicationNotFound
):
827 output
, status
= self
.loop
.run_until_complete(
828 self
.libjuju
.execute_action(
834 self
.assertIsNone(output
)
835 self
.assertIsNone(status
)
837 mock_disconnect_controller
.assert_called()
838 mock_disconnect_model
.assert_called()
842 mock_get_action_status
,
843 mock_get_action_output
,
845 mock__get_application
,
846 mock_disconnect_controller
,
847 mock_disconnect_model
,
852 mock_get_model
.return_value
= juju
.model
.Model()
853 mock__get_application
.return_value
= FakeApplication()
856 with self
.assertRaises(JujuActionNotFound
):
857 output
, status
= self
.loop
.run_until_complete(
858 self
.libjuju
.execute_action(
864 self
.assertIsNone(output
)
865 self
.assertIsNone(status
)
867 mock_disconnect_controller
.assert_called()
868 mock_disconnect_model
.assert_called()
870 @asynctest.mock
.patch("asyncio.sleep")
871 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
874 mock_is_leader_from_status
,
876 mock_get_action_status
,
877 mock_get_action_output
,
879 mock__get_application
,
880 mock_disconnect_controller
,
881 mock_disconnect_model
,
885 mock_get_model
.return_value
= juju
.model
.Model()
886 mock__get_application
.return_value
= FakeApplication()
887 mock_is_leader_from_status
.return_value
= False
890 with self
.assertRaises(JujuLeaderUnitNotFound
):
891 output
, status
= self
.loop
.run_until_complete(
892 self
.libjuju
.execute_action(
898 self
.assertIsNone(output
)
899 self
.assertIsNone(status
)
901 mock_disconnect_controller
.assert_called()
902 mock_disconnect_model
.assert_called()
904 def test_succesful_exec(
906 mock_get_action_status
,
907 mock_get_action_output
,
909 mock__get_application
,
910 mock_disconnect_controller
,
911 mock_disconnect_model
,
915 mock_get_model
.return_value
= juju
.model
.Model()
916 mock__get_application
.return_value
= FakeApplication()
917 mock_get_action_output
.return_value
= "output"
918 mock_get_action_status
.return_value
= {"id": "status"}
919 output
, status
= self
.loop
.run_until_complete(
920 self
.libjuju
.execute_action("app", "model", "existing_action")
922 self
.assertEqual(output
, "output")
923 self
.assertEqual(status
, "status")
925 mock_wait_for
.assert_called_once()
927 mock_disconnect_controller
.assert_called()
928 mock_disconnect_model
.assert_called()
931 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
932 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
933 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
934 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
935 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
936 class GetActionTest(LibjujuTestCase
):
938 super(GetActionTest
, self
).setUp()
942 mock_get_application
,
943 mock_disconnect_controller
,
944 mock_disconnect_model
,
948 mock_get_application
.side_effect
= Exception()
950 with self
.assertRaises(Exception):
951 actions
= self
.loop
.run_until_complete(
952 self
.libjuju
.get_actions("app", "model")
955 self
.assertIsNone(actions
)
956 mock_disconnect_controller
.assert_called_once()
957 mock_disconnect_model
.assert_called_once()
961 mock_get_application
,
962 mock_disconnect_controller
,
963 mock_disconnect_model
,
967 mock_get_application
.return_value
= FakeApplication()
969 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
971 self
.assertEqual(actions
, ["existing_action"])
973 mock_get_controller
.assert_called_once()
974 mock_get_model
.assert_called_once()
975 mock_disconnect_controller
.assert_called_once()
976 mock_disconnect_model
.assert_called_once()
979 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
980 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
981 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
982 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
983 @asynctest.mock
.patch("juju.application.Application.get_metrics")
984 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
985 class GetMetricsTest(LibjujuTestCase
):
987 super(GetMetricsTest
, self
).setUp()
989 def test_get_metrics_success(
991 mock_get_application
,
993 mock_disconnect_controller
,
994 mock_disconnect_model
,
998 mock_get_application
.return_value
= FakeApplication()
999 mock_get_model
.return_value
= juju
.model
.Model()
1001 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
1003 mock_disconnect_controller
.assert_called_once()
1004 mock_disconnect_model
.assert_called_once()
1006 def test_get_metrics_exception(
1008 mock_get_application
,
1010 mock_disconnect_controller
,
1011 mock_disconnect_model
,
1013 mock_get_controller
,
1015 mock_get_model
.return_value
= juju
.model
.Model()
1016 mock_get_metrics
.side_effect
= Exception()
1017 with self
.assertRaises(Exception):
1018 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
1020 mock_disconnect_controller
.assert_called_once()
1021 mock_disconnect_model
.assert_called_once()
1023 def test_missing_args_exception(
1025 mock_get_application
,
1027 mock_disconnect_controller
,
1028 mock_disconnect_model
,
1030 mock_get_controller
,
1032 mock_get_model
.return_value
= juju
.model
.Model()
1034 with self
.assertRaises(Exception):
1035 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
1037 mock_get_controller
.assert_not_called()
1038 mock_get_model
.assert_not_called()
1039 mock_disconnect_controller
.assert_not_called()
1040 mock_disconnect_model
.assert_not_called()
1043 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1044 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1045 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1046 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1047 @asynctest.mock
.patch("juju.model.Model.add_relation")
1048 class AddRelationTest(LibjujuTestCase
):
1050 super(AddRelationTest
, self
).setUp()
1052 @asynctest.mock
.patch("logging.Logger.warning")
1057 mock_disconnect_controller
,
1058 mock_disconnect_model
,
1060 mock_get_controller
,
1062 # TODO in libjuju.py should this fail only with a log message?
1063 result
= {"error": "not found", "response": "response", "request-id": 1}
1065 mock_get_model
.return_value
= juju
.model
.Model()
1066 mock_add_relation
.side_effect
= JujuAPIError(result
)
1068 self
.loop
.run_until_complete(
1069 self
.libjuju
.add_relation(
1076 mock_warning
.assert_called_with("Relation not found: not found")
1077 mock_disconnect_controller
.assert_called_once()
1078 mock_disconnect_model
.assert_called_once()
1080 @asynctest.mock
.patch("logging.Logger.warning")
1081 def test_already_exists(
1085 mock_disconnect_controller
,
1086 mock_disconnect_model
,
1088 mock_get_controller
,
1090 # TODO in libjuju.py should this fail silently?
1091 result
= {"error": "already exists", "response": "response", "request-id": 1}
1093 mock_get_model
.return_value
= juju
.model
.Model()
1094 mock_add_relation
.side_effect
= JujuAPIError(result
)
1096 self
.loop
.run_until_complete(
1097 self
.libjuju
.add_relation(
1104 mock_warning
.assert_called_with("Relation already exists: already exists")
1105 mock_disconnect_controller
.assert_called_once()
1106 mock_disconnect_model
.assert_called_once()
1111 mock_disconnect_controller
,
1112 mock_disconnect_model
,
1114 mock_get_controller
,
1116 mock_get_model
.return_value
= juju
.model
.Model()
1117 result
= {"error": "", "response": "response", "request-id": 1}
1118 mock_add_relation
.side_effect
= JujuAPIError(result
)
1120 with self
.assertRaises(JujuAPIError
):
1121 self
.loop
.run_until_complete(
1122 self
.libjuju
.add_relation(
1129 mock_disconnect_controller
.assert_called_once()
1130 mock_disconnect_model
.assert_called_once()
1135 mock_disconnect_controller
,
1136 mock_disconnect_model
,
1138 mock_get_controller
,
1140 mock_get_model
.return_value
= juju
.model
.Model()
1142 self
.loop
.run_until_complete(
1143 self
.libjuju
.add_relation(
1150 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1151 mock_disconnect_controller
.assert_called_once()
1152 mock_disconnect_model
.assert_called_once()
1157 mock_disconnect_controller
,
1158 mock_disconnect_model
,
1160 mock_get_controller
,
1162 mock_get_model
.return_value
= juju
.model
.Model()
1164 self
.loop
.run_until_complete(
1165 self
.libjuju
.add_relation(
1172 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1173 mock_disconnect_controller
.assert_called_once()
1174 mock_disconnect_model
.assert_called_once()
1177 # TODO destroy_model testcase
1180 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1181 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1182 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1183 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1184 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1185 class DestroyApplicationTest(LibjujuTestCase
):
1187 super(DestroyApplicationTest
, self
).setUp()
1191 mock_get_controller
,
1193 mock_disconnect_controller
,
1194 mock_get_application
,
1195 mock_disconnect_model
,
1197 mock_get_application
.return_value
= FakeApplication()
1198 mock_get_model
.return_value
= None
1199 self
.loop
.run_until_complete(
1200 self
.libjuju
.destroy_application(
1206 mock_get_application
.assert_called()
1207 mock_disconnect_controller
.assert_called_once()
1208 mock_disconnect_model
.assert_called_once()
1210 def test_no_application(
1212 mock_get_controller
,
1214 mock_disconnect_controller
,
1215 mock_get_application
,
1216 mock_disconnect_model
,
1218 mock_get_model
.return_value
= None
1219 mock_get_application
.return_value
= None
1221 self
.loop
.run_until_complete(
1222 self
.libjuju
.destroy_application(
1228 mock_get_application
.assert_called()
1232 mock_get_controller
,
1234 mock_disconnect_controller
,
1235 mock_get_application
,
1236 mock_disconnect_model
,
1238 mock_get_application
.return_value
= FakeApplication
1239 mock_get_model
.return_value
= None
1241 with self
.assertRaises(Exception):
1242 self
.loop
.run_until_complete(
1243 self
.libjuju
.destroy_application(
1249 mock_get_application
.assert_called_once()
1252 # @asynctest.mock.patch("juju.model.Model.get_machines")
1253 # @asynctest.mock.patch("logging.Logger.debug")
1254 # class DestroyMachineTest(LibjujuTestCase):
1256 # super(DestroyMachineTest, self).setUp()
1258 # def test_success_manual_machine(
1259 # self, mock_debug, mock_get_machines,
1261 # mock_get_machines.side_effect = [
1262 # {"machine": FakeManualMachine()},
1263 # {"machine": FakeManualMachine()},
1266 # self.loop.run_until_complete(
1267 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1270 # asynctest.call("Waiting for machine machine is destroyed"),
1271 # asynctest.call("Machine destroyed: machine"),
1273 # mock_debug.assert_has_calls(calls)
1275 # def test_no_machine(
1276 # self, mock_debug, mock_get_machines,
1278 # mock_get_machines.return_value = {}
1279 # self.loop.run_until_complete(
1280 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1282 # mock_debug.assert_called_with("Machine not found: machine")
1285 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1286 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1287 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1288 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1289 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1290 class ConfigureApplicationTest(LibjujuTestCase
):
1292 super(ConfigureApplicationTest
, self
).setUp()
1296 mock_get_application
,
1297 mock_disconnect_controller
,
1298 mock_disconnect_model
,
1300 mock_get_controller
,
1303 mock_get_application
.return_value
= FakeApplication()
1305 self
.loop
.run_until_complete(
1306 self
.libjuju
.configure_application(
1312 mock_get_application
.assert_called_once()
1313 mock_disconnect_controller
.assert_called_once()
1314 mock_disconnect_model
.assert_called_once()
1318 mock_get_application
,
1319 mock_disconnect_controller
,
1320 mock_disconnect_model
,
1322 mock_get_controller
,
1325 mock_get_application
.side_effect
= Exception()
1327 with self
.assertRaises(Exception):
1328 self
.loop
.run_until_complete(
1329 self
.libjuju
.configure_application(
1335 mock_disconnect_controller
.assert_called_once()
1336 mock_disconnect_model
.assert_called_once()
1338 def test_controller_exception(
1340 mock_get_application
,
1341 mock_disconnect_controller
,
1342 mock_disconnect_model
,
1344 mock_get_controller
,
1347 result
= {"error": "not found", "response": "response", "request-id": 1}
1349 mock_get_controller
.side_effect
= JujuAPIError(result
)
1351 with self
.assertRaises(JujuAPIError
):
1352 self
.loop
.run_until_complete(
1353 self
.libjuju
.configure_application(
1359 mock_get_model
.assert_not_called()
1360 mock_disconnect_controller
.assert_not_called()
1361 mock_disconnect_model
.assert_not_called()
1363 def test_get_model_exception(
1365 mock_get_application
,
1366 mock_disconnect_controller
,
1367 mock_disconnect_model
,
1369 mock_get_controller
,
1372 result
= {"error": "not found", "response": "response", "request-id": 1}
1373 mock_get_model
.side_effect
= JujuAPIError(result
)
1375 with self
.assertRaises(JujuAPIError
):
1376 self
.loop
.run_until_complete(
1377 self
.libjuju
.configure_application(
1383 mock_get_model
.assert_called_once()
1384 mock_disconnect_controller
.assert_called_once()
1385 mock_disconnect_model
.assert_not_called()
1388 # TODO _get_api_endpoints_db test case
1389 # TODO _update_api_endpoints_db test case
1390 # TODO healthcheck test case
1393 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1394 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1395 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1396 class ListModelsTest(LibjujuTestCase
):
1398 super(ListModelsTest
, self
).setUp()
1400 def test_containing(
1403 mock_disconnect_controller
,
1404 mock_get_controller
,
1406 mock_get_controller
.return_value
= juju
.controller
.Controller()
1407 mock_list_models
.return_value
= ["existingmodel"]
1408 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1410 mock_disconnect_controller
.assert_called_once()
1411 self
.assertEquals(models
, ["existingmodel"])
1413 def test_not_containing(
1416 mock_disconnect_controller
,
1417 mock_get_controller
,
1419 mock_get_controller
.return_value
= juju
.controller
.Controller()
1420 mock_list_models
.return_value
= ["existingmodel", "model"]
1421 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1423 mock_disconnect_controller
.assert_called_once()
1424 self
.assertEquals(models
, [])
1426 def test_no_contains_arg(
1429 mock_disconnect_controller
,
1430 mock_get_controller
,
1432 mock_get_controller
.return_value
= juju
.controller
.Controller()
1433 mock_list_models
.return_value
= ["existingmodel", "model"]
1434 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1436 mock_disconnect_controller
.assert_called_once()
1437 self
.assertEquals(models
, ["existingmodel", "model"])
1440 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1441 class ModelsExistTest(LibjujuTestCase
):
1443 super(ModelsExistTest
, self
).setUp()
1445 def test_model_names_none(self
, mock_list_models
):
1446 mock_list_models
.return_value
= []
1447 with self
.assertRaises(Exception):
1448 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1450 def test_model_names_empty(self
, mock_list_models
):
1451 mock_list_models
.return_value
= []
1452 with self
.assertRaises(Exception):
1453 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1454 self
.libjuju
.models_exist([])
1457 def test_model_names_not_existing(self
, mock_list_models
):
1458 mock_list_models
.return_value
= ["prometheus", "grafana"]
1459 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1460 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1462 self
.assertFalse(exist
)
1463 self
.assertEqual(non_existing_models
, ["prometheus2"])
1465 def test_model_names_exist(self
, mock_list_models
):
1466 mock_list_models
.return_value
= ["prometheus", "grafana"]
1467 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1468 self
.libjuju
.models_exist(["prometheus", "grafana"])
1470 self
.assertTrue(exist
)
1471 self
.assertEqual(non_existing_models
, [])
1474 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1475 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1476 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1477 class ListOffers(LibjujuTestCase
):
1479 super(ListOffers
, self
).setUp()
1481 def test_disconnect_controller(
1484 mock_disconnect_controller
,
1485 mock_get_controller
,
1487 mock_get_controller
.return_value
= juju
.controller
.Controller()
1488 mock_list_offers
.side_effect
= Exception()
1489 with self
.assertRaises(Exception):
1490 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1491 mock_disconnect_controller
.assert_called_once()
1493 def test_empty_list(
1496 mock_disconnect_controller
,
1497 mock_get_controller
,
1499 mock_get_controller
.return_value
= juju
.controller
.Controller()
1500 mock_list_offers
.return_value
= []
1501 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1502 self
.assertEqual(offers
, [])
1503 mock_disconnect_controller
.assert_called_once()
1505 def test_non_empty_list(
1508 mock_disconnect_controller
,
1509 mock_get_controller
,
1511 mock_get_controller
.return_value
= juju
.controller
.Controller()
1512 mock_list_offers
.return_value
= ["offer"]
1513 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1514 self
.assertEqual(offers
, ["offer"])
1515 mock_disconnect_controller
.assert_called_once()
1518 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1519 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1520 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1521 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1522 @asynctest.mock
.patch("juju.model.Model.consume")
1523 class ConsumeTest(LibjujuTestCase
):
1525 super(ConsumeTest
, self
).setUp()
1530 mock_disconnect_controller
,
1531 mock_disconnect_model
,
1533 mock_get_controller
,
1535 mock_get_controller
.return_value
= juju
.controller
.Controller()
1536 mock_get_model
.return_value
= juju
.model
.Model()
1538 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1539 mock_consume
.assert_called_once()
1540 mock_disconnect_model
.assert_called_once()
1541 mock_disconnect_controller
.assert_called_once()
1543 def test_parsing_error_exception(
1546 mock_disconnect_controller
,
1547 mock_disconnect_model
,
1549 mock_get_controller
,
1551 mock_get_controller
.return_value
= juju
.controller
.Controller()
1552 mock_get_model
.return_value
= juju
.model
.Model()
1553 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1555 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1556 self
.loop
.run_until_complete(
1557 self
.libjuju
.consume("offer_url", "model_name")
1559 mock_consume
.assert_called_once()
1560 mock_disconnect_model
.assert_called_once()
1561 mock_disconnect_controller
.assert_called_once()
1563 def test_juju_error_exception(
1566 mock_disconnect_controller
,
1567 mock_disconnect_model
,
1569 mock_get_controller
,
1571 mock_get_controller
.return_value
= juju
.controller
.Controller()
1572 mock_get_model
.return_value
= juju
.model
.Model()
1573 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1575 with self
.assertRaises(juju
.errors
.JujuError
):
1576 self
.loop
.run_until_complete(
1577 self
.libjuju
.consume("offer_url", "model_name")
1579 mock_consume
.assert_called_once()
1580 mock_disconnect_model
.assert_called_once()
1581 mock_disconnect_controller
.assert_called_once()
1583 def test_juju_api_error_exception(
1586 mock_disconnect_controller
,
1587 mock_disconnect_model
,
1589 mock_get_controller
,
1591 mock_get_controller
.return_value
= juju
.controller
.Controller()
1592 mock_get_model
.return_value
= juju
.model
.Model()
1593 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1594 {"error": "", "response": "", "request-id": ""}
1597 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1598 self
.loop
.run_until_complete(
1599 self
.libjuju
.consume("offer_url", "model_name")
1601 mock_consume
.assert_called_once()
1602 mock_disconnect_model
.assert_called_once()
1603 mock_disconnect_controller
.assert_called_once()
1606 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1607 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1608 class AddK8sTest(LibjujuTestCase
):
1610 super(AddK8sTest
, self
).setUp()
1612 rbac_id
= generate_rbac_id()
1614 client_cert_data
= "cert"
1615 configuration
= kubernetes
.client
.configuration
.Configuration()
1616 storage_class
= "storage_class"
1617 credential_name
= name
1619 self
._add
_k
8s
_args
= {
1623 "client_cert_data": client_cert_data
,
1624 "configuration": configuration
,
1625 "storage_class": storage_class
,
1626 "credential_name": credential_name
,
1629 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1630 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1631 mock_add_cloud
.assert_called_once()
1632 mock_get_k8s_cloud_credential
.assert_called_once()
1634 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1635 mock_add_cloud
.side_effect
= Exception()
1636 with self
.assertRaises(Exception):
1637 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1638 mock_add_cloud
.assert_called_once()
1639 mock_get_k8s_cloud_credential
.assert_called_once()
1641 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1642 self
._add
_k
8s
_args
["name"] = ""
1643 with self
.assertRaises(Exception):
1644 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1645 mock_add_cloud
.assert_not_called()
1647 def test_add_k8s_missing_storage_name(
1648 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1650 self
._add
_k
8s
_args
["storage_class"] = ""
1651 with self
.assertRaises(Exception):
1652 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1653 mock_add_cloud
.assert_not_called()
1655 def test_add_k8s_missing_configuration_keys(
1656 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1658 self
._add
_k
8s
_args
["configuration"] = None
1659 with self
.assertRaises(Exception):
1660 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1661 mock_add_cloud
.assert_not_called()
1664 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1665 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1666 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1667 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1668 class AddCloudTest(LibjujuTestCase
):
1670 super(AddCloudTest
, self
).setUp()
1671 self
.cloud
= juju
.client
.client
.Cloud()
1672 self
.credential
= juju
.client
.client
.CloudCredential()
1674 def test_add_cloud_with_credential(
1676 mock_add_credential
,
1678 mock_disconnect_controller
,
1679 mock_get_controller
,
1681 mock_get_controller
.return_value
= juju
.controller
.Controller()
1683 cloud
= self
.loop
.run_until_complete(
1684 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1686 self
.assertEqual(cloud
, self
.cloud
)
1687 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1688 mock_add_credential
.assert_called_once_with(
1689 "cloud", credential
=self
.credential
, cloud
="cloud"
1691 mock_disconnect_controller
.assert_called_once()
1693 def test_add_cloud_no_credential(
1695 mock_add_credential
,
1697 mock_disconnect_controller
,
1698 mock_get_controller
,
1700 mock_get_controller
.return_value
= juju
.controller
.Controller()
1702 cloud
= self
.loop
.run_until_complete(
1703 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1705 self
.assertEqual(cloud
, self
.cloud
)
1706 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1707 mock_add_credential
.assert_not_called()
1708 mock_disconnect_controller
.assert_called_once()
1710 def test_add_cloud_exception(
1712 mock_add_credential
,
1714 mock_disconnect_controller
,
1715 mock_get_controller
,
1717 mock_get_controller
.return_value
= juju
.controller
.Controller()
1718 mock_add_cloud
.side_effect
= Exception()
1719 with self
.assertRaises(Exception):
1720 self
.loop
.run_until_complete(
1721 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1724 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1725 mock_add_credential
.assert_not_called()
1726 mock_disconnect_controller
.assert_called_once()
1728 def test_add_credential_exception(
1730 mock_add_credential
,
1732 mock_disconnect_controller
,
1733 mock_get_controller
,
1735 mock_get_controller
.return_value
= juju
.controller
.Controller()
1736 mock_add_credential
.side_effect
= Exception()
1737 with self
.assertRaises(Exception):
1738 self
.loop
.run_until_complete(
1739 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1742 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1743 mock_add_credential
.assert_called_once_with(
1744 "cloud", credential
=self
.credential
, cloud
="cloud"
1746 mock_disconnect_controller
.assert_called_once()
1749 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1750 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1751 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1752 class RemoveCloudTest(LibjujuTestCase
):
1754 super(RemoveCloudTest
, self
).setUp()
1756 def test_remove_cloud(
1759 mock_disconnect_controller
,
1760 mock_get_controller
,
1762 mock_get_controller
.return_value
= juju
.controller
.Controller()
1764 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1765 mock_remove_cloud
.assert_called_once_with("cloud")
1766 mock_disconnect_controller
.assert_called_once()
1768 def test_remove_cloud_exception(
1771 mock_disconnect_controller
,
1772 mock_get_controller
,
1774 mock_get_controller
.return_value
= juju
.controller
.Controller()
1775 mock_remove_cloud
.side_effect
= Exception()
1777 with self
.assertRaises(Exception):
1778 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1779 mock_remove_cloud
.assert_called_once_with("cloud")
1780 mock_disconnect_controller
.assert_called_once()
1783 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1784 class GetK8sCloudCredentials(LibjujuTestCase
):
1786 super(GetK8sCloudCredentials
, self
).setUp()
1787 self
.cert_data
= "cert"
1788 self
.token
= "token"
1790 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1791 def test_not_supported(self
, mock_exception
, mock_configuration
):
1792 mock_configuration
.username
= ""
1793 mock_configuration
.password
= ""
1794 mock_configuration
.ssl_ca_cert
= None
1795 mock_configuration
.cert_file
= None
1796 mock_configuration
.key_file
= None
1797 exception_raised
= False
1799 self
.cert_data
= None
1801 _
= self
.libjuju
.get_k8s_cloud_credential(
1806 except JujuInvalidK8sConfiguration
as e
:
1807 exception_raised
= True
1810 "authentication method not supported",
1812 self
.assertTrue(exception_raised
)
1814 def test_user_pass(self
, mock_configuration
):
1815 mock_configuration
.username
= "admin"
1816 mock_configuration
.password
= "admin"
1817 mock_configuration
.ssl_ca_cert
= None
1818 mock_configuration
.cert_file
= None
1819 mock_configuration
.key_file
= None
1821 self
.cert_data
= None
1822 credential
= self
.libjuju
.get_k8s_cloud_credential(
1829 juju
.client
._definitions
.CloudCredential(
1830 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1834 def test_user_pass_with_cert(self
, mock_configuration
):
1835 mock_configuration
.username
= "admin"
1836 mock_configuration
.password
= "admin"
1837 mock_configuration
.ssl_ca_cert
= None
1838 mock_configuration
.cert_file
= None
1839 mock_configuration
.key_file
= None
1841 credential
= self
.libjuju
.get_k8s_cloud_credential(
1848 juju
.client
._definitions
.CloudCredential(
1850 "ClientCertificateData": self
.cert_data
,
1851 "username": "admin",
1852 "password": "admin",
1854 auth_type
="userpasswithcert",
1858 def test_user_no_pass(self
, mock_configuration
):
1859 mock_configuration
.username
= "admin"
1860 mock_configuration
.password
= ""
1861 mock_configuration
.ssl_ca_cert
= None
1862 mock_configuration
.cert_file
= None
1863 mock_configuration
.key_file
= None
1865 self
.cert_data
= None
1866 with mock
.patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1867 credential
= self
.libjuju
.get_k8s_cloud_credential(
1874 juju
.client
._definitions
.CloudCredential(
1875 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1878 mock_debug
.assert_called_once_with(
1879 "credential for user admin has empty password"
1882 def test_cert(self
, mock_configuration
):
1883 mock_configuration
.username
= ""
1884 mock_configuration
.password
= ""
1885 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1886 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1887 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1888 ssl_ca_cert_file
.write("cacert")
1889 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1890 mock_configuration
.cert_file
= None
1891 mock_configuration
.key_file
= None
1892 credential
= self
.libjuju
.get_k8s_cloud_credential(
1899 juju
.client
._definitions
.CloudCredential(
1900 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1901 auth_type
="certificate",
1905 # TODO: Fix this test when oauth authentication is supported
1906 # def test_oauth2(self, mock_configuration):
1907 # mock_configuration.username = ""
1908 # mock_configuration.password = ""
1909 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1910 # key = tempfile.NamedTemporaryFile()
1911 # with open(key.name, "w") as key_file:
1912 # key_file.write("key")
1913 # mock_configuration.ssl_ca_cert = None
1914 # mock_configuration.cert_file = None
1915 # mock_configuration.key_file = key.name
1916 # credential = self.libjuju.get_k8s_cloud_credential(
1917 # mock_configuration,
1923 # juju.client._definitions.CloudCredential(
1924 # attrs={"ClientKeyData": "key", "Token": "Token"},
1925 # auth_type="oauth2",
1929 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1930 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1931 # mock_configuration.username = ""
1932 # mock_configuration.password = ""
1933 # key = tempfile.NamedTemporaryFile()
1934 # with open(key.name, "w") as key_file:
1935 # key_file.write("key")
1936 # mock_configuration.ssl_ca_cert = None
1937 # mock_configuration.cert_file = None
1938 # mock_configuration.key_file = key.name
1939 # exception_raised = False
1941 # _ = self.libjuju.get_k8s_cloud_credential(
1942 # mock_configuration,
1946 # except JujuInvalidK8sConfiguration as e:
1947 # exception_raised = True
1950 # "missing token for auth type oauth2",
1952 # self.assertTrue(exception_raised)
1954 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1955 mock_configuration
.username
= "admin"
1956 mock_configuration
.password
= "pass"
1957 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1958 mock_configuration
.ssl_ca_cert
= None
1959 mock_configuration
.cert_file
= None
1960 mock_configuration
.key_file
= None
1961 exception_raised
= False
1963 _
= self
.libjuju
.get_k8s_cloud_credential(
1968 except JujuInvalidK8sConfiguration
as e
:
1969 exception_raised
= True
1972 "Cannot set both token and user/pass",
1974 self
.assertTrue(exception_raised
)