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
24 from n2vc
.definitions
import Offer
, RelationEndpoint
31 from n2vc
.libjuju
import Libjuju
32 from n2vc
.exceptions
import (
33 JujuControllerFailedConnecting
,
35 JujuApplicationNotFound
,
37 JujuApplicationExists
,
38 JujuInvalidK8sConfiguration
,
39 JujuLeaderUnitNotFound
,
42 from n2vc
.k8s_juju_conn
import generate_rbac_id
43 from n2vc
.tests
.unit
.utils
import AsyncMock
44 from n2vc
.vca
.connection
import Connection
45 from n2vc
.vca
.connection_data
import ConnectionData
48 cacert
= """-----BEGIN CERTIFICATE-----
50 -----END CERTIFICATE-----"""
53 @asynctest.mock
.patch("n2vc.libjuju.Controller")
54 class LibjujuTestCase(asynctest
.TestCase
):
55 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
58 mock_base64_to_cacert
=None,
60 self
.loop
= asyncio
.get_event_loop()
62 mock_base64_to_cacert
.return_value
= cacert
63 Connection
._load
_vca
_connection
_data
= Mock()
64 vca_connection
= Connection(AsyncMock())
65 vca_connection
._data
= ConnectionData(
67 "endpoints": ["1.2.3.4:17070"],
73 "lxd-credentials": "credentials",
74 "k8s-cloud": "k8s_cloud",
75 "k8s-credentials": "k8s_credentials",
77 "api-proxy": "api_proxy",
80 logging
.disable(logging
.CRITICAL
)
81 self
.libjuju
= Libjuju(vca_connection
, self
.loop
)
82 self
.loop
.run_until_complete(self
.libjuju
.disconnect())
85 @asynctest.mock
.patch("juju.controller.Controller.connect")
86 @asynctest.mock
.patch(
87 "juju.controller.Controller.api_endpoints",
88 new_callable
=asynctest
.CoroutineMock(return_value
=["127.0.0.1:17070"]),
90 class GetControllerTest(LibjujuTestCase
):
92 super(GetControllerTest
, self
).setUp()
94 def test_diff_endpoint(self
, mock_api_endpoints
, mock_connect
):
95 self
.libjuju
.endpoints
= []
96 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
97 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
99 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
102 mock_disconnect_controller
,
106 self
.libjuju
.endpoints
= []
108 mock_connect
.side_effect
= Exception()
110 with self
.assertRaises(JujuControllerFailedConnecting
):
111 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
112 self
.assertIsNone(controller
)
113 mock_disconnect_controller
.assert_called()
115 def test_same_endpoint_get_controller(self
, mock_api_endpoints
, mock_connect
):
116 self
.libjuju
.endpoints
= ["127.0.0.1:17070"]
117 controller
= self
.loop
.run_until_complete(self
.libjuju
.get_controller())
118 self
.assertIsInstance(controller
, juju
.controller
.Controller
)
121 class DisconnectTest(LibjujuTestCase
):
123 super(DisconnectTest
, self
).setUp()
125 @asynctest.mock
.patch("juju.model.Model.disconnect")
126 def test_disconnect_model(self
, mock_disconnect
):
127 self
.loop
.run_until_complete(self
.libjuju
.disconnect_model(juju
.model
.Model()))
128 mock_disconnect
.assert_called_once()
130 @asynctest.mock
.patch("juju.controller.Controller.disconnect")
131 def test_disconnect_controller(self
, mock_disconnect
):
132 self
.loop
.run_until_complete(
133 self
.libjuju
.disconnect_controller(juju
.controller
.Controller())
135 mock_disconnect
.assert_called_once()
138 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
139 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.model_exists")
140 @asynctest.mock
.patch("juju.controller.Controller.add_model")
141 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
142 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
143 class AddModelTest(LibjujuTestCase
):
145 super(AddModelTest
, self
).setUp()
147 def test_existing_model(
149 mock_disconnect_model
,
150 mock_disconnect_controller
,
155 mock_model_exists
.return_value
= True
157 # This should not raise an exception
158 self
.loop
.run_until_complete(self
.libjuju
.add_model("existing_model", "cloud"))
160 mock_disconnect_controller
.assert_called()
162 # TODO Check two job executing at the same time and one returning without doing anything.
164 def test_non_existing_model(
166 mock_disconnect_model
,
167 mock_disconnect_controller
,
172 mock_model_exists
.return_value
= False
173 mock_get_controller
.return_value
= juju
.controller
.Controller()
175 self
.loop
.run_until_complete(
176 self
.libjuju
.add_model("nonexisting_model", Mock())
179 mock_add_model
.assert_called_once()
180 mock_disconnect_controller
.assert_called()
181 mock_disconnect_model
.assert_called()
184 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
185 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
186 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
187 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
188 @asynctest.mock
.patch(
189 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
191 @asynctest.mock
.patch("juju.model.Model.get_action_status")
192 @asynctest.mock
.patch("juju.model.Model.get_action_output")
193 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_actions")
194 class GetExecutedActionsTest(LibjujuTestCase
):
196 super(GetExecutedActionsTest
, self
).setUp()
201 mock_get_action_output
,
202 mock_get_action_status
,
204 mock_disconnect_controller
,
205 mock_disconnect_model
,
209 mock_get_model
.return_value
= None
210 with self
.assertRaises(JujuError
):
211 self
.loop
.run_until_complete(self
.libjuju
.get_executed_actions("model"))
213 mock_get_controller
.assert_called_once()
214 mock_disconnect_controller
.assert_called_once()
215 mock_get_model
.assert_called_once()
216 mock_disconnect_model
.assert_not_called()
221 mock_get_action_output
,
222 mock_get_action_status
,
224 mock_disconnect_controller
,
225 mock_disconnect_model
,
229 mock_get_model
.return_value
= juju
.model
.Model()
230 mock_applications
.return_value
= {"existing_app"}
231 mock_get_actions
.return_value
= {"action_name": "description"}
232 mock_get_action_status
.return_value
= {"id": "status"}
233 mock_get_action_output
.return_value
= {"output": "completed"}
235 executed_actions
= self
.loop
.run_until_complete(
236 self
.libjuju
.get_executed_actions("model")
241 "action": "action_name",
243 "output": "completed",
246 self
.assertListEqual(expected_result
, executed_actions
)
247 self
.assertIsInstance(executed_actions
, list)
249 mock_get_controller
.assert_called_once()
250 mock_get_model
.assert_called_once()
251 mock_disconnect_controller
.assert_called_once()
252 mock_disconnect_model
.assert_called_once()
255 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
256 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
257 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
258 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
259 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
260 class GetApplicationConfigsTest(LibjujuTestCase
):
262 super(GetApplicationConfigsTest
, self
).setUp()
266 mock_get_application
,
267 mock_disconnect_controller
,
268 mock_disconnect_model
,
272 mock_get_model
.return_value
= None
273 with self
.assertRaises(JujuError
):
274 self
.loop
.run_until_complete(
275 self
.libjuju
.get_application_configs("model", "app")
278 mock_get_controller
.assert_called_once()
279 mock_disconnect_controller
.assert_called_once()
280 mock_get_model
.assert_called_once()
281 mock_disconnect_model
.assert_not_called()
285 mock_get_application
,
286 mock_disconnect_controller
,
287 mock_disconnect_model
,
291 mock_get_application
.return_value
= FakeApplication()
292 application_configs
= self
.loop
.run_until_complete(
293 self
.libjuju
.get_application_configs("model", "app")
296 self
.assertEqual(application_configs
, ["app_config"])
298 mock_get_controller
.assert_called_once()
299 mock_get_model
.assert_called_once()
300 mock_disconnect_controller
.assert_called_once()
301 mock_disconnect_model
.assert_called_once()
304 @asynctest.mock
.patch("juju.controller.Controller.get_model")
305 class GetModelTest(LibjujuTestCase
):
307 super(GetModelTest
, self
).setUp()
313 mock_get_model
.return_value
= juju
.model
.Model()
314 model
= self
.loop
.run_until_complete(
315 self
.libjuju
.get_model(juju
.controller
.Controller(), "model")
317 self
.assertIsInstance(model
, juju
.model
.Model
)
320 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
321 @asynctest.mock
.patch("juju.controller.Controller.list_models")
322 class ModelExistsTest(LibjujuTestCase
):
324 super(ModelExistsTest
, self
).setUp()
326 async def test_existing_model(
331 mock_list_models
.return_value
= ["existing_model"]
333 await self
.libjuju
.model_exists(
334 "existing_model", juju
.controller
.Controller()
338 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
339 async def test_no_controller(
341 mock_disconnect_controller
,
345 mock_list_models
.return_value
= ["existing_model"]
346 mock_get_controller
.return_value
= juju
.controller
.Controller()
347 self
.assertTrue(await self
.libjuju
.model_exists("existing_model"))
348 mock_disconnect_controller
.assert_called_once()
350 async def test_non_existing_model(
355 mock_list_models
.return_value
= ["existing_model"]
357 await self
.libjuju
.model_exists(
358 "not_existing_model", juju
.controller
.Controller()
363 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
364 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
365 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
366 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
367 @asynctest.mock
.patch("juju.model.Model.get_status")
368 class GetModelStatusTest(LibjujuTestCase
):
370 super(GetModelStatusTest
, self
).setUp()
375 mock_disconnect_controller
,
376 mock_disconnect_model
,
380 mock_get_model
.return_value
= juju
.model
.Model()
381 mock_get_status
.return_value
= {"status"}
383 status
= self
.loop
.run_until_complete(self
.libjuju
.get_model_status("model"))
385 mock_get_status
.assert_called_once()
386 mock_disconnect_controller
.assert_called_once()
387 mock_disconnect_model
.assert_called_once()
389 self
.assertEqual(status
, {"status"})
394 mock_disconnect_controller
,
395 mock_disconnect_model
,
399 mock_get_model
.return_value
= juju
.model
.Model()
400 mock_get_status
.side_effect
= Exception()
402 with self
.assertRaises(Exception):
403 status
= self
.loop
.run_until_complete(
404 self
.libjuju
.get_model_status("model")
407 mock_disconnect_controller
.assert_called_once()
408 mock_disconnect_model
.assert_called_once()
410 self
.assertIsNone(status
)
413 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
414 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
415 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
416 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
417 @asynctest.mock
.patch("juju.model.Model.get_machines")
418 @asynctest.mock
.patch("juju.model.Model.add_machine")
419 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
420 class CreateMachineTest(LibjujuTestCase
):
422 super(CreateMachineTest
, self
).setUp()
424 def test_existing_machine(
429 mock_disconnect_controller
,
430 mock_disconnect_model
,
434 mock_get_model
.return_value
= juju
.model
.Model()
435 mock_get_machines
.return_value
= {"existing_machine": FakeMachine()}
436 machine
, bool_res
= self
.loop
.run_until_complete(
437 self
.libjuju
.create_machine("model", "existing_machine")
440 self
.assertIsInstance(machine
, FakeMachine
)
441 self
.assertFalse(bool_res
)
443 mock_disconnect_controller
.assert_called()
444 mock_disconnect_model
.assert_called()
446 def test_non_existing_machine(
451 mock_disconnect_controller
,
452 mock_disconnect_model
,
458 mock_get_model
.return_value
= juju
.model
.Model()
459 with self
.assertRaises(JujuMachineNotFound
):
460 machine
, bool_res
= self
.loop
.run_until_complete(
461 self
.libjuju
.create_machine("model", "non_existing_machine")
463 self
.assertIsNone(machine
)
464 self
.assertIsNone(bool_res
)
466 mock_disconnect_controller
.assert_called()
467 mock_disconnect_model
.assert_called()
474 mock_disconnect_controller
,
475 mock_disconnect_model
,
479 mock_get_model
.return_value
= juju
.model
.Model()
480 mock_add_machine
.return_value
= FakeMachine()
482 machine
, bool_res
= self
.loop
.run_until_complete(
483 self
.libjuju
.create_machine("model")
486 self
.assertIsInstance(machine
, FakeMachine
)
487 self
.assertTrue(bool_res
)
489 mock_wait_for
.assert_called_once()
490 mock_add_machine
.assert_called_once()
492 mock_disconnect_controller
.assert_called()
493 mock_disconnect_model
.assert_called()
496 # TODO test provision machine
499 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
500 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
501 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
502 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
503 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
504 @asynctest.mock
.patch("juju.model.Model.deploy")
505 class DeployTest(LibjujuTestCase
):
507 super(DeployTest
, self
).setUp()
513 mock_disconnect_controller
,
514 mock_disconnect_model
,
518 mock_get_model
.return_value
= juju
.model
.Model()
519 self
.loop
.run_until_complete(
520 self
.libjuju
.deploy("cs:osm", "model", wait
=True, timeout
=0)
522 mock_deploy
.assert_called_once()
523 mock_wait_for_model
.assert_called_once()
524 mock_disconnect_controller
.assert_called_once()
525 mock_disconnect_model
.assert_called_once()
527 def test_deploy_no_wait(
531 mock_disconnect_controller
,
532 mock_disconnect_model
,
536 mock_get_model
.return_value
= juju
.model
.Model()
537 self
.loop
.run_until_complete(
538 self
.libjuju
.deploy("cs:osm", "model", wait
=False, timeout
=0)
540 mock_deploy
.assert_called_once()
541 mock_wait_for_model
.assert_not_called()
542 mock_disconnect_controller
.assert_called_once()
543 mock_disconnect_model
.assert_called_once()
545 def test_deploy_exception(
549 mock_disconnect_controller
,
550 mock_disconnect_model
,
554 mock_deploy
.side_effect
= Exception()
555 mock_get_model
.return_value
= juju
.model
.Model()
556 with self
.assertRaises(Exception):
557 self
.loop
.run_until_complete(self
.libjuju
.deploy("cs:osm", "model"))
558 mock_deploy
.assert_called_once()
559 mock_wait_for_model
.assert_not_called()
560 mock_disconnect_controller
.assert_called_once()
561 mock_disconnect_model
.assert_called_once()
564 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
565 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
566 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
567 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
568 @asynctest.mock
.patch(
569 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
571 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
572 @asynctest.mock
.patch("juju.model.Model.deploy")
573 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
574 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.create_machine")
575 class DeployCharmTest(LibjujuTestCase
):
577 super(DeployCharmTest
, self
).setUp()
579 def test_existing_app(
586 mock_disconnect_controller
,
587 mock_disconnect_model
,
591 mock_get_model
.return_value
= juju
.model
.Model()
592 mock_applications
.return_value
= {"existing_app"}
595 with self
.assertRaises(JujuApplicationExists
):
596 application
= self
.loop
.run_until_complete(
597 self
.libjuju
.deploy_charm(
604 self
.assertIsNone(application
)
606 mock_disconnect_controller
.assert_called()
607 mock_disconnect_model
.assert_called()
609 def test_non_existing_machine(
616 mock_disconnect_controller
,
617 mock_disconnect_model
,
621 mock_get_model
.return_value
= juju
.model
.Model()
622 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
624 with self
.assertRaises(JujuMachineNotFound
):
625 application
= self
.loop
.run_until_complete(
626 self
.libjuju
.deploy_charm(
634 self
.assertIsNone(application
)
636 mock_disconnect_controller
.assert_called()
637 mock_disconnect_model
.assert_called()
646 mock_disconnect_controller
,
647 mock_disconnect_model
,
651 mock_get_model
.return_value
= juju
.model
.Model()
652 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
653 mock_create_machine
.return_value
= (FakeMachine(), "other")
654 mock_deploy
.return_value
= FakeApplication()
655 application
= self
.loop
.run_until_complete(
656 self
.libjuju
.deploy_charm(
665 self
.assertIsInstance(application
, FakeApplication
)
667 mock_deploy
.assert_called_once()
668 mock_wait_for
.assert_called_once()
670 mock_create_machine
.assert_called_once()
672 mock_disconnect_controller
.assert_called()
673 mock_disconnect_model
.assert_called()
682 mock_disconnect_controller
,
683 mock_disconnect_model
,
687 mock_get_model
.return_value
= juju
.model
.Model()
688 mock_machines
.return_value
= {"existing_machine": FakeMachine()}
689 mock_deploy
.return_value
= FakeApplication()
690 application
= self
.loop
.run_until_complete(
691 self
.libjuju
.deploy_charm("app", "path", "model", "existing_machine")
694 self
.assertIsInstance(application
, FakeApplication
)
696 mock_deploy
.assert_called_once()
697 mock_wait_for
.assert_called_once()
699 mock_disconnect_controller
.assert_called()
700 mock_disconnect_model
.assert_called()
703 @asynctest.mock
.patch(
704 "juju.model.Model.applications", new_callable
=asynctest
.PropertyMock
706 class GetApplicationTest(LibjujuTestCase
):
708 super(GetApplicationTest
, self
).setUp()
710 def test_existing_application(
714 mock_applications
.return_value
= {"existing_app": "exists"}
715 model
= juju
.model
.Model()
716 result
= self
.libjuju
._get
_application
(model
, "existing_app")
717 self
.assertEqual(result
, "exists")
719 def test_non_existing_application(
723 mock_applications
.return_value
= {"existing_app": "exists"}
724 model
= juju
.model
.Model()
725 result
= self
.libjuju
._get
_application
(model
, "nonexisting_app")
726 self
.assertIsNone(result
)
729 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
730 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
731 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
732 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
733 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
734 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
735 @asynctest.mock
.patch("juju.model.Model.get_action_output")
736 @asynctest.mock
.patch("juju.model.Model.get_action_status")
737 class ExecuteActionTest(LibjujuTestCase
):
739 super(ExecuteActionTest
, self
).setUp()
741 def test_no_application(
743 mock_get_action_status
,
744 mock_get_action_output
,
746 mock__get_application
,
747 mock_disconnect_controller
,
748 mock_disconnect_model
,
752 mock__get_application
.return_value
= None
753 mock_get_model
.return_value
= juju
.model
.Model()
756 with self
.assertRaises(JujuApplicationNotFound
):
757 output
, status
= self
.loop
.run_until_complete(
758 self
.libjuju
.execute_action(
764 self
.assertIsNone(output
)
765 self
.assertIsNone(status
)
767 mock_disconnect_controller
.assert_called()
768 mock_disconnect_model
.assert_called()
772 mock_get_action_status
,
773 mock_get_action_output
,
775 mock__get_application
,
776 mock_disconnect_controller
,
777 mock_disconnect_model
,
782 mock_get_model
.return_value
= juju
.model
.Model()
783 mock__get_application
.return_value
= FakeApplication()
786 with self
.assertRaises(JujuActionNotFound
):
787 output
, status
= self
.loop
.run_until_complete(
788 self
.libjuju
.execute_action(
794 self
.assertIsNone(output
)
795 self
.assertIsNone(status
)
797 mock_disconnect_controller
.assert_called()
798 mock_disconnect_model
.assert_called()
800 @asynctest.mock
.patch("asyncio.sleep")
801 @asynctest.mock
.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
804 mock_is_leader_from_status
,
806 mock_get_action_status
,
807 mock_get_action_output
,
809 mock__get_application
,
810 mock_disconnect_controller
,
811 mock_disconnect_model
,
815 mock_get_model
.return_value
= juju
.model
.Model()
816 mock__get_application
.return_value
= FakeApplication()
817 mock_is_leader_from_status
.return_value
= False
820 with self
.assertRaises(JujuLeaderUnitNotFound
):
821 output
, status
= self
.loop
.run_until_complete(
822 self
.libjuju
.execute_action(
828 self
.assertIsNone(output
)
829 self
.assertIsNone(status
)
831 mock_disconnect_controller
.assert_called()
832 mock_disconnect_model
.assert_called()
834 def test_successful_exec(
836 mock_get_action_status
,
837 mock_get_action_output
,
839 mock__get_application
,
840 mock_disconnect_controller
,
841 mock_disconnect_model
,
845 mock_get_model
.return_value
= juju
.model
.Model()
846 mock__get_application
.return_value
= FakeApplication()
847 mock_get_action_output
.return_value
= "output"
848 mock_get_action_status
.return_value
= {"id": "status"}
849 output
, status
= self
.loop
.run_until_complete(
850 self
.libjuju
.execute_action("app", "model", "existing_action")
852 self
.assertEqual(output
, "output")
853 self
.assertEqual(status
, "status")
855 mock_wait_for
.assert_called_once()
857 mock_disconnect_controller
.assert_called()
858 mock_disconnect_model
.assert_called()
861 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
862 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
863 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
864 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
865 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
866 class GetActionTest(LibjujuTestCase
):
868 super(GetActionTest
, self
).setUp()
872 mock_get_application
,
873 mock_disconnect_controller
,
874 mock_disconnect_model
,
878 mock_get_application
.side_effect
= Exception()
880 with self
.assertRaises(Exception):
881 actions
= self
.loop
.run_until_complete(
882 self
.libjuju
.get_actions("app", "model")
885 self
.assertIsNone(actions
)
886 mock_disconnect_controller
.assert_called_once()
887 mock_disconnect_model
.assert_called_once()
891 mock_get_application
,
892 mock_disconnect_controller
,
893 mock_disconnect_model
,
897 mock_get_application
.return_value
= FakeApplication()
899 actions
= self
.loop
.run_until_complete(self
.libjuju
.get_actions("app", "model"))
901 self
.assertEqual(actions
, ["existing_action"])
903 mock_get_controller
.assert_called_once()
904 mock_get_model
.assert_called_once()
905 mock_disconnect_controller
.assert_called_once()
906 mock_disconnect_model
.assert_called_once()
909 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
910 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
911 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
912 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
913 @asynctest.mock
.patch("juju.application.Application.get_metrics")
914 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
915 class GetMetricsTest(LibjujuTestCase
):
917 super(GetMetricsTest
, self
).setUp()
919 def test_get_metrics_success(
921 mock_get_application
,
923 mock_disconnect_controller
,
924 mock_disconnect_model
,
928 mock_get_application
.return_value
= FakeApplication()
929 mock_get_model
.return_value
= juju
.model
.Model()
931 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
933 mock_disconnect_controller
.assert_called_once()
934 mock_disconnect_model
.assert_called_once()
936 def test_get_metrics_exception(
938 mock_get_application
,
940 mock_disconnect_controller
,
941 mock_disconnect_model
,
945 mock_get_model
.return_value
= juju
.model
.Model()
946 mock_get_metrics
.side_effect
= Exception()
947 with self
.assertRaises(Exception):
948 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("model", "app1"))
950 mock_disconnect_controller
.assert_called_once()
951 mock_disconnect_model
.assert_called_once()
953 def test_missing_args_exception(
955 mock_get_application
,
957 mock_disconnect_controller
,
958 mock_disconnect_model
,
962 mock_get_model
.return_value
= juju
.model
.Model()
964 with self
.assertRaises(Exception):
965 self
.loop
.run_until_complete(self
.libjuju
.get_metrics("", ""))
967 mock_get_controller
.assert_not_called()
968 mock_get_model
.assert_not_called()
969 mock_disconnect_controller
.assert_not_called()
970 mock_disconnect_model
.assert_not_called()
973 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
974 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
975 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
976 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
977 @asynctest.mock
.patch("juju.model.Model.add_relation")
978 class AddRelationTest(LibjujuTestCase
):
980 super(AddRelationTest
, self
).setUp()
982 @asynctest.mock
.patch("logging.Logger.warning")
987 mock_disconnect_controller
,
988 mock_disconnect_model
,
992 # TODO in libjuju.py should this fail only with a log message?
993 result
= {"error": "not found", "response": "response", "request-id": 1}
995 mock_get_model
.return_value
= juju
.model
.Model()
996 mock_add_relation
.side_effect
= JujuAPIError(result
)
998 self
.loop
.run_until_complete(
999 self
.libjuju
.add_relation(
1006 mock_warning
.assert_called_with("Relation not found: not found")
1007 mock_disconnect_controller
.assert_called_once()
1008 mock_disconnect_model
.assert_called_once()
1010 @asynctest.mock
.patch("logging.Logger.warning")
1011 def test_already_exists(
1015 mock_disconnect_controller
,
1016 mock_disconnect_model
,
1018 mock_get_controller
,
1020 # TODO in libjuju.py should this fail silently?
1021 result
= {"error": "already exists", "response": "response", "request-id": 1}
1023 mock_get_model
.return_value
= juju
.model
.Model()
1024 mock_add_relation
.side_effect
= JujuAPIError(result
)
1026 self
.loop
.run_until_complete(
1027 self
.libjuju
.add_relation(
1034 mock_warning
.assert_called_with("Relation already exists: already exists")
1035 mock_disconnect_controller
.assert_called_once()
1036 mock_disconnect_model
.assert_called_once()
1041 mock_disconnect_controller
,
1042 mock_disconnect_model
,
1044 mock_get_controller
,
1046 mock_get_model
.return_value
= juju
.model
.Model()
1047 result
= {"error": "", "response": "response", "request-id": 1}
1048 mock_add_relation
.side_effect
= JujuAPIError(result
)
1050 with self
.assertRaises(JujuAPIError
):
1051 self
.loop
.run_until_complete(
1052 self
.libjuju
.add_relation(
1059 mock_disconnect_controller
.assert_called_once()
1060 mock_disconnect_model
.assert_called_once()
1065 mock_disconnect_controller
,
1066 mock_disconnect_model
,
1068 mock_get_controller
,
1070 mock_get_model
.return_value
= juju
.model
.Model()
1072 self
.loop
.run_until_complete(
1073 self
.libjuju
.add_relation(
1080 mock_add_relation
.assert_called_with("app1:relation1", "app2:relation2")
1081 mock_disconnect_controller
.assert_called_once()
1082 mock_disconnect_model
.assert_called_once()
1087 mock_disconnect_controller
,
1088 mock_disconnect_model
,
1090 mock_get_controller
,
1092 mock_get_model
.return_value
= juju
.model
.Model()
1094 self
.loop
.run_until_complete(
1095 self
.libjuju
.add_relation(
1102 mock_add_relation
.assert_called_with("app1:relation1", "saas_name")
1103 mock_disconnect_controller
.assert_called_once()
1104 mock_disconnect_model
.assert_called_once()
1107 # TODO destroy_model testcase
1110 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1111 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1112 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1113 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1114 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1115 class DestroyApplicationTest(LibjujuTestCase
):
1117 super(DestroyApplicationTest
, self
).setUp()
1121 mock_get_controller
,
1123 mock_disconnect_controller
,
1124 mock_get_application
,
1125 mock_disconnect_model
,
1127 mock_get_application
.return_value
= FakeApplication()
1128 mock_get_model
.return_value
= None
1129 self
.loop
.run_until_complete(
1130 self
.libjuju
.destroy_application(
1136 mock_get_application
.assert_called()
1137 mock_disconnect_controller
.assert_called_once()
1138 mock_disconnect_model
.assert_called_once()
1140 def test_no_application(
1142 mock_get_controller
,
1144 mock_disconnect_controller
,
1145 mock_get_application
,
1146 mock_disconnect_model
,
1148 mock_get_model
.return_value
= None
1149 mock_get_application
.return_value
= None
1151 self
.loop
.run_until_complete(
1152 self
.libjuju
.destroy_application(
1158 mock_get_application
.assert_called()
1162 mock_get_controller
,
1164 mock_disconnect_controller
,
1165 mock_get_application
,
1166 mock_disconnect_model
,
1168 mock_get_application
.return_value
= FakeApplication
1169 mock_get_model
.return_value
= None
1171 with self
.assertRaises(Exception):
1172 self
.loop
.run_until_complete(
1173 self
.libjuju
.destroy_application(
1179 mock_get_application
.assert_called_once()
1182 # @asynctest.mock.patch("juju.model.Model.get_machines")
1183 # @asynctest.mock.patch("logging.Logger.debug")
1184 # class DestroyMachineTest(LibjujuTestCase):
1186 # super(DestroyMachineTest, self).setUp()
1188 # def test_success_manual_machine(
1189 # self, mock_debug, mock_get_machines,
1191 # mock_get_machines.side_effect = [
1192 # {"machine": FakeManualMachine()},
1193 # {"machine": FakeManualMachine()},
1196 # self.loop.run_until_complete(
1197 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1200 # asynctest.call("Waiting for machine machine is destroyed"),
1201 # asynctest.call("Machine destroyed: machine"),
1203 # mock_debug.assert_has_calls(calls)
1205 # def test_no_machine(
1206 # self, mock_debug, mock_get_machines,
1208 # mock_get_machines.return_value = {}
1209 # self.loop.run_until_complete(
1210 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1212 # mock_debug.assert_called_with("Machine not found: machine")
1215 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1216 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
1217 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1218 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1219 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
1220 class ConfigureApplicationTest(LibjujuTestCase
):
1222 super(ConfigureApplicationTest
, self
).setUp()
1226 mock_get_application
,
1227 mock_disconnect_controller
,
1228 mock_disconnect_model
,
1230 mock_get_controller
,
1233 mock_get_application
.return_value
= FakeApplication()
1235 self
.loop
.run_until_complete(
1236 self
.libjuju
.configure_application(
1242 mock_get_application
.assert_called_once()
1243 mock_disconnect_controller
.assert_called_once()
1244 mock_disconnect_model
.assert_called_once()
1248 mock_get_application
,
1249 mock_disconnect_controller
,
1250 mock_disconnect_model
,
1252 mock_get_controller
,
1255 mock_get_application
.side_effect
= Exception()
1257 with self
.assertRaises(Exception):
1258 self
.loop
.run_until_complete(
1259 self
.libjuju
.configure_application(
1265 mock_disconnect_controller
.assert_called_once()
1266 mock_disconnect_model
.assert_called_once()
1268 def test_controller_exception(
1270 mock_get_application
,
1271 mock_disconnect_controller
,
1272 mock_disconnect_model
,
1274 mock_get_controller
,
1277 result
= {"error": "not found", "response": "response", "request-id": 1}
1279 mock_get_controller
.side_effect
= JujuAPIError(result
)
1281 with self
.assertRaises(JujuAPIError
):
1282 self
.loop
.run_until_complete(
1283 self
.libjuju
.configure_application(
1289 mock_get_model
.assert_not_called()
1290 mock_disconnect_controller
.assert_not_called()
1291 mock_disconnect_model
.assert_not_called()
1293 def test_get_model_exception(
1295 mock_get_application
,
1296 mock_disconnect_controller
,
1297 mock_disconnect_model
,
1299 mock_get_controller
,
1302 result
= {"error": "not found", "response": "response", "request-id": 1}
1303 mock_get_model
.side_effect
= JujuAPIError(result
)
1305 with self
.assertRaises(JujuAPIError
):
1306 self
.loop
.run_until_complete(
1307 self
.libjuju
.configure_application(
1313 mock_get_model
.assert_called_once()
1314 mock_disconnect_controller
.assert_called_once()
1315 mock_disconnect_model
.assert_not_called()
1318 # TODO _get_api_endpoints_db test case
1319 # TODO _update_api_endpoints_db test case
1320 # TODO healthcheck test case
1323 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1324 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1325 @asynctest.mock
.patch("juju.controller.Controller.list_models")
1326 class ListModelsTest(LibjujuTestCase
):
1328 super(ListModelsTest
, self
).setUp()
1330 def test_containing(
1333 mock_disconnect_controller
,
1334 mock_get_controller
,
1336 mock_get_controller
.return_value
= juju
.controller
.Controller()
1337 mock_list_models
.return_value
= ["existingmodel"]
1338 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("existing"))
1340 mock_disconnect_controller
.assert_called_once()
1341 self
.assertEquals(models
, ["existingmodel"])
1343 def test_not_containing(
1346 mock_disconnect_controller
,
1347 mock_get_controller
,
1349 mock_get_controller
.return_value
= juju
.controller
.Controller()
1350 mock_list_models
.return_value
= ["existingmodel", "model"]
1351 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models("mdl"))
1353 mock_disconnect_controller
.assert_called_once()
1354 self
.assertEquals(models
, [])
1356 def test_no_contains_arg(
1359 mock_disconnect_controller
,
1360 mock_get_controller
,
1362 mock_get_controller
.return_value
= juju
.controller
.Controller()
1363 mock_list_models
.return_value
= ["existingmodel", "model"]
1364 models
= self
.loop
.run_until_complete(self
.libjuju
.list_models())
1366 mock_disconnect_controller
.assert_called_once()
1367 self
.assertEquals(models
, ["existingmodel", "model"])
1370 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.list_models")
1371 class ModelsExistTest(LibjujuTestCase
):
1373 super(ModelsExistTest
, self
).setUp()
1375 def test_model_names_none(self
, mock_list_models
):
1376 mock_list_models
.return_value
= []
1377 with self
.assertRaises(Exception):
1378 self
.loop
.run_until_complete(self
.libjuju
.models_exist(None))
1380 def test_model_names_empty(self
, mock_list_models
):
1381 mock_list_models
.return_value
= []
1382 with self
.assertRaises(Exception):
1383 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1384 self
.libjuju
.models_exist([])
1387 def test_model_names_not_existing(self
, mock_list_models
):
1388 mock_list_models
.return_value
= ["prometheus", "grafana"]
1389 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1390 self
.libjuju
.models_exist(["prometheus2", "grafana"])
1392 self
.assertFalse(exist
)
1393 self
.assertEqual(non_existing_models
, ["prometheus2"])
1395 def test_model_names_exist(self
, mock_list_models
):
1396 mock_list_models
.return_value
= ["prometheus", "grafana"]
1397 (exist
, non_existing_models
) = self
.loop
.run_until_complete(
1398 self
.libjuju
.models_exist(["prometheus", "grafana"])
1400 self
.assertTrue(exist
)
1401 self
.assertEqual(non_existing_models
, [])
1404 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1405 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1406 @asynctest.mock
.patch("juju.controller.Controller.list_offers")
1407 class ListOffers(LibjujuTestCase
):
1409 super(ListOffers
, self
).setUp()
1411 def test_disconnect_controller(
1414 mock_disconnect_controller
,
1415 mock_get_controller
,
1417 mock_get_controller
.return_value
= juju
.controller
.Controller()
1418 mock_list_offers
.side_effect
= Exception()
1419 with self
.assertRaises(Exception):
1420 self
.loop
.run_until_complete(self
.libjuju
._list
_offers
("model"))
1421 mock_disconnect_controller
.assert_called_once()
1423 def test_empty_list(
1426 mock_disconnect_controller
,
1427 mock_get_controller
,
1429 mock_get_controller
.return_value
= juju
.controller
.Controller()
1430 offer_results
= Mock()
1431 offer_results
.results
= []
1432 mock_list_offers
.return_value
= offer_results
1433 offers
= self
.loop
.run_until_complete(self
.libjuju
._list
_offers
("model"))
1434 self
.assertEqual(offers
, [])
1435 mock_disconnect_controller
.assert_called_once()
1437 def test_non_empty_list(
1440 mock_disconnect_controller
,
1441 mock_get_controller
,
1443 mock_get_controller
.return_value
= juju
.controller
.Controller()
1445 offer_results
= Mock()
1446 offer_results
.results
= [offer
]
1447 mock_list_offers
.return_value
= offer_results
1448 offers
= self
.loop
.run_until_complete(self
.libjuju
._list
_offers
("model"))
1449 self
.assertEqual(offers
, [offer
])
1450 mock_disconnect_controller
.assert_called_once()
1452 def test_matching_offer_name(
1455 mock_disconnect_controller
,
1456 mock_get_controller
,
1458 mock_get_controller
.return_value
= juju
.controller
.Controller()
1460 offer_1
.offer_name
= "offer1"
1462 offer_2
.offer_name
= "offer2"
1463 offer_results
= Mock()
1464 offer_results
.results
= [offer_1
, offer_2
]
1465 mock_list_offers
.return_value
= offer_results
1466 offers
= self
.loop
.run_until_complete(
1467 self
.libjuju
._list
_offers
("model", offer_name
="offer2")
1469 self
.assertEqual(offers
, [offer_2
])
1470 mock_disconnect_controller
.assert_called_once()
1472 def test_not_matching_offer_name(
1475 mock_disconnect_controller
,
1476 mock_get_controller
,
1478 mock_get_controller
.return_value
= juju
.controller
.Controller()
1480 offer_1
.offer_name
= "offer1"
1482 offer_2
.offer_name
= "offer2"
1483 offer_results
= Mock()
1484 offer_results
.results
= [offer_1
, offer_2
]
1485 mock_list_offers
.return_value
= offer_results
1486 offers
= self
.loop
.run_until_complete(
1487 self
.libjuju
._list
_offers
("model", offer_name
="offer3")
1489 self
.assertEqual(offers
, [])
1490 mock_disconnect_controller
.assert_called_once()
1493 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1494 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1495 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1496 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1497 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._list_offers")
1498 @asynctest.mock
.patch("juju.model.Model.create_offer")
1499 class OfferTest(LibjujuTestCase
):
1501 super(OfferTest
, self
).setUp()
1507 mock_disconnect_controller
,
1508 mock_disconnect_model
,
1510 mock_get_controller
,
1512 controller
= juju
.controller
.Controller()
1513 model
= juju
.model
.Model()
1514 mock_get_controller
.return_value
= controller
1515 mock_get_model
.return_value
= model
1516 endpoint
= RelationEndpoint("model.app-name.0", "vca", "endpoint")
1517 self
.loop
.run_until_complete(self
.libjuju
.offer(endpoint
))
1518 mock_create_offer
.assert_called_with(
1519 "app-name:endpoint", offer_name
="app-name-endpoint"
1521 mock_disconnect_model
.assert_called_once_with(model
)
1522 mock_disconnect_controller
.assert_called_once_with(controller
)
1524 def test_offer_exception(
1528 mock_disconnect_controller
,
1529 mock_disconnect_model
,
1531 mock_get_controller
,
1533 controller
= juju
.controller
.Controller()
1534 model
= juju
.model
.Model()
1535 mock_get_controller
.return_value
= controller
1536 mock_get_model
.return_value
= model
1537 mock__list_offers
.return_value
= []
1538 endpoint
= RelationEndpoint("model.app-name.0", "vca", "endpoint")
1539 with self
.assertRaises(Exception):
1540 self
.loop
.run_until_complete(self
.libjuju
.offer(endpoint
))
1541 mock_create_offer
.assert_called_with(
1542 "app-name:endpoint", offer_name
="app-name-endpoint"
1544 mock_disconnect_model
.assert_called_once_with(model
)
1545 mock_disconnect_controller
.assert_called_once_with(controller
)
1548 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1549 @asynctest.mock
.patch("juju.controller.Controller.get_model")
1550 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
1551 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1552 @asynctest.mock
.patch("juju.model.Model.consume")
1553 class ConsumeTest(LibjujuTestCase
):
1555 self
.offer_url
= "admin/model.offer_name"
1556 super(ConsumeTest
, self
).setUp()
1557 self
.provider_libjuju
= self
.libjuju
1562 mock_disconnect_controller
,
1563 mock_disconnect_model
,
1565 mock_get_controller
,
1567 self_controller
= juju
.controller
.Controller()
1568 provider_controller
= juju
.controller
.Controller()
1569 mock_get_controller
.side_effect
= [self_controller
, provider_controller
]
1570 mock_get_model
.return_value
= juju
.model
.Model()
1572 self
.loop
.run_until_complete(
1573 self
.libjuju
.consume(
1575 Offer(self
.offer_url
, vca_id
="vca-id"),
1576 self
.provider_libjuju
,
1579 mock_consume
.assert_called_once_with(
1580 "admin/model.offer_name",
1581 application_alias
="offer_name-model-vca-id",
1582 controller
=provider_controller
,
1584 mock_disconnect_model
.assert_called_once()
1585 self
.assertEqual(mock_disconnect_controller
.call_count
, 2)
1587 def test_parsing_error_exception(
1590 mock_disconnect_controller
,
1591 mock_disconnect_model
,
1593 mock_get_controller
,
1595 mock_get_controller
.return_value
= juju
.controller
.Controller()
1596 mock_get_model
.return_value
= juju
.model
.Model()
1597 mock_consume
.side_effect
= juju
.offerendpoints
.ParseError("")
1599 with self
.assertRaises(juju
.offerendpoints
.ParseError
):
1600 self
.loop
.run_until_complete(
1601 self
.libjuju
.consume(
1602 "model_name", Offer(self
.offer_url
), self
.provider_libjuju
1605 mock_consume
.assert_called_once()
1606 mock_disconnect_model
.assert_called_once()
1607 self
.assertEqual(mock_disconnect_controller
.call_count
, 2)
1609 def test_juju_error_exception(
1612 mock_disconnect_controller
,
1613 mock_disconnect_model
,
1615 mock_get_controller
,
1617 mock_get_controller
.return_value
= juju
.controller
.Controller()
1618 mock_get_model
.return_value
= juju
.model
.Model()
1619 mock_consume
.side_effect
= juju
.errors
.JujuError("")
1621 with self
.assertRaises(juju
.errors
.JujuError
):
1622 self
.loop
.run_until_complete(
1623 self
.libjuju
.consume(
1624 "model_name", Offer(self
.offer_url
), self
.provider_libjuju
1627 mock_consume
.assert_called_once()
1628 mock_disconnect_model
.assert_called_once()
1629 self
.assertEqual(mock_disconnect_controller
.call_count
, 2)
1631 def test_juju_api_error_exception(
1634 mock_disconnect_controller
,
1635 mock_disconnect_model
,
1637 mock_get_controller
,
1639 mock_get_controller
.return_value
= juju
.controller
.Controller()
1640 mock_get_model
.return_value
= juju
.model
.Model()
1641 mock_consume
.side_effect
= juju
.errors
.JujuAPIError(
1642 {"error": "", "response": "", "request-id": ""}
1645 with self
.assertRaises(juju
.errors
.JujuAPIError
):
1646 self
.loop
.run_until_complete(
1647 self
.libjuju
.consume(
1648 "model_name", Offer(self
.offer_url
), self
.provider_libjuju
1651 mock_consume
.assert_called_once()
1652 mock_disconnect_model
.assert_called_once()
1653 self
.assertEqual(mock_disconnect_controller
.call_count
, 2)
1656 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1657 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.add_cloud")
1658 class AddK8sTest(LibjujuTestCase
):
1660 super(AddK8sTest
, self
).setUp()
1662 rbac_id
= generate_rbac_id()
1664 client_cert_data
= "cert"
1665 configuration
= kubernetes
.client
.configuration
.Configuration()
1666 storage_class
= "storage_class"
1667 credential_name
= name
1669 self
._add
_k
8s
_args
= {
1673 "client_cert_data": client_cert_data
,
1674 "configuration": configuration
,
1675 "storage_class": storage_class
,
1676 "credential_name": credential_name
,
1679 def test_add_k8s(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1680 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1681 mock_add_cloud
.assert_called_once()
1682 mock_get_k8s_cloud_credential
.assert_called_once()
1684 def test_add_k8s_exception(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1685 mock_add_cloud
.side_effect
= Exception()
1686 with self
.assertRaises(Exception):
1687 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1688 mock_add_cloud
.assert_called_once()
1689 mock_get_k8s_cloud_credential
.assert_called_once()
1691 def test_add_k8s_missing_name(self
, mock_add_cloud
, mock_get_k8s_cloud_credential
):
1692 self
._add
_k
8s
_args
["name"] = ""
1693 with self
.assertRaises(Exception):
1694 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1695 mock_add_cloud
.assert_not_called()
1697 def test_add_k8s_missing_storage_name(
1698 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1700 self
._add
_k
8s
_args
["storage_class"] = ""
1701 with self
.assertRaises(Exception):
1702 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1703 mock_add_cloud
.assert_not_called()
1705 def test_add_k8s_missing_configuration_keys(
1706 self
, mock_add_cloud
, mock_get_k8s_cloud_credential
1708 self
._add
_k
8s
_args
["configuration"] = None
1709 with self
.assertRaises(Exception):
1710 self
.loop
.run_until_complete(self
.libjuju
.add_k8s(**self
._add
_k
8s
_args
))
1711 mock_add_cloud
.assert_not_called()
1714 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1715 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1716 @asynctest.mock
.patch("juju.controller.Controller.add_cloud")
1717 @asynctest.mock
.patch("juju.controller.Controller.add_credential")
1718 class AddCloudTest(LibjujuTestCase
):
1720 super(AddCloudTest
, self
).setUp()
1721 self
.cloud
= juju
.client
.client
.Cloud()
1722 self
.credential
= juju
.client
.client
.CloudCredential()
1724 def test_add_cloud_with_credential(
1726 mock_add_credential
,
1728 mock_disconnect_controller
,
1729 mock_get_controller
,
1731 mock_get_controller
.return_value
= juju
.controller
.Controller()
1733 cloud
= self
.loop
.run_until_complete(
1734 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1736 self
.assertEqual(cloud
, self
.cloud
)
1737 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1738 mock_add_credential
.assert_called_once_with(
1739 "cloud", credential
=self
.credential
, cloud
="cloud"
1741 mock_disconnect_controller
.assert_called_once()
1743 def test_add_cloud_no_credential(
1745 mock_add_credential
,
1747 mock_disconnect_controller
,
1748 mock_get_controller
,
1750 mock_get_controller
.return_value
= juju
.controller
.Controller()
1752 cloud
= self
.loop
.run_until_complete(
1753 self
.libjuju
.add_cloud("cloud", self
.cloud
)
1755 self
.assertEqual(cloud
, self
.cloud
)
1756 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1757 mock_add_credential
.assert_not_called()
1758 mock_disconnect_controller
.assert_called_once()
1760 def test_add_cloud_exception(
1762 mock_add_credential
,
1764 mock_disconnect_controller
,
1765 mock_get_controller
,
1767 mock_get_controller
.return_value
= juju
.controller
.Controller()
1768 mock_add_cloud
.side_effect
= Exception()
1769 with self
.assertRaises(Exception):
1770 self
.loop
.run_until_complete(
1771 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1774 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1775 mock_add_credential
.assert_not_called()
1776 mock_disconnect_controller
.assert_called_once()
1778 def test_add_credential_exception(
1780 mock_add_credential
,
1782 mock_disconnect_controller
,
1783 mock_get_controller
,
1785 mock_get_controller
.return_value
= juju
.controller
.Controller()
1786 mock_add_credential
.side_effect
= Exception()
1787 with self
.assertRaises(Exception):
1788 self
.loop
.run_until_complete(
1789 self
.libjuju
.add_cloud("cloud", self
.cloud
, credential
=self
.credential
)
1792 mock_add_cloud
.assert_called_once_with("cloud", self
.cloud
)
1793 mock_add_credential
.assert_called_once_with(
1794 "cloud", credential
=self
.credential
, cloud
="cloud"
1796 mock_disconnect_controller
.assert_called_once()
1799 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
1800 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1801 @asynctest.mock
.patch("juju.controller.Controller.remove_cloud")
1802 class RemoveCloudTest(LibjujuTestCase
):
1804 super(RemoveCloudTest
, self
).setUp()
1806 def test_remove_cloud(
1809 mock_disconnect_controller
,
1810 mock_get_controller
,
1812 mock_get_controller
.return_value
= juju
.controller
.Controller()
1814 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1815 mock_remove_cloud
.assert_called_once_with("cloud")
1816 mock_disconnect_controller
.assert_called_once()
1818 def test_remove_cloud_exception(
1821 mock_disconnect_controller
,
1822 mock_get_controller
,
1824 mock_get_controller
.return_value
= juju
.controller
.Controller()
1825 mock_remove_cloud
.side_effect
= Exception()
1827 with self
.assertRaises(Exception):
1828 self
.loop
.run_until_complete(self
.libjuju
.remove_cloud("cloud"))
1829 mock_remove_cloud
.assert_called_once_with("cloud")
1830 mock_disconnect_controller
.assert_called_once()
1833 @asynctest.mock
.patch("kubernetes.client.configuration.Configuration")
1834 class GetK8sCloudCredentials(LibjujuTestCase
):
1836 super(GetK8sCloudCredentials
, self
).setUp()
1837 self
.cert_data
= "cert"
1838 self
.token
= "token"
1840 @asynctest.mock
.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1841 def test_not_supported(self
, mock_exception
, mock_configuration
):
1842 mock_configuration
.username
= ""
1843 mock_configuration
.password
= ""
1844 mock_configuration
.ssl_ca_cert
= None
1845 mock_configuration
.cert_file
= None
1846 mock_configuration
.key_file
= None
1847 exception_raised
= False
1849 self
.cert_data
= None
1851 _
= self
.libjuju
.get_k8s_cloud_credential(
1856 except JujuInvalidK8sConfiguration
as e
:
1857 exception_raised
= True
1860 "authentication method not supported",
1862 self
.assertTrue(exception_raised
)
1864 def test_user_pass(self
, mock_configuration
):
1865 mock_configuration
.username
= "admin"
1866 mock_configuration
.password
= "admin"
1867 mock_configuration
.ssl_ca_cert
= None
1868 mock_configuration
.cert_file
= None
1869 mock_configuration
.key_file
= None
1871 self
.cert_data
= None
1872 credential
= self
.libjuju
.get_k8s_cloud_credential(
1879 juju
.client
._definitions
.CloudCredential(
1880 attrs
={"username": "admin", "password": "admin"}, auth_type
="userpass"
1884 def test_user_pass_with_cert(self
, mock_configuration
):
1885 mock_configuration
.username
= "admin"
1886 mock_configuration
.password
= "admin"
1887 mock_configuration
.ssl_ca_cert
= None
1888 mock_configuration
.cert_file
= None
1889 mock_configuration
.key_file
= None
1891 credential
= self
.libjuju
.get_k8s_cloud_credential(
1898 juju
.client
._definitions
.CloudCredential(
1900 "ClientCertificateData": self
.cert_data
,
1901 "username": "admin",
1902 "password": "admin",
1904 auth_type
="userpasswithcert",
1908 def test_user_no_pass(self
, mock_configuration
):
1909 mock_configuration
.username
= "admin"
1910 mock_configuration
.password
= ""
1911 mock_configuration
.ssl_ca_cert
= None
1912 mock_configuration
.cert_file
= None
1913 mock_configuration
.key_file
= None
1915 self
.cert_data
= None
1916 with patch
.object(self
.libjuju
.log
, "debug") as mock_debug
:
1917 credential
= self
.libjuju
.get_k8s_cloud_credential(
1924 juju
.client
._definitions
.CloudCredential(
1925 attrs
={"username": "admin", "password": ""}, auth_type
="userpass"
1928 mock_debug
.assert_called_once_with(
1929 "credential for user admin has empty password"
1932 def test_cert(self
, mock_configuration
):
1933 mock_configuration
.username
= ""
1934 mock_configuration
.password
= ""
1935 mock_configuration
.api_key
= {"authorization": "Bearer Token"}
1936 ssl_ca_cert
= tempfile
.NamedTemporaryFile()
1937 with
open(ssl_ca_cert
.name
, "w") as ssl_ca_cert_file
:
1938 ssl_ca_cert_file
.write("cacert")
1939 mock_configuration
.ssl_ca_cert
= ssl_ca_cert
.name
1940 mock_configuration
.cert_file
= None
1941 mock_configuration
.key_file
= None
1942 credential
= self
.libjuju
.get_k8s_cloud_credential(
1949 juju
.client
._definitions
.CloudCredential(
1950 attrs
={"ClientCertificateData": self
.cert_data
, "Token": self
.token
},
1951 auth_type
="certificate",
1955 # TODO: Fix this test when oauth authentication is supported
1956 # def test_oauth2(self, mock_configuration):
1957 # mock_configuration.username = ""
1958 # mock_configuration.password = ""
1959 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1960 # key = tempfile.NamedTemporaryFile()
1961 # with open(key.name, "w") as key_file:
1962 # key_file.write("key")
1963 # mock_configuration.ssl_ca_cert = None
1964 # mock_configuration.cert_file = None
1965 # mock_configuration.key_file = key.name
1966 # credential = self.libjuju.get_k8s_cloud_credential(
1967 # mock_configuration,
1973 # juju.client._definitions.CloudCredential(
1974 # attrs={"ClientKeyData": "key", "Token": "Token"},
1975 # auth_type="oauth2",
1979 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1980 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1981 # mock_configuration.username = ""
1982 # mock_configuration.password = ""
1983 # key = tempfile.NamedTemporaryFile()
1984 # with open(key.name, "w") as key_file:
1985 # key_file.write("key")
1986 # mock_configuration.ssl_ca_cert = None
1987 # mock_configuration.cert_file = None
1988 # mock_configuration.key_file = key.name
1989 # exception_raised = False
1991 # _ = self.libjuju.get_k8s_cloud_credential(
1992 # mock_configuration,
1996 # except JujuInvalidK8sConfiguration as e:
1997 # exception_raised = True
2000 # "missing token for auth type oauth2",
2002 # self.assertTrue(exception_raised)
2004 def test_exception_cannot_set_token_and_userpass(self
, mock_configuration
):
2005 mock_configuration
.username
= "admin"
2006 mock_configuration
.password
= "pass"
2007 mock_configuration
.api_key
= {"authorization": "No_bearer_token"}
2008 mock_configuration
.ssl_ca_cert
= None
2009 mock_configuration
.cert_file
= None
2010 mock_configuration
.key_file
= None
2011 exception_raised
= False
2013 _
= self
.libjuju
.get_k8s_cloud_credential(
2018 except JujuInvalidK8sConfiguration
as e
:
2019 exception_raised
= True
2022 "Cannot set both token and user/pass",
2024 self
.assertTrue(exception_raised
)
2027 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2028 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2029 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2030 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2031 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2032 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
2033 class ScaleApplicationTest(LibjujuTestCase
):
2035 super(ScaleApplicationTest
, self
).setUp()
2037 @asynctest.mock
.patch("asyncio.sleep")
2038 def test_scale_application(
2041 mock_wait_for_model
,
2042 mock_disconnect_controller
,
2043 mock_disconnect_model
,
2044 mock_get_application
,
2046 mock_get_controller
,
2048 mock_get_model
.return_value
= juju
.model
.Model()
2049 mock_get_application
.return_value
= FakeApplication()
2050 self
.loop
.run_until_complete(self
.libjuju
.scale_application("model", "app", 2))
2051 mock_wait_for_model
.assert_called_once()
2052 mock_disconnect_controller
.assert_called_once()
2053 mock_disconnect_model
.assert_called_once()
2055 def test_no_application(
2058 mock_disconnect_controller
,
2059 mock_disconnect_model
,
2060 mock_get_application
,
2062 mock_get_controller
,
2064 mock_get_application
.return_value
= None
2065 mock_get_model
.return_value
= juju
.model
.Model()
2066 with self
.assertRaises(JujuApplicationNotFound
):
2067 self
.loop
.run_until_complete(
2068 self
.libjuju
.scale_application("model", "app", 2)
2070 mock_disconnect_controller
.assert_called()
2071 mock_disconnect_model
.assert_called()
2076 mock_disconnect_controller
,
2077 mock_disconnect_model
,
2078 mock_get_application
,
2080 mock_get_controller
,
2082 mock_get_model
.return_value
= None
2083 mock_get_application
.return_value
= FakeApplication()
2084 with self
.assertRaises(Exception):
2085 self
.loop
.run_until_complete(
2086 self
.libjuju
.scale_application("model", "app", 2, total_timeout
=0)
2088 mock_disconnect_controller
.assert_called_once()
2091 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2092 class GetUnitNumberTest(LibjujuTestCase
):
2094 super(GetUnitNumberTest
, self
).setUp()
2096 def test_successful_get_unit_number(
2098 mock_get_applications
,
2100 mock_get_applications
.return_value
= FakeApplication()
2101 model
= juju
.model
.Model()
2102 result
= self
.libjuju
._get
_application
_count
(model
, "app")
2103 self
.assertEqual(result
, 2)
2105 def test_non_existing_application(
2107 mock_get_applications
,
2109 mock_get_applications
.return_value
= None
2110 model
= juju
.model
.Model()
2111 result
= self
.libjuju
._get
_application
_count
(model
, "app")
2112 self
.assertEqual(result
, None)
2115 @asynctest.mock
.patch("juju.model.Model.machines", new_callable
=asynctest
.PropertyMock
)
2116 class GetMachineInfoTest(LibjujuTestCase
):
2118 super(GetMachineInfoTest
, self
).setUp()
2120 def test_successful(
2124 machine_id
= "existing_machine"
2125 model
= juju
.model
.Model()
2126 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2127 machine
, series
= self
.libjuju
._get
_machine
_info
(
2128 machine_id
=machine_id
,
2131 self
.assertIsNotNone(machine
, series
)
2137 machine_id
= "not_existing_machine"
2138 machine
= series
= None
2139 model
= juju
.model
.Model()
2140 mock_machines
.return_value
= {"existing_machine": FakeManualMachine()}
2141 with self
.assertRaises(JujuMachineNotFound
):
2142 machine
, series
= self
.libjuju
._get
_machine
_info
(
2143 machine_id
=machine_id
,
2146 self
.assertIsNone(machine
, series
)
2149 class GetUnitTest(LibjujuTestCase
):
2151 super(GetUnitTest
, self
).setUp()
2153 def test_successful(self
):
2154 result
= self
.libjuju
._get
_unit
(FakeApplication(), "existing_machine_id")
2155 self
.assertIsInstance(result
, FakeUnit
)
2157 def test_return_none(self
):
2158 result
= self
.libjuju
._get
_unit
(FakeApplication(), "not_existing_machine_id")
2159 self
.assertIsNone(result
)
2162 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2163 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2164 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2165 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2166 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2167 class CheckApplicationExists(LibjujuTestCase
):
2169 super(CheckApplicationExists
, self
).setUp()
2171 def test_successful(
2173 mock_get_application
,
2174 mock_disconnect_controller
,
2175 mock_disconnect_model
,
2177 mock_get_controller
,
2179 mock_get_model
.return_value
= juju
.model
.Model()
2180 mock_get_application
.return_value
= FakeApplication()
2181 result
= self
.loop
.run_until_complete(
2182 self
.libjuju
.check_application_exists(
2187 self
.assertEqual(result
, True)
2189 mock_get_application
.assert_called_once()
2190 mock_get_controller
.assert_called_once()
2191 mock_get_model
.assert_called_once()
2192 mock_disconnect_controller
.assert_called_once()
2193 mock_disconnect_model
.assert_called_once()
2195 def test_no_application(
2197 mock_get_application
,
2198 mock_disconnect_controller
,
2199 mock_disconnect_model
,
2201 mock_get_controller
,
2203 mock_get_model
.return_value
= juju
.model
.Model()
2204 mock_get_application
.return_value
= None
2205 result
= self
.loop
.run_until_complete(
2206 self
.libjuju
.check_application_exists(
2211 self
.assertEqual(result
, False)
2213 mock_get_application
.assert_called_once()
2214 mock_get_controller
.assert_called_once()
2215 mock_get_model
.assert_called_once()
2216 mock_disconnect_controller
.assert_called_once()
2217 mock_disconnect_model
.assert_called_once()
2220 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2221 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2222 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2223 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2224 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2225 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_machine_info")
2226 class AddUnitTest(LibjujuTestCase
):
2228 super(AddUnitTest
, self
).setUp()
2230 @asynctest.mock
.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
2231 @asynctest.mock
.patch("asyncio.sleep")
2232 def test_successful(
2236 mock_get_machine_info
,
2237 mock_get_application
,
2238 mock_disconnect_controller
,
2239 mock_disconnect_model
,
2241 mock_get_controller
,
2243 mock_get_model
.return_value
= juju
.model
.Model()
2244 mock_get_application
.return_value
= FakeApplication()
2245 mock_get_machine_info
.return_value
= FakeMachine(), "series"
2246 self
.loop
.run_until_complete(
2247 self
.libjuju
.add_unit(
2254 mock_wait_for
.assert_called_once()
2255 mock_get_application
.assert_called_once()
2256 mock_get_controller
.assert_called_once()
2257 mock_get_model
.assert_called_once()
2258 mock_disconnect_controller
.assert_called_once()
2259 mock_disconnect_model
.assert_called_once()
2263 mock_get_machine_info
,
2264 mock_get_application
,
2265 mock_disconnect_controller
,
2266 mock_disconnect_model
,
2268 mock_get_controller
,
2270 mock_get_model
.return_value
= juju
.model
.Model()
2271 mock_get_application
.return_value
= None
2272 with self
.assertRaises(JujuApplicationNotFound
):
2273 self
.loop
.run_until_complete(
2274 self
.libjuju
.add_unit(
2281 mock_get_application
.assert_called_once()
2282 mock_get_controller
.assert_called_once()
2283 mock_get_model
.assert_called_once()
2284 mock_disconnect_controller
.assert_called_once()
2285 mock_disconnect_model
.assert_called_once()
2287 def test_no_machine(
2289 mock_get_machine_info
,
2290 mock_get_application
,
2291 mock_disconnect_controller
,
2292 mock_disconnect_model
,
2294 mock_get_controller
,
2296 mock_get_model
.return_value
= juju
.model
.Model()
2297 mock_get_application
.return_value
= FakeApplication()
2298 mock_get_machine_info
.side_effect
= JujuMachineNotFound()
2299 with self
.assertRaises(JujuMachineNotFound
):
2300 self
.loop
.run_until_complete(
2301 self
.libjuju
.add_unit(
2308 mock_get_application
.assert_called_once()
2309 mock_get_controller
.assert_called_once()
2310 mock_get_model
.assert_called_once()
2311 mock_disconnect_controller
.assert_called_once()
2312 mock_disconnect_model
.assert_called_once()
2315 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_controller")
2316 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.get_model")
2317 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_model")
2318 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2319 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_application")
2320 @asynctest.mock
.patch("n2vc.libjuju.Libjuju._get_unit")
2321 class DestroyUnitTest(LibjujuTestCase
):
2323 super(DestroyUnitTest
, self
).setUp()
2325 @asynctest.mock
.patch("asyncio.sleep")
2326 def test_successful(
2330 mock_get_application
,
2331 mock_disconnect_controller
,
2332 mock_disconnect_model
,
2334 mock_get_controller
,
2336 mock_get_model
.return_value
= juju
.model
.Model()
2337 mock_get_application
.return_value
= FakeApplication()
2339 self
.loop
.run_until_complete(
2340 self
.libjuju
.destroy_unit("app", "model", "machine", 0)
2343 mock_get_unit
.assert_called()
2344 mock_get_application
.assert_called_once()
2345 mock_get_controller
.assert_called_once()
2346 mock_get_model
.assert_called_once()
2347 mock_disconnect_controller
.assert_called_once()
2348 mock_disconnect_model
.assert_called_once()
2353 mock_get_application
,
2354 mock_disconnect_controller
,
2355 mock_disconnect_model
,
2357 mock_get_controller
,
2359 mock_get_model
.return_value
= juju
.model
.Model()
2360 mock_get_application
.return_value
= None
2362 with self
.assertRaises(JujuApplicationNotFound
):
2363 self
.loop
.run_until_complete(
2364 self
.libjuju
.destroy_unit("app", "model", "machine")
2367 mock_get_application
.assert_called_once()
2368 mock_get_controller
.assert_called_once()
2369 mock_get_model
.assert_called_once()
2370 mock_disconnect_controller
.assert_called_once()
2371 mock_disconnect_model
.assert_called_once()
2376 mock_get_application
,
2377 mock_disconnect_controller
,
2378 mock_disconnect_model
,
2380 mock_get_controller
,
2382 mock_get_model
.return_value
= juju
.model
.Model()
2383 mock_get_application
.return_value
= FakeApplication()
2384 mock_get_unit
.return_value
= None
2386 with self
.assertRaises(JujuError
):
2387 self
.loop
.run_until_complete(
2388 self
.libjuju
.destroy_unit("app", "model", "machine")
2391 mock_get_unit
.assert_called_once()
2392 mock_get_application
.assert_called_once()
2393 mock_get_controller
.assert_called_once()
2394 mock_get_model
.assert_called_once()
2395 mock_disconnect_controller
.assert_called_once()
2396 mock_disconnect_model
.assert_called_once()