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
.mock
import Mock
, patch
21 from juju
.errors
import JujuAPIError
23 from .utils
import 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
36 from n2vc
.tests
.unit
.utils
import AsyncMock
37 from n2vc
.vca
.connection
import Connection
38 from n2vc
.vca
.connection_data
import ConnectionData
41 cacert
= """-----BEGIN CERTIFICATE-----
43 -----END CERTIFICATE-----"""
46 @asynctest.mock
.patch("n2vc.libjuju.Controller")
47 class LibjujuTestCase(asynctest
.TestCase
):
48 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
51 mock_base64_to_cacert
=None,
53 self
.loop
= asyncio
.get_event_loop()
55 mock_base64_to_cacert
.return_value
= cacert
56 Connection
._load
_vca
_connection
_data
= Mock()
57 vca_connection
= Connection(AsyncMock())
58 vca_connection
._data
= ConnectionData(
60 "endpoints": ["1.2.3.4:17070"],
66 "lxd-credentials": "credentials",
67 "k8s-cloud": "k8s_cloud",
68 "k8s-credentials": "k8s_credentials",
70 "api-proxy": "api_proxy",
73 logging
.disable(logging
.CRITICAL
)
74 self
.libjuju
= Libjuju(vca_connection
, self
.loop
)
75 self
.loop
.run_until_complete(self
.libjuju
.disconnect())
78 @asynctest.mock
.patch("juju.controller.Controller.connect")
79 @asynctest.mock
.patch(
80 "juju.controller.Controller.api_endpoints",
81 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
83 class GetControllerTest(LibjujuTestCase
):
85 super(GetControllerTest
, self
).setUp()
87 def test_diff_endpoint(self
, mock_api_endpoints
, mock_connect
):
88 self
.libjuju
.endpoints
= []
89 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
90 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
92 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
95 mock_disconnect_controller
,
99 self
.libjuju
.endpoints
= []
101 mock_connect
.side_effect
= Exception()
103 with self
.assertRaises(JujuControllerFailedConnecting
):
104 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
105 self
.assertIsNone(controller
)
106 mock_disconnect_controller
.assert_called_once()
108 def test_same_endpoint_get_controller(self
, mock_api_endpoints
, mock_connect
):
109 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
110 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
111 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
114 class DisconnectTest(LibjujuTestCase
):
116 super(DisconnectTest
, self
).setUp()
118 @asynctest.mock
.patch("juju.model.Model.disconnect")
119 def test_disconnect_model(self
, mock_disconnect
):
120 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
121 mock_disconnect
.assert_called_once()
123 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
124 def test_disconnect_controller(self
, mock_disconnect
):
125 self
.loop
.run_until_complete(
126 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
128 mock_disconnect
.assert_called_once()
131 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
132 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
133 @asynctest.mock
.patch("juju.controller.Controller.add_model")
134 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
135 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
136 class AddModelTest(LibjujuTestCase
):
138 super(AddModelTest
, self
).setUp()
140 def test_existing_model(
142 mock_disconnect_model
,
143 mock_disconnect_controller
,
148 mock_model_exists
.return_value
= True
150 # This should not raise an exception
151 self
.loop
.run_until_complete(self
.libjuju
.add_model("existing_model", "cloud"))
153 mock_disconnect_controller
.assert_called()
155 # TODO Check two job executing at the same time and one returning without doing anything.
157 def test_non_existing_model(
159 mock_disconnect_model
,
160 mock_disconnect_controller
,
165 mock_model_exists
.return_value
= False
166 mock_get_controller
.return_value
= juju
.controller
.Controller()
168 self
.loop
.run_until_complete(
169 self
.libjuju
.add_model("nonexisting_model", Mock())
172 mock_add_model
.assert_called_once()
173 mock_disconnect_controller
.assert_called()
174 mock_disconnect_model
.assert_called()
177 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
178 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
179 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
180 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
181 @asynctest.mock
.patch(
182 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
184 @asynctest.mock
.patch("juju.model.Model.get_action_status")
185 @asynctest.mock
.patch("juju.model.Model.get_action_output")
186 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_actions")
187 class GetExecutedActionsTest(LibjujuTestCase
):
189 super(GetExecutedActionsTest
, self
).setUp()
194 mock_get_action_output
,
195 mock_get_action_status
,
197 mock_disconnect_controller
,
198 mock_disconnect_model
,
202 mock_get_model
.return_value
= None
203 with self
.assertRaises(JujuError
):
204 self
.loop
.run_until_complete(self
.libjuju
.get_executed_actions("model"))
206 mock_get_controller
.assert_called_once()
207 mock_disconnect_controller
.assert_called_once()
208 mock_get_model
.assert_called_once()
209 mock_disconnect_model
.assert_not_called()
214 mock_get_action_output
,
215 mock_get_action_status
,
217 mock_disconnect_controller
,
218 mock_disconnect_model
,
222 mock_get_model
.return_value
= juju
.model
.Model()
223 mock_applications
.return_value
= {"existing_app"}
224 mock_get_actions
.return_value
= {"action_name": "description"}
225 mock_get_action_status
.return_value
= {"id": "status"}
226 mock_get_action_output
.return_value
= {"output": "completed"}
228 executed_actions
= self
.loop
.run_until_complete(
229 self
.libjuju
.get_executed_actions("model")
231 expected_result
= [{'id': 'id', 'action': 'action_name',
232 'status': 'status', 'output': 'completed'}]
233 self
.assertListEqual(expected_result
, executed_actions
)
234 self
.assertIsInstance(executed_actions
, list)
236 mock_get_controller
.assert_called_once()
237 mock_get_model
.assert_called_once()
238 mock_disconnect_controller
.assert_called_once()
239 mock_disconnect_model
.assert_called_once()
242 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
243 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
244 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
245 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
246 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
247 class GetApplicationConfigsTest(LibjujuTestCase
):
249 super(GetApplicationConfigsTest
, self
).setUp()
253 mock_get_application
,
254 mock_disconnect_controller
,
255 mock_disconnect_model
,
259 mock_get_model
.return_value
= None
260 with self
.assertRaises(JujuError
):
261 self
.loop
.run_until_complete(
262 self
.libjuju
.get_application_configs("model", "app"))
264 mock_get_controller
.assert_called_once()
265 mock_disconnect_controller
.assert_called_once()
266 mock_get_model
.assert_called_once()
267 mock_disconnect_model
.assert_not_called()
271 mock_get_application
,
272 mock_disconnect_controller
,
273 mock_disconnect_model
,
277 mock_get_application
.return_value
= FakeApplication()
278 application_configs
= self
.loop
.run_until_complete(self
.libjuju
279 .get_application_configs("model", "app"))
281 self
.assertEqual(application_configs
, ["app_config"])
283 mock_get_controller
.assert_called_once()
284 mock_get_model
.assert_called_once()
285 mock_disconnect_controller
.assert_called_once()
286 mock_disconnect_model
.assert_called_once()
289 @asynctest.mock
.patch("juju.controller.Controller.get_model")
290 class GetModelTest(LibjujuTestCase
):
292 super(GetModelTest
, self
).setUp()
298 mock_get_model
.return_value
= juju
.model
.Model()
299 model
= self
.loop
.run_until_complete(
300 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
302 self
.assertIsInstance(model
, juju
.model
.Model
)
305 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
306 @asynctest.mock
.patch("juju.controller.Controller.list_models")
307 class ModelExistsTest(LibjujuTestCase
):
309 super(ModelExistsTest
, self
).setUp()
311 async def test_existing_model(
316 mock_list_models
.return_value
= ["existing_model"]
318 await self
.libjuju
.model_exists(
319 "existing_model", juju
.controller
.Controller()
323 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
324 async def test_no_controller(
326 mock_disconnect_controller
,
330 mock_list_models
.return_value
= ["existing_model"]
331 mock_get_controller
.return_value
= juju
.controller
.Controller()
332 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
333 mock_disconnect_controller
.assert_called_once()
335 async def test_non_existing_model(
340 mock_list_models
.return_value
= ["existing_model"]
342 await self
.libjuju
.model_exists(
343 "not_existing_model", juju
.controller
.Controller()
348 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
349 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
350 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
351 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
352 @asynctest.mock
.patch("juju.model.Model.get_status")
353 class GetModelStatusTest(LibjujuTestCase
):
355 super(GetModelStatusTest
, self
).setUp()
360 mock_disconnect_controller
,
361 mock_disconnect_model
,
365 mock_get_model
.return_value
= juju
.model
.Model()
366 mock_get_status
.return_value
= {"status"}
368 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
370 mock_get_status
.assert_called_once()
371 mock_disconnect_controller
.assert_called_once()
372 mock_disconnect_model
.assert_called_once()
374 self
.assertEqual(status
, {"status"})
379 mock_disconnect_controller
,
380 mock_disconnect_model
,
384 mock_get_model
.return_value
= juju
.model
.Model()
385 mock_get_status
.side_effect
= Exception()
387 with self
.assertRaises(Exception):
388 status
= self
.loop
.run_until_complete(
389 self
.libjuju
.get_model_status("model")
392 mock_disconnect_controller
.assert_called_once()
393 mock_disconnect_model
.assert_called_once()
395 self
.assertIsNone(status
)
398 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
399 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
400 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
401 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
402 @asynctest.mock
.patch("juju.model.Model.get_machines")
403 @asynctest.mock
.patch("juju.model.Model.add_machine")
404 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
405 class CreateMachineTest(LibjujuTestCase
):
407 super(CreateMachineTest
, self
).setUp()
409 def test_existing_machine(
414 mock_disconnect_controller
,
415 mock_disconnect_model
,
419 mock_get_model
.return_value
= juju
.model
.Model()
420 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
421 machine
, bool_res
= self
.loop
.run_until_complete(
422 self
.libjuju
.create_machine("model", "existing_machine")
425 self
.assertIsInstance(machine
, FakeMachine
)
426 self
.assertFalse(bool_res
)
428 mock_disconnect_controller
.assert_called()
429 mock_disconnect_model
.assert_called()
431 def test_non_existing_machine(
436 mock_disconnect_controller
,
437 mock_disconnect_model
,
443 mock_get_model
.return_value
= juju
.model
.Model()
444 with self
.assertRaises(JujuMachineNotFound
):
445 machine
, bool_res
= self
.loop
.run_until_complete(
446 self
.libjuju
.create_machine("model", "non_existing_machine")
448 self
.assertIsNone(machine
)
449 self
.assertIsNone(bool_res
)
451 mock_disconnect_controller
.assert_called()
452 mock_disconnect_model
.assert_called()
459 mock_disconnect_controller
,
460 mock_disconnect_model
,
464 mock_get_model
.return_value
= juju
.model
.Model()
465 mock_add_machine
.return_value
= FakeMachine()
467 machine
, bool_res
= self
.loop
.run_until_complete(
468 self
.libjuju
.create_machine("model")
471 self
.assertIsInstance(machine
, FakeMachine
)
472 self
.assertTrue(bool_res
)
474 mock_wait_for
.assert_called_once()
475 mock_add_machine
.assert_called_once()
477 mock_disconnect_controller
.assert_called()
478 mock_disconnect_model
.assert_called()
481 # TODO test provision machine
484 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
485 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
486 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
487 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
488 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
489 @asynctest.mock
.patch("juju.model.Model.deploy")
490 class DeployTest(LibjujuTestCase
):
492 super(DeployTest
, self
).setUp()
498 mock_disconnect_controller
,
499 mock_disconnect_model
,
503 mock_get_model
.return_value
= juju
.model
.Model()
504 self
.loop
.run_until_complete(
505 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
507 mock_deploy
.assert_called_once()
508 mock_wait_for_model
.assert_called_once()
509 mock_disconnect_controller
.assert_called_once()
510 mock_disconnect_model
.assert_called_once()
512 def test_deploy_no_wait(
516 mock_disconnect_controller
,
517 mock_disconnect_model
,
521 mock_get_model
.return_value
= juju
.model
.Model()
522 self
.loop
.run_until_complete(
523 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
525 mock_deploy
.assert_called_once()
526 mock_wait_for_model
.assert_not_called()
527 mock_disconnect_controller
.assert_called_once()
528 mock_disconnect_model
.assert_called_once()
530 def test_deploy_exception(
534 mock_disconnect_controller
,
535 mock_disconnect_model
,
539 mock_deploy
.side_effect
= Exception()
540 mock_get_model
.return_value
= juju
.model
.Model()
541 with self
.assertRaises(Exception):
542 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
543 mock_deploy
.assert_called_once()
544 mock_wait_for_model
.assert_not_called()
545 mock_disconnect_controller
.assert_called_once()
546 mock_disconnect_model
.assert_called_once()
549 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
550 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
551 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
552 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
553 @asynctest.mock
.patch(
554 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
556 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
557 @asynctest.mock
.patch("juju.model.Model.deploy")
558 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
559 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
560 class DeployCharmTest(LibjujuTestCase
):
562 super(DeployCharmTest
, self
).setUp()
564 def test_existing_app(
571 mock_disconnect_controller
,
572 mock_disconnect_model
,
576 mock_get_model
.return_value
= juju
.model
.Model()
577 mock_applications
.return_value
= {"existing_app"}
580 with self
.assertRaises(JujuApplicationExists
):
581 application
= self
.loop
.run_until_complete(
582 self
.libjuju
.deploy_charm(
589 self
.assertIsNone(application
)
591 mock_disconnect_controller
.assert_called()
592 mock_disconnect_model
.assert_called()
594 def test_non_existing_machine(
601 mock_disconnect_controller
,
602 mock_disconnect_model
,
606 mock_get_model
.return_value
= juju
.model
.Model()
607 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
609 with self
.assertRaises(JujuMachineNotFound
):
610 application
= self
.loop
.run_until_complete(
611 self
.libjuju
.deploy_charm(
619 self
.assertIsNone(application
)
621 mock_disconnect_controller
.assert_called()
622 mock_disconnect_model
.assert_called()
631 mock_disconnect_controller
,
632 mock_disconnect_model
,
636 mock_get_model
.return_value
= juju
.model
.Model()
637 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
638 mock_create_machine
.return_value
= (FakeMachine(), "other")
639 mock_deploy
.return_value
= FakeApplication()
640 application
= self
.loop
.run_until_complete(
641 self
.libjuju
.deploy_charm(
650 self
.assertIsInstance(application
, FakeApplication
)
652 mock_deploy
.assert_called_once()
653 mock_wait_for
.assert_called_once()
655 mock_create_machine
.assert_called_once()
657 mock_disconnect_controller
.assert_called()
658 mock_disconnect_model
.assert_called()
667 mock_disconnect_controller
,
668 mock_disconnect_model
,
672 mock_get_model
.return_value
= juju
.model
.Model()
673 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
674 mock_deploy
.return_value
= FakeApplication()
675 application
= self
.loop
.run_until_complete(
676 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
679 self
.assertIsInstance(application
, FakeApplication
)
681 mock_deploy
.assert_called_once()
682 mock_wait_for
.assert_called_once()
684 mock_disconnect_controller
.assert_called()
685 mock_disconnect_model
.assert_called()
688 @asynctest.mock
.patch(
689 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
691 class GetApplicationTest(LibjujuTestCase
):
693 super(GetApplicationTest
, self
).setUp()
695 def test_existing_application(
699 mock_applications
.return_value
= {"existing_app": "exists"}
700 model
= juju
.model
.Model()
701 result
= self
.libjuju
._get
_application
(model
, "existing_app")
702 self
.assertEqual(result
, "exists")
704 def test_non_existing_application(
708 mock_applications
.return_value
= {"existing_app": "exists"}
709 model
= juju
.model
.Model()
710 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
711 self
.assertIsNone(result
)
714 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
715 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
716 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
717 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
718 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
719 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
720 @asynctest.mock
.patch("juju.model.Model.get_action_output")
721 @asynctest.mock
.patch("juju.model.Model.get_action_status")
722 class ExecuteActionTest(LibjujuTestCase
):
724 super(ExecuteActionTest
, self
).setUp()
726 def test_no_application(
728 mock_get_action_status
,
729 mock_get_action_output
,
731 mock__get_application
,
732 mock_disconnect_controller
,
733 mock_disconnect_model
,
737 mock__get_application
.return_value
= None
738 mock_get_model
.return_value
= juju
.model
.Model()
741 with self
.assertRaises(JujuApplicationNotFound
):
742 output
, status
= self
.loop
.run_until_complete(
743 self
.libjuju
.execute_action(
749 self
.assertIsNone(output
)
750 self
.assertIsNone(status
)
752 mock_disconnect_controller
.assert_called()
753 mock_disconnect_model
.assert_called()
757 mock_get_action_status
,
758 mock_get_action_output
,
760 mock__get_application
,
761 mock_disconnect_controller
,
762 mock_disconnect_model
,
767 mock_get_model
.return_value
= juju
.model
.Model()
768 mock__get_application
.return_value
= FakeApplication()
771 with self
.assertRaises(JujuActionNotFound
):
772 output
, status
= self
.loop
.run_until_complete(
773 self
.libjuju
.execute_action(
779 self
.assertIsNone(output
)
780 self
.assertIsNone(status
)
782 mock_disconnect_controller
.assert_called()
783 mock_disconnect_model
.assert_called()
785 @asynctest.mock
.patch("asyncio.sleep")
786 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
789 mock_is_leader_from_status
,
791 mock_get_action_status
,
792 mock_get_action_output
,
794 mock__get_application
,
795 mock_disconnect_controller
,
796 mock_disconnect_model
,
800 mock_get_model
.return_value
= juju
.model
.Model()
801 mock__get_application
.return_value
= FakeApplication()
802 mock_is_leader_from_status
.return_value
= False
805 with self
.assertRaises(JujuLeaderUnitNotFound
):
806 output
, status
= self
.loop
.run_until_complete(
807 self
.libjuju
.execute_action(
813 self
.assertIsNone(output
)
814 self
.assertIsNone(status
)
816 mock_disconnect_controller
.assert_called()
817 mock_disconnect_model
.assert_called()
819 def test_succesful_exec(
821 mock_get_action_status
,
822 mock_get_action_output
,
824 mock__get_application
,
825 mock_disconnect_controller
,
826 mock_disconnect_model
,
830 mock_get_model
.return_value
= juju
.model
.Model()
831 mock__get_application
.return_value
= FakeApplication()
832 mock_get_action_output
.return_value
= "output"
833 mock_get_action_status
.return_value
= {"id": "status"}
834 output
, status
= self
.loop
.run_until_complete(
835 self
.libjuju
.execute_action("app", "model", "existing_action")
837 self
.assertEqual(output
, "output")
838 self
.assertEqual(status
, "status")
840 mock_wait_for
.assert_called_once()
842 mock_disconnect_controller
.assert_called()
843 mock_disconnect_model
.assert_called()
846 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
847 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
848 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
849 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
850 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
851 class GetActionTest(LibjujuTestCase
):
853 super(GetActionTest
, self
).setUp()
857 mock_get_application
,
858 mock_disconnect_controller
,
859 mock_disconnect_model
,
863 mock_get_application
.side_effect
= Exception()
865 with self
.assertRaises(Exception):
866 actions
= self
.loop
.run_until_complete(
867 self
.libjuju
.get_actions("app", "model")
870 self
.assertIsNone(actions
)
871 mock_disconnect_controller
.assert_called_once()
872 mock_disconnect_model
.assert_called_once()
876 mock_get_application
,
877 mock_disconnect_controller
,
878 mock_disconnect_model
,
882 mock_get_application
.return_value
= FakeApplication()
884 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
886 self
.assertEqual(actions
, ["existing_action"])
888 mock_get_controller
.assert_called_once()
889 mock_get_model
.assert_called_once()
890 mock_disconnect_controller
.assert_called_once()
891 mock_disconnect_model
.assert_called_once()
894 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
895 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
896 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
897 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
898 @asynctest.mock
.patch("juju.application.Application.get_metrics")
899 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
900 class GetMetricsTest(LibjujuTestCase
):
902 super(GetMetricsTest
, self
).setUp()
904 def test_get_metrics_success(
906 mock_get_application
,
908 mock_disconnect_controller
,
909 mock_disconnect_model
,
913 mock_get_application
.return_value
= FakeApplication()
914 mock_get_model
.return_value
= juju
.model
.Model()
916 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
918 mock_disconnect_controller
.assert_called_once()
919 mock_disconnect_model
.assert_called_once()
921 def test_get_metrics_exception(
923 mock_get_application
,
925 mock_disconnect_controller
,
926 mock_disconnect_model
,
930 mock_get_model
.return_value
= juju
.model
.Model()
931 mock_get_metrics
.side_effect
= Exception()
932 with self
.assertRaises(Exception):
933 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
935 mock_disconnect_controller
.assert_called_once()
936 mock_disconnect_model
.assert_called_once()
938 def test_missing_args_exception(
940 mock_get_application
,
942 mock_disconnect_controller
,
943 mock_disconnect_model
,
947 mock_get_model
.return_value
= juju
.model
.Model()
949 with self
.assertRaises(Exception):
950 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
952 mock_get_controller
.assert_not_called()
953 mock_get_model
.assert_not_called()
954 mock_disconnect_controller
.assert_not_called()
955 mock_disconnect_model
.assert_not_called()
958 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
959 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
960 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
961 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
962 @asynctest.mock
.patch("juju.model.Model.add_relation")
963 class AddRelationTest(LibjujuTestCase
):
965 super(AddRelationTest
, self
).setUp()
967 @asynctest.mock
.patch("logging.Logger.warning")
972 mock_disconnect_controller
,
973 mock_disconnect_model
,
977 # TODO in libjuju.py should this fail only with a log message?
978 result
= {"error": "not found", "response": "response", "request-id": 1}
980 mock_get_model
.return_value
= juju
.model
.Model()
981 mock_add_relation
.side_effect
= JujuAPIError(result
)
983 self
.loop
.run_until_complete(
984 self
.libjuju
.add_relation(
991 mock_warning
.assert_called_with("Relation not found: not found")
992 mock_disconnect_controller
.assert_called_once()
993 mock_disconnect_model
.assert_called_once()
995 @asynctest.mock
.patch("logging.Logger.warning")
996 def test_already_exists(
1000 mock_disconnect_controller
,
1001 mock_disconnect_model
,
1003 mock_get_controller
,
1005 # TODO in libjuju.py should this fail silently?
1006 result
= {"error": "already exists", "response": "response", "request-id": 1}
1008 mock_get_model
.return_value
= juju
.model
.Model()
1009 mock_add_relation
.side_effect
= JujuAPIError(result
)
1011 self
.loop
.run_until_complete(
1012 self
.libjuju
.add_relation(
1019 mock_warning
.assert_called_with("Relation already exists: already exists")
1020 mock_disconnect_controller
.assert_called_once()
1021 mock_disconnect_model
.assert_called_once()
1026 mock_disconnect_controller
,
1027 mock_disconnect_model
,
1029 mock_get_controller
,
1031 mock_get_model
.return_value
= juju
.model
.Model()
1032 result
= {"error": "", "response": "response", "request-id": 1}
1033 mock_add_relation
.side_effect
= JujuAPIError(result
)
1035 with self
.assertRaises(JujuAPIError
):
1036 self
.loop
.run_until_complete(
1037 self
.libjuju
.add_relation(
1044 mock_disconnect_controller
.assert_called_once()
1045 mock_disconnect_model
.assert_called_once()
1050 mock_disconnect_controller
,
1051 mock_disconnect_model
,
1053 mock_get_controller
,
1055 mock_get_model
.return_value
= juju
.model
.Model()
1057 self
.loop
.run_until_complete(
1058 self
.libjuju
.add_relation(
1065 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1066 mock_disconnect_controller
.assert_called_once()
1067 mock_disconnect_model
.assert_called_once()
1072 mock_disconnect_controller
,
1073 mock_disconnect_model
,
1075 mock_get_controller
,
1077 mock_get_model
.return_value
= juju
.model
.Model()
1079 self
.loop
.run_until_complete(
1080 self
.libjuju
.add_relation(
1087 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1088 mock_disconnect_controller
.assert_called_once()
1089 mock_disconnect_model
.assert_called_once()
1092 # TODO destroy_model testcase
1095 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1096 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1097 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1098 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1099 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1100 class DestroyApplicationTest(LibjujuTestCase
):
1102 super(DestroyApplicationTest
, self
).setUp()
1106 mock_get_controller
,
1108 mock_disconnect_controller
,
1109 mock_get_application
,
1110 mock_disconnect_model
,
1112 mock_get_application
.return_value
= FakeApplication()
1113 mock_get_model
.return_value
= None
1114 self
.loop
.run_until_complete(
1115 self
.libjuju
.destroy_application(
1121 mock_get_application
.assert_called()
1122 mock_disconnect_controller
.assert_called_once()
1123 mock_disconnect_model
.assert_called_once()
1125 def test_no_application(
1127 mock_get_controller
,
1129 mock_disconnect_controller
,
1130 mock_get_application
,
1131 mock_disconnect_model
,
1133 mock_get_model
.return_value
= None
1134 mock_get_application
.return_value
= None
1136 self
.loop
.run_until_complete(
1137 self
.libjuju
.destroy_application(
1143 mock_get_application
.assert_called()
1147 mock_get_controller
,
1149 mock_disconnect_controller
,
1150 mock_get_application
,
1151 mock_disconnect_model
,
1153 mock_get_application
.return_value
= FakeApplication
1154 mock_get_model
.return_value
= None
1156 with self
.assertRaises(Exception):
1157 self
.loop
.run_until_complete(
1158 self
.libjuju
.destroy_application(
1164 mock_get_application
.assert_called_once()
1167 # @asynctest.mock.patch("juju.model.Model.get_machines")
1168 # @asynctest.mock.patch("logging.Logger.debug")
1169 # class DestroyMachineTest(LibjujuTestCase):
1171 # super(DestroyMachineTest, self).setUp()
1173 # def test_success_manual_machine(
1174 # self, mock_debug, mock_get_machines,
1176 # mock_get_machines.side_effect = [
1177 # {"machine": FakeManualMachine()},
1178 # {"machine": FakeManualMachine()},
1181 # self.loop.run_until_complete(
1182 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1185 # asynctest.call("Waiting for machine machine is destroyed"),
1186 # asynctest.call("Machine destroyed: machine"),
1188 # mock_debug.assert_has_calls(calls)
1190 # def test_no_machine(
1191 # self, mock_debug, mock_get_machines,
1193 # mock_get_machines.return_value = {}
1194 # self.loop.run_until_complete(
1195 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1197 # mock_debug.assert_called_with("Machine not found: machine")
1200 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1201 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1202 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1203 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1204 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1205 class ConfigureApplicationTest(LibjujuTestCase
):
1207 super(ConfigureApplicationTest
, self
).setUp()
1211 mock_get_application
,
1212 mock_disconnect_controller
,
1213 mock_disconnect_model
,
1215 mock_get_controller
,
1218 mock_get_application
.return_value
= FakeApplication()
1220 self
.loop
.run_until_complete(
1221 self
.libjuju
.configure_application(
1227 mock_get_application
.assert_called_once()
1228 mock_disconnect_controller
.assert_called_once()
1229 mock_disconnect_model
.assert_called_once()
1233 mock_get_application
,
1234 mock_disconnect_controller
,
1235 mock_disconnect_model
,
1237 mock_get_controller
,
1240 mock_get_application
.side_effect
= Exception()
1242 with self
.assertRaises(Exception):
1243 self
.loop
.run_until_complete(
1244 self
.libjuju
.configure_application(
1250 mock_disconnect_controller
.assert_called_once()
1251 mock_disconnect_model
.assert_called_once()
1253 def test_controller_exception(
1255 mock_get_application
,
1256 mock_disconnect_controller
,
1257 mock_disconnect_model
,
1259 mock_get_controller
,
1262 result
= {"error": "not found", "response": "response", "request-id": 1}
1264 mock_get_controller
.side_effect
= JujuAPIError(result
)
1266 with self
.assertRaises(JujuAPIError
):
1267 self
.loop
.run_until_complete(
1268 self
.libjuju
.configure_application(
1274 mock_get_model
.assert_not_called()
1275 mock_disconnect_controller
.assert_not_called()
1276 mock_disconnect_model
.assert_not_called()
1278 def test_get_model_exception(
1280 mock_get_application
,
1281 mock_disconnect_controller
,
1282 mock_disconnect_model
,
1284 mock_get_controller
,
1287 result
= {"error": "not found", "response": "response", "request-id": 1}
1288 mock_get_model
.side_effect
= JujuAPIError(result
)
1290 with self
.assertRaises(JujuAPIError
):
1291 self
.loop
.run_until_complete(
1292 self
.libjuju
.configure_application(
1298 mock_get_model
.assert_called_once()
1299 mock_disconnect_controller
.assert_called_once()
1300 mock_disconnect_model
.assert_not_called()
1303 # TODO _get_api_endpoints_db test case
1304 # TODO _update_api_endpoints_db test case
1305 # TODO healthcheck test case
1308 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1309 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1310 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1311 class ListModelsTest(LibjujuTestCase
):
1313 super(ListModelsTest
, self
).setUp()
1315 def test_containing(
1318 mock_disconnect_controller
,
1319 mock_get_controller
,
1321 mock_get_controller
.return_value
= juju
.controller
.Controller()
1322 mock_list_models
.return_value
= ["existingmodel"]
1323 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1325 mock_disconnect_controller
.assert_called_once()
1326 self
.assertEquals(models
, ["existingmodel"])
1328 def test_not_containing(
1331 mock_disconnect_controller
,
1332 mock_get_controller
,
1334 mock_get_controller
.return_value
= juju
.controller
.Controller()
1335 mock_list_models
.return_value
= ["existingmodel", "model"]
1336 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1338 mock_disconnect_controller
.assert_called_once()
1339 self
.assertEquals(models
, [])
1341 def test_no_contains_arg(
1344 mock_disconnect_controller
,
1345 mock_get_controller
,
1347 mock_get_controller
.return_value
= juju
.controller
.Controller()
1348 mock_list_models
.return_value
= ["existingmodel", "model"]
1349 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1351 mock_disconnect_controller
.assert_called_once()
1352 self
.assertEquals(models
, ["existingmodel", "model"])
1355 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1356 class ModelsExistTest(LibjujuTestCase
):
1358 super(ModelsExistTest
, self
).setUp()
1360 def test_model_names_none(self
, mock_list_models
):
1361 mock_list_models
.return_value
= []
1362 with self
.assertRaises(Exception):
1363 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1365 def test_model_names_empty(self
, mock_list_models
):
1366 mock_list_models
.return_value
= []
1367 with self
.assertRaises(Exception):
1368 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1369 self
.libjuju
.models_exist([])
1372 def test_model_names_not_existing(self
, mock_list_models
):
1373 mock_list_models
.return_value
= ["prometheus", "grafana"]
1374 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1375 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1377 self
.assertFalse(exist
)
1378 self
.assertEqual(non_existing_models
, ["prometheus2"])
1380 def test_model_names_exist(self
, mock_list_models
):
1381 mock_list_models
.return_value
= ["prometheus", "grafana"]
1382 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1383 self
.libjuju
.models_exist(["prometheus", "grafana"])
1385 self
.assertTrue(exist
)
1386 self
.assertEqual(non_existing_models
, [])
1389 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1390 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1391 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1392 class ListOffers(LibjujuTestCase
):
1394 super(ListOffers
, self
).setUp()
1396 def test_disconnect_controller(
1399 mock_disconnect_controller
,
1400 mock_get_controller
,
1402 mock_get_controller
.return_value
= juju
.controller
.Controller()
1403 mock_list_offers
.side_effect
= Exception()
1404 with self
.assertRaises(Exception):
1405 self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1406 mock_disconnect_controller
.assert_called_once()
1408 def test_empty_list(
1411 mock_disconnect_controller
,
1412 mock_get_controller
,
1414 mock_get_controller
.return_value
= juju
.controller
.Controller()
1415 mock_list_offers
.return_value
= []
1416 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1417 self
.assertEqual(offers
, [])
1418 mock_disconnect_controller
.assert_called_once()
1420 def test_non_empty_list(
1423 mock_disconnect_controller
,
1424 mock_get_controller
,
1426 mock_get_controller
.return_value
= juju
.controller
.Controller()
1427 mock_list_offers
.return_value
= ["offer"]
1428 offers
= self
.loop
.run_until_complete(self
.libjuju
.list_offers("model"))
1429 self
.assertEqual(offers
, ["offer"])
1430 mock_disconnect_controller
.assert_called_once()
1433 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1434 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1435 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1436 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1437 @asynctest.mock
.patch("juju.model.Model.consume")
1438 class ConsumeTest(LibjujuTestCase
):
1440 super(ConsumeTest
, self
).setUp()
1445 mock_disconnect_controller
,
1446 mock_disconnect_model
,
1448 mock_get_controller
,
1450 mock_get_controller
.return_value
= juju
.controller
.Controller()
1451 mock_get_model
.return_value
= juju
.model
.Model()
1453 self
.loop
.run_until_complete(self
.libjuju
.consume("offer_url", "model_name"))
1454 mock_consume
.assert_called_once()
1455 mock_disconnect_model
.assert_called_once()
1456 mock_disconnect_controller
.assert_called_once()
1458 def test_parsing_error_exception(
1461 mock_disconnect_controller
,
1462 mock_disconnect_model
,
1464 mock_get_controller
,
1466 mock_get_controller
.return_value
= juju
.controller
.Controller()
1467 mock_get_model
.return_value
= juju
.model
.Model()
1468 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1470 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1471 self
.loop
.run_until_complete(
1472 self
.libjuju
.consume("offer_url", "model_name")
1474 mock_consume
.assert_called_once()
1475 mock_disconnect_model
.assert_called_once()
1476 mock_disconnect_controller
.assert_called_once()
1478 def test_juju_error_exception(
1481 mock_disconnect_controller
,
1482 mock_disconnect_model
,
1484 mock_get_controller
,
1486 mock_get_controller
.return_value
= juju
.controller
.Controller()
1487 mock_get_model
.return_value
= juju
.model
.Model()
1488 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1490 with self
.assertRaises(juju
.errors
.JujuError
):
1491 self
.loop
.run_until_complete(
1492 self
.libjuju
.consume("offer_url", "model_name")
1494 mock_consume
.assert_called_once()
1495 mock_disconnect_model
.assert_called_once()
1496 mock_disconnect_controller
.assert_called_once()
1498 def test_juju_api_error_exception(
1501 mock_disconnect_controller
,
1502 mock_disconnect_model
,
1504 mock_get_controller
,
1506 mock_get_controller
.return_value
= juju
.controller
.Controller()
1507 mock_get_model
.return_value
= juju
.model
.Model()
1508 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1509 {"error": "", "response": "", "request-id": ""}
1512 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1513 self
.loop
.run_until_complete(
1514 self
.libjuju
.consume("offer_url", "model_name")
1516 mock_consume
.assert_called_once()
1517 mock_disconnect_model
.assert_called_once()
1518 mock_disconnect_controller
.assert_called_once()
1521 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1522 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1523 class AddK8sTest(LibjujuTestCase
):
1525 super(AddK8sTest
, self
).setUp()
1527 rbac_id
= generate_rbac_id()
1529 client_cert_data
= "cert"
1530 configuration
= kubernetes
.client
.configuration
.Configuration()
1531 storage_class
= "storage_class"
1532 credential_name
= name
1534 self
._add
_k
8s
_args
= {
1538 "client_cert_data": client_cert_data
,
1539 "configuration": configuration
,
1540 "storage_class": storage_class
,
1541 "credential_name": credential_name
,
1544 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1545 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1546 mock_add_cloud
.assert_called_once()
1547 mock_get_k8s_cloud_credential
.assert_called_once()
1549 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1550 mock_add_cloud
.side_effect
= Exception()
1551 with self
.assertRaises(Exception):
1552 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1553 mock_add_cloud
.assert_called_once()
1554 mock_get_k8s_cloud_credential
.assert_called_once()
1556 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1557 self
._add
_k
8s
_args
["name"] = ""
1558 with self
.assertRaises(Exception):
1559 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1560 mock_add_cloud
.assert_not_called()
1562 def test_add_k8s_missing_storage_name(
1563 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1565 self
._add
_k
8s
_args
["storage_class"] = ""
1566 with self
.assertRaises(Exception):
1567 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1568 mock_add_cloud
.assert_not_called()
1570 def test_add_k8s_missing_configuration_keys(
1571 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1573 self
._add
_k
8s
_args
["configuration"] = None
1574 with self
.assertRaises(Exception):
1575 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1576 mock_add_cloud
.assert_not_called()
1579 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1580 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1581 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1582 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1583 class AddCloudTest(LibjujuTestCase
):
1585 super(AddCloudTest
, self
).setUp()
1586 self
.cloud
= juju
.client
.client
.Cloud()
1587 self
.credential
= juju
.client
.client
.CloudCredential()
1589 def test_add_cloud_with_credential(
1591 mock_add_credential
,
1593 mock_disconnect_controller
,
1594 mock_get_controller
,
1596 mock_get_controller
.return_value
= juju
.controller
.Controller()
1598 cloud
= self
.loop
.run_until_complete(
1599 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1601 self
.assertEqual(cloud
, self
.cloud
)
1602 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1603 mock_add_credential
.assert_called_once_with(
1604 "cloud", credential
=self
.credential
, cloud
="cloud"
1606 mock_disconnect_controller
.assert_called_once()
1608 def test_add_cloud_no_credential(
1610 mock_add_credential
,
1612 mock_disconnect_controller
,
1613 mock_get_controller
,
1615 mock_get_controller
.return_value
= juju
.controller
.Controller()
1617 cloud
= self
.loop
.run_until_complete(
1618 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1620 self
.assertEqual(cloud
, self
.cloud
)
1621 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1622 mock_add_credential
.assert_not_called()
1623 mock_disconnect_controller
.assert_called_once()
1625 def test_add_cloud_exception(
1627 mock_add_credential
,
1629 mock_disconnect_controller
,
1630 mock_get_controller
,
1632 mock_get_controller
.return_value
= juju
.controller
.Controller()
1633 mock_add_cloud
.side_effect
= Exception()
1634 with self
.assertRaises(Exception):
1635 self
.loop
.run_until_complete(
1636 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1639 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1640 mock_add_credential
.assert_not_called()
1641 mock_disconnect_controller
.assert_called_once()
1643 def test_add_credential_exception(
1645 mock_add_credential
,
1647 mock_disconnect_controller
,
1648 mock_get_controller
,
1650 mock_get_controller
.return_value
= juju
.controller
.Controller()
1651 mock_add_credential
.side_effect
= Exception()
1652 with self
.assertRaises(Exception):
1653 self
.loop
.run_until_complete(
1654 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1657 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1658 mock_add_credential
.assert_called_once_with(
1659 "cloud", credential
=self
.credential
, cloud
="cloud"
1661 mock_disconnect_controller
.assert_called_once()
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.remove_cloud")
1667 class RemoveCloudTest(LibjujuTestCase
):
1669 super(RemoveCloudTest
, self
).setUp()
1671 def test_remove_cloud(
1674 mock_disconnect_controller
,
1675 mock_get_controller
,
1677 mock_get_controller
.return_value
= juju
.controller
.Controller()
1679 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1680 mock_remove_cloud
.assert_called_once_with("cloud")
1681 mock_disconnect_controller
.assert_called_once()
1683 def test_remove_cloud_exception(
1686 mock_disconnect_controller
,
1687 mock_get_controller
,
1689 mock_get_controller
.return_value
= juju
.controller
.Controller()
1690 mock_remove_cloud
.side_effect
= Exception()
1692 with self
.assertRaises(Exception):
1693 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1694 mock_remove_cloud
.assert_called_once_with("cloud")
1695 mock_disconnect_controller
.assert_called_once()
1698 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1699 class GetK8sCloudCredentials(LibjujuTestCase
):
1701 super(GetK8sCloudCredentials
, self
).setUp()
1702 self
.cert_data
= "cert"
1703 self
.token
= "token"
1705 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1706 def test_not_supported(self
, mock_exception
, mock_configuration
):
1707 mock_configuration
.username
= ""
1708 mock_configuration
.password
= ""
1709 mock_configuration
.ssl_ca_cert
= None
1710 mock_configuration
.cert_file
= None
1711 mock_configuration
.key_file
= None
1712 exception_raised
= False
1714 self
.cert_data
= None
1716 _
= self
.libjuju
.get_k8s_cloud_credential(
1721 except JujuInvalidK8sConfiguration
as e
:
1722 exception_raised
= True
1725 "authentication method not supported",
1727 self
.assertTrue(exception_raised
)
1729 def test_user_pass(self
, mock_configuration
):
1730 mock_configuration
.username
= "admin"
1731 mock_configuration
.password
= "admin"
1732 mock_configuration
.ssl_ca_cert
= None
1733 mock_configuration
.cert_file
= None
1734 mock_configuration
.key_file
= None
1736 self
.cert_data
= None
1737 credential
= self
.libjuju
.get_k8s_cloud_credential(
1744 juju
.client
._definitions
.CloudCredential(
1745 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1749 def test_user_pass_with_cert(self
, mock_configuration
):
1750 mock_configuration
.username
= "admin"
1751 mock_configuration
.password
= "admin"
1752 mock_configuration
.ssl_ca_cert
= None
1753 mock_configuration
.cert_file
= None
1754 mock_configuration
.key_file
= None
1756 credential
= self
.libjuju
.get_k8s_cloud_credential(
1763 juju
.client
._definitions
.CloudCredential(
1765 "ClientCertificateData": self
.cert_data
,
1766 "username": "admin",
1767 "password": "admin",
1769 auth_type
="userpasswithcert",
1773 def test_user_no_pass(self
, mock_configuration
):
1774 mock_configuration
.username
= "admin"
1775 mock_configuration
.password
= ""
1776 mock_configuration
.ssl_ca_cert
= None
1777 mock_configuration
.cert_file
= None
1778 mock_configuration
.key_file
= None
1780 self
.cert_data
= None
1781 with patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1782 credential
= self
.libjuju
.get_k8s_cloud_credential(
1789 juju
.client
._definitions
.CloudCredential(
1790 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1793 mock_debug
.assert_called_once_with(
1794 "credential for user admin has empty password"
1797 def test_cert(self
, mock_configuration
):
1798 mock_configuration
.username
= ""
1799 mock_configuration
.password
= ""
1800 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1801 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1802 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1803 ssl_ca_cert_file
.write("cacert")
1804 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1805 mock_configuration
.cert_file
= None
1806 mock_configuration
.key_file
= None
1807 credential
= self
.libjuju
.get_k8s_cloud_credential(
1814 juju
.client
._definitions
.CloudCredential(
1815 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1816 auth_type
="certificate",
1820 # TODO: Fix this test when oauth authentication is supported
1821 # def test_oauth2(self, mock_configuration):
1822 # mock_configuration.username = ""
1823 # mock_configuration.password = ""
1824 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1825 # key = tempfile.NamedTemporaryFile()
1826 # with open(key.name, "w") as key_file:
1827 # key_file.write("key")
1828 # mock_configuration.ssl_ca_cert = None
1829 # mock_configuration.cert_file = None
1830 # mock_configuration.key_file = key.name
1831 # credential = self.libjuju.get_k8s_cloud_credential(
1832 # mock_configuration,
1838 # juju.client._definitions.CloudCredential(
1839 # attrs={"ClientKeyData": "key", "Token": "Token"},
1840 # auth_type="oauth2",
1844 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1845 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1846 # mock_configuration.username = ""
1847 # mock_configuration.password = ""
1848 # key = tempfile.NamedTemporaryFile()
1849 # with open(key.name, "w") as key_file:
1850 # key_file.write("key")
1851 # mock_configuration.ssl_ca_cert = None
1852 # mock_configuration.cert_file = None
1853 # mock_configuration.key_file = key.name
1854 # exception_raised = False
1856 # _ = self.libjuju.get_k8s_cloud_credential(
1857 # mock_configuration,
1861 # except JujuInvalidK8sConfiguration as e:
1862 # exception_raised = True
1865 # "missing token for auth type oauth2",
1867 # self.assertTrue(exception_raised)
1869 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
1870 mock_configuration
.username
= "admin"
1871 mock_configuration
.password
= "pass"
1872 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
1873 mock_configuration
.ssl_ca_cert
= None
1874 mock_configuration
.cert_file
= None
1875 mock_configuration
.key_file
= None
1876 exception_raised
= False
1878 _
= self
.libjuju
.get_k8s_cloud_credential(
1883 except JujuInvalidK8sConfiguration
as e
:
1884 exception_raised
= True
1887 "Cannot set both token and user/pass",
1889 self
.assertTrue(exception_raised
)
1892 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1893 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1894 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1895 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1896 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1897 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
1898 class ScaleApplicationTest(LibjujuTestCase
):
1900 super(ScaleApplicationTest
, self
).setUp()
1902 @asynctest.mock
.patch("asyncio.sleep")
1903 def test_scale_application(
1906 mock_wait_for_model
,
1907 mock_disconnect_controller
,
1908 mock_disconnect_model
,
1909 mock_get_application
,
1911 mock_get_controller
,
1913 mock_get_model
.return_value
= juju
.model
.Model()
1914 mock_get_application
.return_value
= FakeApplication()
1915 self
.loop
.run_until_complete(
1916 self
.libjuju
.scale_application(
1922 mock_wait_for_model
.assert_called_once()
1923 mock_disconnect_controller
.assert_called_once()
1924 mock_disconnect_model
.assert_called_once()
1926 def test_no_application(
1929 mock_disconnect_controller
,
1930 mock_disconnect_model
,
1931 mock_get_application
,
1933 mock_get_controller
,
1935 mock_get_application
.return_value
= None
1936 mock_get_model
.return_value
= juju
.model
.Model()
1937 with self
.assertRaises(JujuApplicationNotFound
):
1938 self
.loop
.run_until_complete(
1939 self
.libjuju
.scale_application(
1945 mock_disconnect_controller
.assert_called()
1946 mock_disconnect_model
.assert_called()
1951 mock_disconnect_controller
,
1952 mock_disconnect_model
,
1953 mock_get_application
,
1955 mock_get_controller
,
1957 mock_get_model
.return_value
= None
1958 mock_get_application
.return_value
= FakeApplication()
1959 with self
.assertRaises(Exception):
1960 self
.loop
.run_until_complete(
1961 self
.libjuju
.scale_application(
1968 mock_disconnect_controller
.assert_called_once()
1971 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1972 class GetUnitNumberTest(LibjujuTestCase
):
1974 super(GetUnitNumberTest
, self
).setUp()
1976 def test_succesful_get_unit_number(
1978 mock_get_applications
,
1980 mock_get_applications
.return_value
= FakeApplication()
1981 model
= juju
.model
.Model()
1982 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1983 self
.assertEqual(result
, 2)
1985 def test_non_existing_application(
1987 mock_get_applications
,
1989 mock_get_applications
.return_value
= None
1990 model
= juju
.model
.Model()
1991 result
= self
.libjuju
._get
_application
_count
(model
, "app")
1992 self
.assertEqual(result
, None)