X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_libjuju.py;h=67cd19f896716c04116cfcf307d55e637dfd7ef4;hp=123da4a03692a5ffe0efab1e28e817c201efce3e;hb=HEAD;hpb=59f520da90fb12b9d9871889dfbc5d57aa14c591 diff --git a/n2vc/tests/unit/test_libjuju.py b/n2vc/tests/unit/test_libjuju.py index 123da4a..38d8d0e 100644 --- a/n2vc/tests/unit/test_libjuju.py +++ b/n2vc/tests/unit/test_libjuju.py @@ -15,62 +15,71 @@ 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 n2vc.definitions import Offer, RelationEndpoint +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, + mock_base64_to_cacert=None, + ): + self.loop = asyncio.get_event_loop() + 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", + } ) logging.disable(logging.CRITICAL) - loop.run_until_complete(self.libjuju.disconnect()) + self.libjuju = Libjuju(vca_connection) + self.loop.run_until_complete(self.libjuju.disconnect()) @asynctest.mock.patch("juju.controller.Controller.connect") @@ -78,41 +87,34 @@ class LibjujuTestCase(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) @@ -152,10 +154,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() @@ -173,7 +173,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() @@ -181,13 +181,134 @@ 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): super(GetModelTest, self).setUp() def test_get_model( - self, mock_get_model, + self, + mock_get_model, ): mock_get_model.return_value = juju.model.Model() model = self.loop.run_until_complete( @@ -203,7 +324,9 @@ class ModelExistsTest(LibjujuTestCase): super(ModelExistsTest, self).setUp() async def test_existing_model( - self, mock_list_models, mock_get_controller, + self, + mock_list_models, + mock_get_controller, ): mock_list_models.return_value = ["existing_model"] self.assertTrue( @@ -214,7 +337,10 @@ class ModelExistsTest(LibjujuTestCase): @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") async def test_no_controller( - self, mock_disconnect_controller, mock_list_models, mock_get_controller, + self, + mock_disconnect_controller, + mock_list_models, + mock_get_controller, ): mock_list_models.return_value = ["existing_model"] mock_get_controller.return_value = juju.controller.Controller() @@ -222,7 +348,9 @@ class ModelExistsTest(LibjujuTestCase): mock_disconnect_controller.assert_called_once() async def test_non_existing_model( - self, mock_list_models, mock_get_controller, + self, + mock_list_models, + mock_get_controller, ): mock_list_models.return_value = ["existing_model"] self.assertFalse( @@ -368,140 +496,558 @@ class CreateMachineTest(LibjujuTestCase): # TODO test provision machine +@asynctest.mock.patch("os.remove") +@asynctest.mock.patch("n2vc.libjuju.yaml.dump") +@asynctest.mock.patch("builtins.open", create=True) @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.machines", new_callable=asynctest.PropertyMock) +@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model") @asynctest.mock.patch("juju.model.Model.deploy") -@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.create_machine") -class DeployCharmTest(LibjujuTestCase): +@asynctest.mock.patch("juju.model.CharmhubDeployType.resolve") +@asynctest.mock.patch("n2vc.libjuju.BundleHandler") +@asynctest.mock.patch("juju.url.URL.parse") +class DeployTest(LibjujuTestCase): def setUp(self): - super(DeployCharmTest, self).setUp() + super(DeployTest, self).setUp() + self.instantiation_params = {"applications": {"squid": {"scale": 2}}} + self.architecture = "amd64" + self.uri = "cs:osm" + self.url = AsyncMock() + self.url.schema = juju.url.Schema.CHARM_HUB + self.bundle_instance = None + + def setup_bundle_download_mocks( + self, mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ): + mock_url_parse.return_value = self.url + mock_bundle.return_value = AsyncMock() + mock_resolve.return_value = AsyncMock() + mock_resolve.origin = AsyncMock() + mock_get_model.return_value = juju.model.Model() + self.bundle_instance = mock_bundle.return_value + self.bundle_instance.applications = {"squid"} - def test_existing_app( + def assert_overlay_file_is_written(self, filename, mocked_file, mock_yaml, mock_os): + mocked_file.assert_called_once_with(filename, "w") + mock_yaml.assert_called_once_with( + self.instantiation_params, mocked_file.return_value.__enter__.return_value + ) + mock_os.assert_called_once_with(filename) + + def assert_overlay_file_is_not_written(self, mocked_file, mock_yaml, mock_os): + mocked_file.assert_not_called() + mock_yaml.assert_not_called() + mock_os.assert_not_called() + + def assert_bundle_is_downloaded(self, mock_resolve, mock_url_parse): + mock_resolve.assert_called_once_with( + self.url, self.architecture, entity_url=self.uri + ) + mock_url_parse.assert_called_once_with(self.uri) + self.bundle_instance.fetch_plan.assert_called_once_with( + self.url, mock_resolve.origin + ) + + def assert_bundle_is_not_downloaded(self, mock_resolve, mock_url_parse): + mock_resolve.assert_not_called() + mock_url_parse.assert_not_called() + self.bundle_instance.fetch_plan.assert_not_called() + + def test_deploy( self, - mock_create_machine, - mock_wait_for, + mock_url_parse, + mock_bundle, + mock_resolve, mock_deploy, - mock_machines, - mock_applications, + mock_wait_for_model, mock_disconnect_controller, mock_disconnect_model, mock_get_model, mock_get_controller, + mocked_file, + mock_yaml, + mock_os, ): - mock_get_model.return_value = juju.model.Model() - mock_applications.return_value = {"existing_app"} + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + model_name = "model1" - application = None - with self.assertRaises(JujuApplicationExists): - application = self.loop.run_until_complete( - self.libjuju.deploy_charm("existing_app", "path", "model", "machine",) + self.loop.run_until_complete( + self.libjuju.deploy( + "cs:osm", + model_name, + wait=True, + timeout=0, + instantiation_params=None, ) - self.assertIsNone(application) - - mock_disconnect_controller.assert_called() - mock_disconnect_model.assert_called() + ) + self.assert_overlay_file_is_not_written(mocked_file, mock_yaml, mock_os) + self.assert_bundle_is_not_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with("cs:osm", trust=True, overlays=[]) + mock_wait_for_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - def test_non_existing_machine( + def test_deploy_no_wait( self, - mock_create_machine, - mock_wait_for, + mock_url_parse, + mock_bundle, + mock_resolve, mock_deploy, - mock_machines, - mock_applications, + mock_wait_for_model, mock_disconnect_controller, mock_disconnect_model, mock_get_model, mock_get_controller, + mocked_file, + mock_yaml, + mock_os, ): - mock_get_model.return_value = juju.model.Model() - mock_machines.return_value = {"existing_machine": FakeMachine()} - application = None - with self.assertRaises(JujuMachineNotFound): - application = self.loop.run_until_complete( - self.libjuju.deploy_charm("app", "path", "model", "machine",) + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + self.loop.run_until_complete( + self.libjuju.deploy( + "cs:osm", "model", wait=False, timeout=0, instantiation_params={} ) + ) + self.assert_overlay_file_is_not_written(mocked_file, mock_yaml, mock_os) + self.assert_bundle_is_not_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with("cs:osm", trust=True, overlays=[]) + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - self.assertIsNone(application) - - mock_disconnect_controller.assert_called() - mock_disconnect_model.assert_called() + def test_deploy_exception( + self, + mock_url_parse, + mock_bundle, + mock_resolve, + mock_deploy, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + mocked_file, + mock_yaml, + mock_os, + ): + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + mock_deploy.side_effect = Exception() + with self.assertRaises(Exception): + self.loop.run_until_complete(self.libjuju.deploy("cs:osm", "model")) + self.assert_overlay_file_is_not_written(mocked_file, mock_yaml, mock_os) + self.assert_bundle_is_not_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once() + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - def test_2_units( + def test_deploy_with_instantiation_params( self, - mock_create_machine, - mock_wait_for, + mock_url_parse, + mock_bundle, + mock_resolve, mock_deploy, - mock_machines, - mock_applications, + mock_wait_for_model, mock_disconnect_controller, mock_disconnect_model, mock_get_model, mock_get_controller, + mocked_file, + mock_yaml, + mock_os, ): - mock_get_model.return_value = juju.model.Model() - mock_machines.return_value = {"existing_machine": FakeMachine()} - mock_create_machine.return_value = (FakeMachine(), "other") - mock_deploy.return_value = FakeApplication() - application = self.loop.run_until_complete( - self.libjuju.deploy_charm( - "app", "path", "model", "existing_machine", num_units=2, + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + model_name = "model1" + expected_filename = "{}-overlay.yaml".format(model_name) + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + model_name, + wait=True, + timeout=0, + instantiation_params=self.instantiation_params, ) ) + self.assert_overlay_file_is_written( + expected_filename, mocked_file, mock_yaml, mock_os + ) + self.assert_bundle_is_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with( + self.uri, trust=True, overlays=[expected_filename] + ) + mock_wait_for_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - self.assertIsInstance(application, FakeApplication) - - mock_deploy.assert_called_once() - mock_wait_for.assert_called_once() + def test_deploy_with_instantiation_params_no_applications( + self, + mock_url_parse, + mock_bundle, + mock_resolve, + mock_deploy, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + mocked_file, + mock_yaml, + mock_os, + ): + self.instantiation_params = {"applications": {}} + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) - mock_create_machine.assert_called_once() + model_name = "model3" + expected_filename = "{}-overlay.yaml".format(model_name) + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + model_name, + wait=False, + timeout=0, + instantiation_params=self.instantiation_params, + ) + ) - mock_disconnect_controller.assert_called() - mock_disconnect_model.assert_called() + self.assert_overlay_file_is_written( + expected_filename, mocked_file, mock_yaml, mock_os + ) + self.assert_bundle_is_not_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with( + self.uri, trust=True, overlays=[expected_filename] + ) + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - def test_1_unit( + def test_deploy_with_instantiation_params_applications_not_found( self, - mock_create_machine, - mock_wait_for, + mock_url_parse, + mock_bundle, + mock_resolve, mock_deploy, - mock_machines, - mock_applications, + mock_wait_for_model, mock_disconnect_controller, mock_disconnect_model, mock_get_model, mock_get_controller, + mocked_file, + mock_yaml, + mock_os, ): - mock_get_model.return_value = juju.model.Model() - mock_machines.return_value = {"existing_machine": FakeMachine()} - mock_deploy.return_value = FakeApplication() - application = self.loop.run_until_complete( - self.libjuju.deploy_charm("app", "path", "model", "existing_machine") + self.instantiation_params = {"some_key": {"squid": {"scale": 2}}} + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model ) - self.assertIsInstance(application, FakeApplication) + with self.assertRaises(JujuError): + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + "model1", + wait=True, + timeout=0, + instantiation_params=self.instantiation_params, + ) + ) - mock_deploy.assert_called_once() - mock_wait_for.assert_called_once() + self.assert_overlay_file_is_not_written(mocked_file, mock_yaml, mock_os) + self.assert_bundle_is_not_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_not_called() + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - mock_disconnect_controller.assert_called() - mock_disconnect_model.assert_called() + def test_deploy_overlay_contains_invalid_app( + self, + mock_url_parse, + mock_bundle, + mock_resolve, + mock_deploy, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + mocked_file, + mock_yaml, + mock_os, + ): + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + self.bundle_instance.applications = {"new_app"} + with self.assertRaises(JujuApplicationNotFound) as error: + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + "model2", + wait=True, + timeout=0, + instantiation_params=self.instantiation_params, + ) + ) + error_msg = "Cannot find application ['squid'] in original bundle {'new_app'}" + self.assertEqual(str(error.exception), error_msg) -@asynctest.mock.patch( - "juju.model.Model.applications", new_callable=asynctest.PropertyMock -) -class GetApplicationTest(LibjujuTestCase): - def setUp(self): - super(GetApplicationTest, self).setUp() + self.assert_overlay_file_is_not_written(mocked_file, mock_yaml, mock_os) + self.assert_bundle_is_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_not_called() + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() - def test_existing_application( - self, mock_applications, + def test_deploy_exception_with_instantiation_params( + self, + mock_url_parse, + mock_bundle, + mock_resolve, + mock_deploy, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + mocked_file, + mock_yaml, + mock_os, + ): + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + + mock_deploy.side_effect = Exception() + model_name = "model2" + expected_filename = "{}-overlay.yaml".format(model_name) + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + model_name, + instantiation_params=self.instantiation_params, + ) + ) + + self.assert_overlay_file_is_written( + expected_filename, mocked_file, mock_yaml, mock_os + ) + self.assert_bundle_is_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with( + self.uri, trust=True, overlays=[expected_filename] + ) + mock_wait_for_model.assert_not_called() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + @asynctest.mock.patch("logging.Logger.warning") + def test_deploy_exception_when_deleting_file_is_not_propagated( + self, + mock_warning, + mock_url_parse, + mock_bundle, + mock_resolve, + mock_deploy, + mock_wait_for_model, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + mocked_file, + mock_yaml, + mock_os, + ): + self.setup_bundle_download_mocks( + mock_url_parse, mock_bundle, mock_resolve, mock_get_model + ) + + mock_os.side_effect = OSError("Error") + model_name = "model2" + expected_filename = "{}-overlay.yaml".format(model_name) + self.loop.run_until_complete( + self.libjuju.deploy( + self.uri, + model_name, + instantiation_params=self.instantiation_params, + ) + ) + + self.assert_overlay_file_is_written( + expected_filename, mocked_file, mock_yaml, mock_os + ) + self.assert_bundle_is_downloaded(mock_resolve, mock_url_parse) + mock_deploy.assert_called_once_with( + self.uri, trust=True, overlays=[expected_filename] + ) + mock_wait_for_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + mock_warning.assert_called_with( + "Overlay file {} could not be removed: Error".format(expected_filename) + ) + + +@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.machines", new_callable=asynctest.PropertyMock) +@asynctest.mock.patch("juju.model.Model.deploy") +@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.create_machine") +class DeployCharmTest(LibjujuTestCase): + def setUp(self): + super(DeployCharmTest, self).setUp() + + def test_existing_app( + self, + mock_create_machine, + mock_wait_for, + mock_deploy, + mock_machines, + 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"} + + application = None + with self.assertRaises(JujuApplicationExists): + application = self.loop.run_until_complete( + self.libjuju.deploy_charm( + "existing_app", + "path", + "model", + "machine", + ) + ) + self.assertIsNone(application) + + mock_disconnect_controller.assert_called() + mock_disconnect_model.assert_called() + + def test_non_existing_machine( + self, + mock_create_machine, + mock_wait_for, + mock_deploy, + mock_machines, + mock_applications, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeMachine()} + application = None + with self.assertRaises(JujuMachineNotFound): + application = self.loop.run_until_complete( + self.libjuju.deploy_charm( + "app", + "path", + "model", + "machine", + ) + ) + + self.assertIsNone(application) + + mock_disconnect_controller.assert_called() + mock_disconnect_model.assert_called() + + def test_2_units( + self, + mock_create_machine, + mock_wait_for, + mock_deploy, + mock_machines, + mock_applications, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeMachine()} + mock_create_machine.return_value = (FakeMachine(), "other") + mock_deploy.return_value = FakeApplication() + application = self.loop.run_until_complete( + self.libjuju.deploy_charm( + "app", + "path", + "model", + "existing_machine", + num_units=2, + ) + ) + + self.assertIsInstance(application, FakeApplication) + + mock_deploy.assert_called_once() + mock_wait_for.assert_called_once() + + mock_create_machine.assert_called_once() + + mock_disconnect_controller.assert_called() + mock_disconnect_model.assert_called() + + def test_1_unit( + self, + mock_create_machine, + mock_wait_for, + mock_deploy, + mock_machines, + mock_applications, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeMachine()} + mock_deploy.return_value = FakeApplication() + application = self.loop.run_until_complete( + self.libjuju.deploy_charm("app", "path", "model", "existing_machine") + ) + + self.assertIsInstance(application, FakeApplication) + + mock_deploy.assert_called_once() + mock_wait_for.assert_called_once() + + mock_disconnect_controller.assert_called() + mock_disconnect_model.assert_called() + + +@asynctest.mock.patch( + "juju.model.Model.applications", new_callable=asynctest.PropertyMock +) +class GetApplicationTest(LibjujuTestCase): + def setUp(self): + super(GetApplicationTest, self).setUp() + + def test_existing_application( + self, + mock_applications, ): mock_applications.return_value = {"existing_app": "exists"} model = juju.model.Model() @@ -509,7 +1055,8 @@ class GetApplicationTest(LibjujuTestCase): self.assertEqual(result, "exists") def test_non_existing_application( - self, mock_applications, + self, + mock_applications, ): mock_applications.return_value = {"existing_app": "exists"} model = juju.model.Model() @@ -546,7 +1093,11 @@ class ExecuteActionTest(LibjujuTestCase): status = None with self.assertRaises(JujuApplicationNotFound): output, status = self.loop.run_until_complete( - self.libjuju.execute_action("app", "model", "action",) + self.libjuju.execute_action( + "app", + "model", + "action", + ) ) self.assertIsNone(output) self.assertIsNone(status) @@ -565,14 +1116,17 @@ class ExecuteActionTest(LibjujuTestCase): mock_get_model, mock_get_controller, ): - mock_get_model.return_value = juju.model.Model() mock__get_application.return_value = FakeApplication() output = None status = None with self.assertRaises(JujuActionNotFound): output, status = self.loop.run_until_complete( - self.libjuju.execute_action("app", "model", "action",) + self.libjuju.execute_action( + "app", + "model", + "action", + ) ) self.assertIsNone(output) self.assertIsNone(status) @@ -602,7 +1156,11 @@ class ExecuteActionTest(LibjujuTestCase): status = None with self.assertRaises(JujuLeaderUnitNotFound): output, status = self.loop.run_until_complete( - self.libjuju.execute_action("app", "model", "action",) + self.libjuju.execute_action( + "app", + "model", + "action", + ) ) self.assertIsNone(output) self.assertIsNone(status) @@ -610,7 +1168,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, @@ -775,13 +1333,49 @@ class AddRelationTest(LibjujuTestCase): mock_add_relation.side_effect = JujuAPIError(result) self.loop.run_until_complete( - self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) ) mock_warning.assert_called_with("Relation not found: not found") mock_disconnect_controller.assert_called_once() mock_disconnect_model.assert_called_once() + @asynctest.mock.patch("logging.Logger.warning") + def test_not_found_in_error_code( + self, + mock_warning, + mock_add_relation, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + result = { + "error": "relation cannot be added", + "error-code": "not found", + "response": "response", + "request-id": 1, + } + + mock_get_model.return_value = juju.model.Model() + mock_add_relation.side_effect = JujuAPIError(result) + + self.loop.run_until_complete( + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) + ) + + mock_warning.assert_called_with("Relation not found: relation cannot be added") + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + @asynctest.mock.patch("logging.Logger.warning") def test_already_exists( self, @@ -799,13 +1393,51 @@ class AddRelationTest(LibjujuTestCase): mock_add_relation.side_effect = JujuAPIError(result) self.loop.run_until_complete( - self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) ) mock_warning.assert_called_with("Relation already exists: already exists") mock_disconnect_controller.assert_called_once() mock_disconnect_model.assert_called_once() + @asynctest.mock.patch("logging.Logger.warning") + def test_already_exists_error_code( + self, + mock_warning, + mock_add_relation, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + result = { + "error": "relation cannot be added", + "error-code": "already exists", + "response": "response", + "request-id": 1, + } + + mock_get_model.return_value = juju.model.Model() + mock_add_relation.side_effect = JujuAPIError(result) + + self.loop.run_until_complete( + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) + ) + + mock_warning.assert_called_with( + "Relation already exists: relation cannot be added" + ) + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + def test_exception( self, mock_add_relation, @@ -820,7 +1452,11 @@ class AddRelationTest(LibjujuTestCase): with self.assertRaises(JujuAPIError): self.loop.run_until_complete( - self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) ) mock_disconnect_controller.assert_called_once() @@ -837,7 +1473,11 @@ class AddRelationTest(LibjujuTestCase): mock_get_model.return_value = juju.model.Model() self.loop.run_until_complete( - self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) + self.libjuju.add_relation( + "model", + "app1:relation1", + "app2:relation2", + ) ) mock_add_relation.assert_called_with("app1:relation1", "app2:relation2") @@ -855,7 +1495,11 @@ class AddRelationTest(LibjujuTestCase): mock_get_model.return_value = juju.model.Model() self.loop.run_until_complete( - self.libjuju.add_relation("model", "app1:relation1", "saas_name",) + self.libjuju.add_relation( + "model", + "app1:relation1", + "saas_name", + ) ) mock_add_relation.assert_called_with("app1:relation1", "saas_name") @@ -866,84 +1510,209 @@ class AddRelationTest(LibjujuTestCase): # TODO destroy_model testcase -@asynctest.mock.patch("juju.model.Model.get_machines") -@asynctest.mock.patch("logging.Logger.debug") -class DestroyMachineTest(LibjujuTestCase): - def setUp(self): - super(DestroyMachineTest, self).setUp() - - def test_success( - self, mock_debug, mock_get_machines, - ): - mock_get_machines.side_effect = [ - {"machine": FakeMachine()}, - {"machine": FakeMachine()}, - {}, - ] - self.loop.run_until_complete( - self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,) - ) - calls = [ - asynctest.call("Waiting for machine machine is destroyed"), - asynctest.call("Machine destroyed: machine"), - ] - mock_debug.assert_has_calls(calls) - - def test_no_machine( - self, mock_debug, mock_get_machines, - ): - mock_get_machines.return_value = {} - self.loop.run_until_complete( - self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,) - ) - mock_debug.assert_called_with("Machine not found: machine") - - @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 ConfigureApplicationTest(LibjujuTestCase): +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +class DestroyApplicationTest(LibjujuTestCase): def setUp(self): - super(ConfigureApplicationTest, self).setUp() + super(DestroyApplicationTest, self).setUp() def test_success( self, - mock_get_application, + mock_get_controller, + mock_get_model, mock_disconnect_controller, + mock_get_application, mock_disconnect_model, - mock_get_model, - mock_get_controller, ): - mock_get_application.return_value = FakeApplication() - + mock_get_model.return_value = None self.loop.run_until_complete( - self.libjuju.configure_application("model", "app", {"config"},) + self.libjuju.destroy_application( + "existing_model", + "existing_app", + 3600, + ) ) - mock_get_application.assert_called_once() + mock_get_application.assert_called() mock_disconnect_controller.assert_called_once() mock_disconnect_model.assert_called_once() - def test_exception( + def test_no_application( self, - mock_get_application, + 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): +# def setUp(self): +# super(DestroyMachineTest, self).setUp() + +# def test_success_manual_machine( +# self, mock_debug, mock_get_machines, +# ): +# mock_get_machines.side_effect = [ +# {"machine": FakeManualMachine()}, +# {"machine": FakeManualMachine()}, +# {}, +# ] +# self.loop.run_until_complete( +# self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,) +# ) +# calls = [ +# asynctest.call("Waiting for machine machine is destroyed"), +# asynctest.call("Machine destroyed: machine"), +# ] +# mock_debug.assert_has_calls(calls) + +# def test_no_machine( +# self, mock_debug, mock_get_machines, +# ): +# mock_get_machines.return_value = {} +# self.loop.run_until_complete( +# self.libjuju.destroy_machine(juju.model.Model(), "machine", 2) +# ) +# mock_debug.assert_called_with("Machine not found: machine") + + +@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 ConfigureApplicationTest(LibjujuTestCase): + def setUp(self): + super(ConfigureApplicationTest, self).setUp() + + 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() + + self.loop.run_until_complete( + self.libjuju.configure_application( + "model", + "app", + {"config"}, + ) + ) + mock_get_application.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + def test_exception( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): mock_get_application.side_effect = Exception() with self.assertRaises(Exception): self.loop.run_until_complete( - self.libjuju.configure_application("model", "app", {"config"},) + self.libjuju.configure_application( + "model", + "app", + {"config"}, + ) ) mock_disconnect_controller.assert_called_once() mock_disconnect_model.assert_called_once() + def test_controller_exception( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + result = {"error": "not found", "response": "response", "request-id": 1} + + mock_get_controller.side_effect = JujuAPIError(result) + + with self.assertRaises(JujuAPIError): + self.loop.run_until_complete( + self.libjuju.configure_application( + "model", + "app", + {"config"}, + ) + ) + mock_get_model.assert_not_called() + mock_disconnect_controller.assert_not_called() + mock_disconnect_model.assert_not_called() + + def test_get_model_exception( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + result = {"error": "not found", "response": "response", "request-id": 1} + mock_get_model.side_effect = JujuAPIError(result) + + with self.assertRaises(JujuAPIError): + self.loop.run_until_complete( + self.libjuju.configure_application( + "model", + "app", + {"config"}, + ) + ) + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_not_called() + # TODO _get_api_endpoints_db test case # TODO _update_api_endpoints_db test case @@ -958,7 +1727,10 @@ class ListModelsTest(LibjujuTestCase): super(ListModelsTest, self).setUp() def test_containing( - self, mock_list_models, mock_disconnect_controller, mock_get_controller, + self, + mock_list_models, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() mock_list_models.return_value = ["existingmodel"] @@ -968,7 +1740,10 @@ class ListModelsTest(LibjujuTestCase): self.assertEquals(models, ["existingmodel"]) def test_not_containing( - self, mock_list_models, mock_disconnect_controller, mock_get_controller, + self, + mock_list_models, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() mock_list_models.return_value = ["existingmodel", "model"] @@ -978,7 +1753,10 @@ class ListModelsTest(LibjujuTestCase): self.assertEquals(models, []) def test_no_contains_arg( - self, mock_list_models, mock_disconnect_controller, mock_get_controller, + self, + mock_list_models, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() mock_list_models.return_value = ["existingmodel", "model"] @@ -1030,33 +1808,142 @@ class ListOffers(LibjujuTestCase): super(ListOffers, self).setUp() def test_disconnect_controller( - self, mock_list_offers, mock_disconnect_controller, mock_get_controller, + self, + mock_list_offers, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() mock_list_offers.side_effect = Exception() with self.assertRaises(Exception): - self.loop.run_until_complete(self.libjuju.list_offers("model")) + self.loop.run_until_complete(self.libjuju._list_offers("model")) mock_disconnect_controller.assert_called_once() def test_empty_list( - self, mock_list_offers, mock_disconnect_controller, mock_get_controller, + self, + mock_list_offers, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() - mock_list_offers.return_value = [] - offers = self.loop.run_until_complete(self.libjuju.list_offers("model")) + offer_results = Mock() + offer_results.results = [] + mock_list_offers.return_value = offer_results + offers = self.loop.run_until_complete(self.libjuju._list_offers("model")) self.assertEqual(offers, []) mock_disconnect_controller.assert_called_once() def test_non_empty_list( - self, mock_list_offers, mock_disconnect_controller, mock_get_controller, + self, + mock_list_offers, + mock_disconnect_controller, + mock_get_controller, + ): + mock_get_controller.return_value = juju.controller.Controller() + offer = Mock() + offer_results = Mock() + offer_results.results = [offer] + mock_list_offers.return_value = offer_results + offers = self.loop.run_until_complete(self.libjuju._list_offers("model")) + self.assertEqual(offers, [offer]) + mock_disconnect_controller.assert_called_once() + + def test_matching_offer_name( + self, + mock_list_offers, + mock_disconnect_controller, + mock_get_controller, + ): + mock_get_controller.return_value = juju.controller.Controller() + offer_1 = Mock() + offer_1.offer_name = "offer1" + offer_2 = Mock() + offer_2.offer_name = "offer2" + offer_results = Mock() + offer_results.results = [offer_1, offer_2] + mock_list_offers.return_value = offer_results + offers = self.loop.run_until_complete( + self.libjuju._list_offers("model", offer_name="offer2") + ) + self.assertEqual(offers, [offer_2]) + mock_disconnect_controller.assert_called_once() + + def test_not_matching_offer_name( + self, + mock_list_offers, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() - mock_list_offers.return_value = ["offer"] - offers = self.loop.run_until_complete(self.libjuju.list_offers("model")) - self.assertEqual(offers, ["offer"]) + offer_1 = Mock() + offer_1.offer_name = "offer1" + offer_2 = Mock() + offer_2.offer_name = "offer2" + offer_results = Mock() + offer_results.results = [offer_1, offer_2] + mock_list_offers.return_value = offer_results + offers = self.loop.run_until_complete( + self.libjuju._list_offers("model", offer_name="offer3") + ) + self.assertEqual(offers, []) mock_disconnect_controller.assert_called_once() +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("juju.controller.Controller.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._list_offers") +@asynctest.mock.patch("juju.model.Model.create_offer") +class OfferTest(LibjujuTestCase): + def setUp(self): + super(OfferTest, self).setUp() + + def test_offer( + self, + mock_create_offer, + mock__list_offers, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + controller = juju.controller.Controller() + model = juju.model.Model() + mock_get_controller.return_value = controller + mock_get_model.return_value = model + endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint") + self.loop.run_until_complete(self.libjuju.offer(endpoint)) + mock_create_offer.assert_called_with( + "app-name:endpoint", offer_name="app-name-endpoint" + ) + mock_disconnect_model.assert_called_once_with(model) + mock_disconnect_controller.assert_called_once_with(controller) + + def test_offer_exception( + self, + mock_create_offer, + mock__list_offers, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + controller = juju.controller.Controller() + model = juju.model.Model() + mock_get_controller.return_value = controller + mock_get_model.return_value = model + mock__list_offers.return_value = [] + endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint") + with self.assertRaises(Exception): + self.loop.run_until_complete(self.libjuju.offer(endpoint)) + mock_create_offer.assert_called_with( + "app-name:endpoint", offer_name="app-name-endpoint" + ) + mock_disconnect_model.assert_called_once_with(model) + mock_disconnect_controller.assert_called_once_with(controller) + + @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") @asynctest.mock.patch("juju.controller.Controller.get_model") @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") @@ -1064,7 +1951,9 @@ class ListOffers(LibjujuTestCase): @asynctest.mock.patch("juju.model.Model.consume") class ConsumeTest(LibjujuTestCase): def setUp(self): + self.offer_url = "admin/model.offer_name" super(ConsumeTest, self).setUp() + self.provider_libjuju = self.libjuju def test_consume( self, @@ -1074,13 +1963,25 @@ class ConsumeTest(LibjujuTestCase): mock_get_model, mock_get_controller, ): - mock_get_controller.return_value = juju.controller.Controller() + self_controller = juju.controller.Controller() + provider_controller = juju.controller.Controller() + mock_get_controller.side_effect = [self_controller, provider_controller] mock_get_model.return_value = juju.model.Model() - self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name")) - mock_consume.assert_called_once() + self.loop.run_until_complete( + self.libjuju.consume( + "model_name", + Offer(self.offer_url, vca_id="vca-id"), + self.provider_libjuju, + ) + ) + mock_consume.assert_called_once_with( + "admin/model.offer_name", + application_alias="offer_name-model-vca-id", + controller=provider_controller, + ) mock_disconnect_model.assert_called_once() - mock_disconnect_controller.assert_called_once() + self.assertEqual(mock_disconnect_controller.call_count, 2) def test_parsing_error_exception( self, @@ -1096,11 +1997,13 @@ class ConsumeTest(LibjujuTestCase): with self.assertRaises(juju.offerendpoints.ParseError): self.loop.run_until_complete( - self.libjuju.consume("offer_url", "model_name") + self.libjuju.consume( + "model_name", Offer(self.offer_url), self.provider_libjuju + ) ) mock_consume.assert_called_once() mock_disconnect_model.assert_called_once() - mock_disconnect_controller.assert_called_once() + self.assertEqual(mock_disconnect_controller.call_count, 2) def test_juju_error_exception( self, @@ -1116,11 +2019,13 @@ class ConsumeTest(LibjujuTestCase): with self.assertRaises(juju.errors.JujuError): self.loop.run_until_complete( - self.libjuju.consume("offer_url", "model_name") + self.libjuju.consume( + "model_name", Offer(self.offer_url), self.provider_libjuju + ) ) mock_consume.assert_called_once() mock_disconnect_model.assert_called_once() - mock_disconnect_controller.assert_called_once() + self.assertEqual(mock_disconnect_controller.call_count, 2) def test_juju_api_error_exception( self, @@ -1138,11 +2043,13 @@ class ConsumeTest(LibjujuTestCase): with self.assertRaises(juju.errors.JujuAPIError): self.loop.run_until_complete( - self.libjuju.consume("offer_url", "model_name") + self.libjuju.consume( + "model_name", Offer(self.offer_url), self.provider_libjuju + ) ) mock_consume.assert_called_once() mock_disconnect_model.assert_called_once() - mock_disconnect_controller.assert_called_once() + self.assertEqual(mock_disconnect_controller.call_count, 2) @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential") @@ -1150,45 +2057,56 @@ class ConsumeTest(LibjujuTestCase): class AddK8sTest(LibjujuTestCase): def setUp(self): super(AddK8sTest, self).setUp() - self.configuration = kubernetes.client.configuration.Configuration() + name = "cloud" + rbac_id = generate_rbac_id() + token = "token" + client_cert_data = "cert" + configuration = kubernetes.client.configuration.Configuration() + storage_class = "storage_class" + credential_name = name + + self._add_k8s_args = { + "name": name, + "rbac_id": rbac_id, + "token": token, + "client_cert_data": client_cert_data, + "configuration": configuration, + "storage_class": storage_class, + "credential_name": credential_name, + } def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential): - self.loop.run_until_complete( - self.libjuju.add_k8s("cloud", self.configuration, "storage_class") - ) + self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args)) mock_add_cloud.assert_called_once() mock_get_k8s_cloud_credential.assert_called_once() def test_add_k8s_exception(self, mock_add_cloud, mock_get_k8s_cloud_credential): mock_add_cloud.side_effect = Exception() with self.assertRaises(Exception): - self.loop.run_until_complete( - self.libjuju.add_k8s("cloud", self.configuration, "storage_class") - ) + self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args)) mock_add_cloud.assert_called_once() mock_get_k8s_cloud_credential.assert_called_once() def test_add_k8s_missing_name(self, mock_add_cloud, mock_get_k8s_cloud_credential): + self._add_k8s_args["name"] = "" with self.assertRaises(Exception): - self.loop.run_until_complete( - self.libjuju.add_k8s("", self.configuration, "storage_class") - ) + self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args)) mock_add_cloud.assert_not_called() def test_add_k8s_missing_storage_name( self, mock_add_cloud, mock_get_k8s_cloud_credential ): + self._add_k8s_args["storage_class"] = "" with self.assertRaises(Exception): - self.loop.run_until_complete( - self.libjuju.add_k8s("cloud", self.configuration, "") - ) + self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args)) mock_add_cloud.assert_not_called() def test_add_k8s_missing_configuration_keys( self, mock_add_cloud, mock_get_k8s_cloud_credential ): + self._add_k8s_args["configuration"] = None with self.assertRaises(Exception): - self.loop.run_until_complete(self.libjuju.add_k8s("cloud", None, "")) + self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args)) mock_add_cloud.assert_not_called() @@ -1285,7 +2203,10 @@ class RemoveCloudTest(LibjujuTestCase): super(RemoveCloudTest, self).setUp() def test_remove_cloud( - self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller, + self, + mock_remove_cloud, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() @@ -1294,7 +2215,10 @@ class RemoveCloudTest(LibjujuTestCase): mock_disconnect_controller.assert_called_once() def test_remove_cloud_exception( - self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller, + self, + mock_remove_cloud, + mock_disconnect_controller, + mock_get_controller, ): mock_get_controller.return_value = juju.controller.Controller() mock_remove_cloud.side_effect = Exception() @@ -1309,6 +2233,8 @@ class RemoveCloudTest(LibjujuTestCase): class GetK8sCloudCredentials(LibjujuTestCase): def setUp(self): super(GetK8sCloudCredentials, self).setUp() + self.cert_data = "cert" + self.token = "token" @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration") def test_not_supported(self, mock_exception, mock_configuration): @@ -1318,12 +2244,19 @@ class GetK8sCloudCredentials(LibjujuTestCase): mock_configuration.cert_file = None mock_configuration.key_file = None exception_raised = False + self.token = None + self.cert_data = None try: - _ = self.libjuju.get_k8s_cloud_credential(mock_configuration) + _ = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, + ) except JujuInvalidK8sConfiguration as e: exception_raised = True self.assertEqual( - e.message, "authentication method not supported", + e.message, + "authentication method not supported", ) self.assertTrue(exception_raised) @@ -1333,7 +2266,13 @@ class GetK8sCloudCredentials(LibjujuTestCase): mock_configuration.ssl_ca_cert = None mock_configuration.cert_file = None mock_configuration.key_file = None - credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) + self.token = None + self.cert_data = None + credential = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, + ) self.assertEqual( credential, juju.client._definitions.CloudCredential( @@ -1341,46 +2280,54 @@ class GetK8sCloudCredentials(LibjujuTestCase): ), ) - def test_user_no_pass(self, mock_configuration): - mock_configuration.username = "admin" - mock_configuration.password = "" - mock_configuration.ssl_ca_cert = None - mock_configuration.cert_file = None - mock_configuration.key_file = None - with mock.patch.object(self.libjuju.log, "debug") as mock_debug: - credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) - self.assertEqual( - credential, - juju.client._definitions.CloudCredential( - attrs={"username": "admin", "password": ""}, auth_type="userpass" - ), - ) - mock_debug.assert_called_once_with( - "credential for user admin has empty password" - ) - def test_user_pass_with_cert(self, mock_configuration): mock_configuration.username = "admin" mock_configuration.password = "admin" - ssl_ca_cert = tempfile.NamedTemporaryFile() - with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file: - ssl_ca_cert_file.write("cacert") - mock_configuration.ssl_ca_cert = ssl_ca_cert.name + mock_configuration.ssl_ca_cert = None mock_configuration.cert_file = None mock_configuration.key_file = None - credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) + self.token = None + credential = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, + ) self.assertEqual( credential, juju.client._definitions.CloudCredential( attrs={ + "ClientCertificateData": self.cert_data, "username": "admin", "password": "admin", - "ClientCertificateData": "cacert", }, auth_type="userpasswithcert", ), ) + def test_user_no_pass(self, mock_configuration): + mock_configuration.username = "admin" + mock_configuration.password = "" + mock_configuration.ssl_ca_cert = None + mock_configuration.cert_file = None + mock_configuration.key_file = None + self.token = None + self.cert_data = None + with patch.object(self.libjuju.log, "debug") as mock_debug: + credential = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, + ) + self.assertEqual( + credential, + juju.client._definitions.CloudCredential( + attrs={"username": "admin", "password": ""}, auth_type="userpass" + ), + ) + mock_debug.assert_called_once_with( + "credential for user admin has empty password" + ) + def test_cert(self, mock_configuration): mock_configuration.username = "" mock_configuration.password = "" @@ -1391,69 +2338,67 @@ class GetK8sCloudCredentials(LibjujuTestCase): mock_configuration.ssl_ca_cert = ssl_ca_cert.name mock_configuration.cert_file = None mock_configuration.key_file = None - credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) - self.assertEqual( - credential, - juju.client._definitions.CloudCredential( - attrs={"ClientCertificateData": "cacert", "Token": "Token"}, - auth_type="certificate", - ), + credential = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, ) - - def test_oauth2(self, mock_configuration): - mock_configuration.username = "" - mock_configuration.password = "" - mock_configuration.api_key = {"authorization": "Bearer Token"} - key = tempfile.NamedTemporaryFile() - with open(key.name, "w") as key_file: - key_file.write("key") - mock_configuration.ssl_ca_cert = None - mock_configuration.cert_file = None - mock_configuration.key_file = key.name - credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) self.assertEqual( credential, juju.client._definitions.CloudCredential( - attrs={"ClientKeyData": "key", "Token": "Token"}, auth_type="oauth2", + attrs={"ClientCertificateData": self.cert_data, "Token": self.token}, + auth_type="certificate", ), ) - @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration") - def test_oauth2_missing_token(self, mock_exception, mock_configuration): - mock_configuration.username = "" - mock_configuration.password = "" - key = tempfile.NamedTemporaryFile() - with open(key.name, "w") as key_file: - key_file.write("key") - mock_configuration.ssl_ca_cert = None - mock_configuration.cert_file = None - mock_configuration.key_file = key.name - exception_raised = False - try: - _ = self.libjuju.get_k8s_cloud_credential(mock_configuration) - except JujuInvalidK8sConfiguration as e: - exception_raised = True - self.assertEqual( - e.message, "missing token for auth type oauth2", - ) - self.assertTrue(exception_raised) - - def test_unknown_api_key(self, mock_configuration): - mock_configuration.username = "" - mock_configuration.password = "" - mock_configuration.api_key = {"authorization": "Bearer Token Wrong"} - mock_configuration.ssl_ca_cert = None - mock_configuration.cert_file = None - mock_configuration.key_file = None - exception_raised = False - try: - _ = self.libjuju.get_k8s_cloud_credential(mock_configuration) - except JujuInvalidK8sConfiguration as e: - exception_raised = True - self.assertEqual( - e.message, "unknown format of api_key", - ) - self.assertTrue(exception_raised) + # TODO: Fix this test when oauth authentication is supported + # def test_oauth2(self, mock_configuration): + # mock_configuration.username = "" + # mock_configuration.password = "" + # mock_configuration.api_key = {"authorization": "Bearer Token"} + # key = tempfile.NamedTemporaryFile() + # with open(key.name, "w") as key_file: + # key_file.write("key") + # mock_configuration.ssl_ca_cert = None + # mock_configuration.cert_file = None + # mock_configuration.key_file = key.name + # credential = self.libjuju.get_k8s_cloud_credential( + # mock_configuration, + # self.cert_data, + # self.token, + # ) + # self.assertEqual( + # credential, + # juju.client._definitions.CloudCredential( + # attrs={"ClientKeyData": "key", "Token": "Token"}, + # auth_type="oauth2", + # ), + # ) + + # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration") + # def test_oauth2_missing_token(self, mock_exception, mock_configuration): + # mock_configuration.username = "" + # mock_configuration.password = "" + # key = tempfile.NamedTemporaryFile() + # with open(key.name, "w") as key_file: + # key_file.write("key") + # mock_configuration.ssl_ca_cert = None + # mock_configuration.cert_file = None + # mock_configuration.key_file = key.name + # exception_raised = False + # try: + # _ = self.libjuju.get_k8s_cloud_credential( + # mock_configuration, + # self.cert_data, + # self.token, + # ) + # except JujuInvalidK8sConfiguration as e: + # exception_raised = True + # self.assertEqual( + # e.message, + # "missing token for auth type oauth2", + # ) + # self.assertTrue(exception_raised) def test_exception_cannot_set_token_and_userpass(self, mock_configuration): mock_configuration.username = "admin" @@ -1464,10 +2409,387 @@ class GetK8sCloudCredentials(LibjujuTestCase): mock_configuration.key_file = None exception_raised = False try: - _ = self.libjuju.get_k8s_cloud_credential(mock_configuration) + _ = self.libjuju.get_k8s_cloud_credential( + mock_configuration, + self.cert_data, + self.token, + ) except JujuInvalidK8sConfiguration as e: exception_raised = True self.assertEqual( - e.message, "Cannot set both token and user/pass", + e.message, + "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()