X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_libjuju.py;h=918a2fb87b2fc39decf78181d10e535a40bf7194;hp=b7c7901525e39b260a4d4f4cbaf5bb6a7bb3ed0d;hb=fa02f8a90b7fe1e1b7a80feedef4132bef1ca3e4;hpb=f6e9b00b6f7cd35e45ace4c84b53fe8d12b2438c diff --git a/n2vc/tests/unit/test_libjuju.py b/n2vc/tests/unit/test_libjuju.py index b7c7901..918a2fb 100644 --- a/n2vc/tests/unit/test_libjuju.py +++ b/n2vc/tests/unit/test_libjuju.py @@ -15,148 +15,69 @@ import asyncio import asynctest import tempfile -from unittest import mock +from unittest.mock import Mock, patch import juju import kubernetes from juju.errors import JujuAPIError import logging -from .utils import FakeN2VC, FakeMachine, FakeApplication +from .utils import ( + FakeApplication, + FakeMachine, + FakeManualMachine, + FakeUnit, +) from n2vc.libjuju import Libjuju from n2vc.exceptions import ( JujuControllerFailedConnecting, - JujuModelAlreadyExists, JujuMachineNotFound, JujuApplicationNotFound, JujuActionNotFound, JujuApplicationExists, JujuInvalidK8sConfiguration, JujuLeaderUnitNotFound, + JujuError, ) from n2vc.k8s_juju_conn import generate_rbac_id +from n2vc.tests.unit.utils import AsyncMock +from n2vc.vca.connection import Connection +from n2vc.vca.connection_data import ConnectionData + +cacert = """-----BEGIN CERTIFICATE----- +SOMECERT +-----END CERTIFICATE-----""" + +@asynctest.mock.patch("n2vc.libjuju.Controller") class LibjujuTestCase(asynctest.TestCase): - @asynctest.mock.patch("juju.controller.Controller.update_endpoints") - @asynctest.mock.patch("juju.client.connector.Connector.connect") - @asynctest.mock.patch("juju.controller.Controller.connection") - @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db") + @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert") def setUp( self, - mock__get_api_endpoints_db=None, - mock_connection=None, - mock_connect=None, - mock_update_endpoints=None, - ): - loop = asyncio.get_event_loop() - n2vc = FakeN2VC() - mock__get_api_endpoints_db.return_value = ["127.0.0.1:17070"] - endpoints = "127.0.0.1:17070" - username = "admin" - password = "secret" - cacert = """ - -----BEGIN CERTIFICATE----- - SOMECERT - -----END CERTIFICATE-----""" - self.libjuju = Libjuju( - endpoints, - "192.168.0.155:17070", - username, - password, - cacert, - loop, - log=None, - db={"get_one": []}, - n2vc=n2vc, - apt_mirror="192.168.0.100", - enable_os_upgrade=True, - ) - logging.disable(logging.CRITICAL) - loop.run_until_complete(self.libjuju.disconnect()) - - -@asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task") -@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db") -@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db") -class LibjujuInitTestCase(asynctest.TestCase): - def setUp(self): + mock_base64_to_cacert=None, + ): self.loop = asyncio.get_event_loop() - self.n2vc = FakeN2VC() - self.endpoint = "192.168.100.100:17070" - self.username = "admin" - self.password = "secret" - self.cacert = """ - -----BEGIN CERTIFICATE----- - SOMECERT - -----END CERTIFICATE-----""" - - def test_endpoint_not_in_db( - self, - mock__get_api_endpoints_db, - mock_update_endpoints, - mock_create_health_check_task, - ): - mock__get_api_endpoints_db.return_value = ["another_ip"] - Libjuju( - self.endpoint, - "192.168.0.155:17070", - self.username, - self.password, - self.cacert, - self.loop, - log=None, - db={"get_one": []}, - n2vc=self.n2vc, - apt_mirror="192.168.0.100", - enable_os_upgrade=True, - ) - mock_update_endpoints.assert_called_once_with([self.endpoint]) - mock__get_api_endpoints_db.assert_called_once() - - def test_endpoint_in_db( - self, - mock__get_api_endpoints_db, - mock_update_endpoints, - mock_create_health_check_task, - ): - mock__get_api_endpoints_db.return_value = [self.endpoint, "another_ip"] - Libjuju( - self.endpoint, - "192.168.0.155:17070", - self.username, - self.password, - self.cacert, - self.loop, - log=None, - db={"get_one": []}, - n2vc=self.n2vc, - apt_mirror="192.168.0.100", - enable_os_upgrade=True, - ) - mock_update_endpoints.assert_not_called() - mock__get_api_endpoints_db.assert_called_once() - - def test_no_db_endpoints( - self, - mock__get_api_endpoints_db, - mock_update_endpoints, - mock_create_health_check_task, - ): - mock__get_api_endpoints_db.return_value = None - Libjuju( - self.endpoint, - "192.168.0.155:17070", - self.username, - self.password, - self.cacert, - self.loop, - log=None, - db={"get_one": []}, - n2vc=self.n2vc, - apt_mirror="192.168.0.100", - enable_os_upgrade=True, + self.db = Mock() + mock_base64_to_cacert.return_value = cacert + Connection._load_vca_connection_data = Mock() + vca_connection = Connection(AsyncMock()) + vca_connection._data = ConnectionData( + **{ + "endpoints": ["1.2.3.4:17070"], + "user": "user", + "secret": "secret", + "cacert": "cacert", + "pubkey": "pubkey", + "lxd-cloud": "cloud", + "lxd-credentials": "credentials", + "k8s-cloud": "k8s_cloud", + "k8s-credentials": "k8s_credentials", + "model-config": {}, + "api-proxy": "api_proxy", + } ) - mock_update_endpoints.assert_called_once_with([self.endpoint]) - mock__get_api_endpoints_db.assert_called_once() + logging.disable(logging.CRITICAL) + self.libjuju = Libjuju(vca_connection, self.loop) + self.loop.run_until_complete(self.libjuju.disconnect()) @asynctest.mock.patch("juju.controller.Controller.connect") @@ -164,41 +85,34 @@ class LibjujuInitTestCase(asynctest.TestCase): "juju.controller.Controller.api_endpoints", new_callable=asynctest.CoroutineMock(return_value=["127.0.0.1:17070"]), ) -@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db") class GetControllerTest(LibjujuTestCase): def setUp(self): super(GetControllerTest, self).setUp() - def test_diff_endpoint( - self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect - ): + def test_diff_endpoint(self, mock_api_endpoints, mock_connect): self.libjuju.endpoints = [] controller = self.loop.run_until_complete(self.libjuju.get_controller()) - mock__update_api_endpoints_db.assert_called_once_with(["127.0.0.1:17070"]) self.assertIsInstance(controller, juju.controller.Controller) @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") def test_exception( self, mock_disconnect_controller, - mock__update_api_endpoints_db, mock_api_endpoints, mock_connect, ): self.libjuju.endpoints = [] - mock__update_api_endpoints_db.side_effect = Exception() + + mock_connect.side_effect = Exception() controller = None with self.assertRaises(JujuControllerFailedConnecting): controller = self.loop.run_until_complete(self.libjuju.get_controller()) self.assertIsNone(controller) - mock_disconnect_controller.assert_called_once() + mock_disconnect_controller.assert_called() - def test_same_endpoint_get_controller( - self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect - ): + def test_same_endpoint_get_controller(self, mock_api_endpoints, mock_connect): self.libjuju.endpoints = ["127.0.0.1:17070"] controller = self.loop.run_until_complete(self.libjuju.get_controller()) - mock__update_api_endpoints_db.assert_not_called() self.assertIsInstance(controller, juju.controller.Controller) @@ -238,10 +152,8 @@ class AddModelTest(LibjujuTestCase): ): mock_model_exists.return_value = True - with self.assertRaises(JujuModelAlreadyExists): - self.loop.run_until_complete( - self.libjuju.add_model("existing_model", "cloud") - ) + # This should not raise an exception + self.loop.run_until_complete(self.libjuju.add_model("existing_model", "cloud")) mock_disconnect_controller.assert_called() @@ -259,7 +171,7 @@ class AddModelTest(LibjujuTestCase): mock_get_controller.return_value = juju.controller.Controller() self.loop.run_until_complete( - self.libjuju.add_model("nonexisting_model", "cloud") + self.libjuju.add_model("nonexisting_model", Mock()) ) mock_add_model.assert_called_once() @@ -267,6 +179,126 @@ class AddModelTest(LibjujuTestCase): mock_disconnect_model.assert_called() +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch( + "juju.model.Model.applications", new_callable=asynctest.PropertyMock +) +@asynctest.mock.patch("juju.model.Model.get_action_status") +@asynctest.mock.patch("juju.model.Model.get_action_output") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions") +class GetExecutedActionsTest(LibjujuTestCase): + def setUp(self): + super(GetExecutedActionsTest, self).setUp() + + def test_exception( + self, + mock_get_actions, + mock_get_action_output, + mock_get_action_status, + mock_applications, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = None + with self.assertRaises(JujuError): + self.loop.run_until_complete(self.libjuju.get_executed_actions("model")) + + mock_get_controller.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_model.assert_not_called() + + def test_success( + self, + mock_get_actions, + mock_get_action_output, + mock_get_action_status, + mock_applications, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_applications.return_value = {"existing_app"} + mock_get_actions.return_value = {"action_name": "description"} + mock_get_action_status.return_value = {"id": "status"} + mock_get_action_output.return_value = {"output": "completed"} + + executed_actions = self.loop.run_until_complete( + self.libjuju.get_executed_actions("model") + ) + expected_result = [ + { + "id": "id", + "action": "action_name", + "status": "status", + "output": "completed", + } + ] + self.assertListEqual(expected_result, executed_actions) + self.assertIsInstance(executed_actions, list) + + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +class GetApplicationConfigsTest(LibjujuTestCase): + def setUp(self): + super(GetApplicationConfigsTest, self).setUp() + + def test_exception( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = None + with self.assertRaises(JujuError): + self.loop.run_until_complete( + self.libjuju.get_application_configs("model", "app") + ) + + mock_get_controller.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_model.assert_not_called() + + def test_success( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_application.return_value = FakeApplication() + application_configs = self.loop.run_until_complete( + self.libjuju.get_application_configs("model", "app") + ) + + self.assertEqual(application_configs, ["app_config"]) + + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + @asynctest.mock.patch("juju.controller.Controller.get_model") class GetModelTest(LibjujuTestCase): def setUp(self): @@ -797,7 +829,7 @@ class ExecuteActionTest(LibjujuTestCase): mock_disconnect_controller.assert_called() mock_disconnect_model.assert_called() - def test_succesful_exec( + def test_successful_exec( self, mock_get_action_status, mock_get_action_output, @@ -1073,6 +1105,78 @@ class AddRelationTest(LibjujuTestCase): # TODO destroy_model testcase +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +class DestroyApplicationTest(LibjujuTestCase): + def setUp(self): + super(DestroyApplicationTest, self).setUp() + + def test_success( + self, + mock_get_controller, + mock_get_model, + mock_disconnect_controller, + mock_get_application, + mock_disconnect_model, + ): + mock_get_application.return_value = FakeApplication() + mock_get_model.return_value = None + self.loop.run_until_complete( + self.libjuju.destroy_application( + "existing_model", + "existing_app", + 3600, + ) + ) + mock_get_application.assert_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_application( + self, + mock_get_controller, + mock_get_model, + mock_disconnect_controller, + mock_get_application, + mock_disconnect_model, + ): + mock_get_model.return_value = None + mock_get_application.return_value = None + + self.loop.run_until_complete( + self.libjuju.destroy_application( + "existing_model", + "existing_app", + 3600, + ) + ) + mock_get_application.assert_called() + + def test_exception( + self, + mock_get_controller, + mock_get_model, + mock_disconnect_controller, + mock_get_application, + mock_disconnect_model, + ): + mock_get_application.return_value = FakeApplication + mock_get_model.return_value = None + + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.libjuju.destroy_application( + "existing_model", + "existing_app", + 0, + ) + ) + mock_get_application.assert_called_once() + + # @asynctest.mock.patch("juju.model.Model.get_machines") # @asynctest.mock.patch("logging.Logger.debug") # class DestroyMachineTest(LibjujuTestCase): @@ -1687,7 +1791,7 @@ class GetK8sCloudCredentials(LibjujuTestCase): mock_configuration.key_file = None self.token = None self.cert_data = None - with mock.patch.object(self.libjuju.log, "debug") as mock_debug: + with patch.object(self.libjuju.log, "debug") as mock_debug: credential = self.libjuju.get_k8s_cloud_credential( mock_configuration, self.cert_data, @@ -1796,3 +1900,375 @@ class GetK8sCloudCredentials(LibjujuTestCase): "Cannot set both token and user/pass", ) self.assertTrue(exception_raised) + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model") +class ScaleApplicationTest(LibjujuTestCase): + def setUp(self): + super(ScaleApplicationTest, self).setUp() + + @asynctest.mock.patch("asyncio.sleep") + def test_scale_application( + self, + mock_sleep, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_application, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + self.loop.run_until_complete(self.libjuju.scale_application("model", "app", 2)) + mock_wait_for_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_application( + self, + mock_wait_for, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_application, + mock_get_model, + mock_get_controller, + ): + mock_get_application.return_value = None + mock_get_model.return_value = juju.model.Model() + with self.assertRaises(JujuApplicationNotFound): + self.loop.run_until_complete( + self.libjuju.scale_application("model", "app", 2) + ) + mock_disconnect_controller.assert_called() + mock_disconnect_model.assert_called() + + def test_exception( + self, + mock_wait_for, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_application, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = None + mock_get_application.return_value = FakeApplication() + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.libjuju.scale_application("model", "app", 2, total_timeout=0) + ) + mock_disconnect_controller.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +class GetUnitNumberTest(LibjujuTestCase): + def setUp(self): + super(GetUnitNumberTest, self).setUp() + + def test_successful_get_unit_number( + self, + mock_get_applications, + ): + mock_get_applications.return_value = FakeApplication() + model = juju.model.Model() + result = self.libjuju._get_application_count(model, "app") + self.assertEqual(result, 2) + + def test_non_existing_application( + self, + mock_get_applications, + ): + mock_get_applications.return_value = None + model = juju.model.Model() + result = self.libjuju._get_application_count(model, "app") + self.assertEqual(result, None) + + +@asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock) +class GetMachineInfoTest(LibjujuTestCase): + def setUp(self): + super(GetMachineInfoTest, self).setUp() + + def test_successful( + self, + mock_machines, + ): + machine_id = "existing_machine" + model = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeManualMachine()} + machine, series = self.libjuju._get_machine_info( + machine_id=machine_id, + model=model, + ) + self.assertIsNotNone(machine, series) + + def test_exception( + self, + mock_machines, + ): + machine_id = "not_existing_machine" + machine = series = None + model = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeManualMachine()} + with self.assertRaises(JujuMachineNotFound): + machine, series = self.libjuju._get_machine_info( + machine_id=machine_id, + model=model, + ) + self.assertIsNone(machine, series) + + +class GetUnitTest(LibjujuTestCase): + def setUp(self): + super(GetUnitTest, self).setUp() + + def test_successful(self): + result = self.libjuju._get_unit(FakeApplication(), "existing_machine_id") + self.assertIsInstance(result, FakeUnit) + + def test_return_none(self): + result = self.libjuju._get_unit(FakeApplication(), "not_existing_machine_id") + self.assertIsNone(result) + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +class CheckApplicationExists(LibjujuTestCase): + def setUp(self): + super(CheckApplicationExists, self).setUp() + + def test_successful( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + result = self.loop.run_until_complete( + self.libjuju.check_application_exists( + "model", + "app", + ) + ) + self.assertEqual(result, True) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_application( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + result = self.loop.run_until_complete( + self.libjuju.check_application_exists( + "model", + "app", + ) + ) + self.assertEqual(result, False) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_machine_info") +class AddUnitTest(LibjujuTestCase): + def setUp(self): + super(AddUnitTest, self).setUp() + + @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for") + @asynctest.mock.patch("asyncio.sleep") + def test_successful( + self, + mock_sleep, + mock_wait_for, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_machine_info.return_value = FakeMachine(), "series" + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_wait_for.assert_called_once() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_app( + self, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + with self.assertRaises(JujuApplicationNotFound): + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_machine( + self, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_machine_info.side_effect = JujuMachineNotFound() + with self.assertRaises(JujuMachineNotFound): + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_unit") +class DestroyUnitTest(LibjujuTestCase): + def setUp(self): + super(DestroyUnitTest, self).setUp() + + @asynctest.mock.patch("asyncio.sleep") + def test_successful( + self, + mock_sleep, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine", 0) + ) + + mock_get_unit.assert_called() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_app( + self, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + + with self.assertRaises(JujuApplicationNotFound): + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine") + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_unit( + self, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_unit.return_value = None + + with self.assertRaises(JujuError): + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine") + ) + + mock_get_unit.assert_called_once() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once()