}
)
logging.disable(logging.CRITICAL)
- self.libjuju = Libjuju(vca_connection, self.loop)
+ self.libjuju = Libjuju(vca_connection)
self.loop.run_until_complete(self.libjuju.disconnect())
# 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("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
@asynctest.mock.patch("juju.model.Model.deploy")
+@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(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 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_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,
):
- mock_get_model.return_value = juju.model.Model()
+ self.setup_bundle_download_mocks(
+ mock_url_parse, mock_bundle, mock_resolve, mock_get_model
+ )
+ model_name = "model1"
+
self.loop.run_until_complete(
- self.libjuju.deploy("cs:osm", "model", wait=True, timeout=0)
+ self.libjuju.deploy(
+ "cs:osm",
+ model_name,
+ wait=True,
+ timeout=0,
+ instantiation_params=None,
+ )
)
- mock_deploy.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_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_deploy_no_wait(
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,
):
- mock_get_model.return_value = juju.model.Model()
+ 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)
+ self.libjuju.deploy(
+ "cs:osm", "model", wait=False, timeout=0, instantiation_params={}
+ )
)
- mock_deploy.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_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()
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()
- mock_get_model.return_value = juju.model.Model()
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_deploy_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
+ )
+ 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()
+
+ 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
+ )
+
+ 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,
+ )
+ )
+
+ 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_deploy_with_instantiation_params_applications_not_found(
+ 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 = {"some_key": {"squid": {"scale": 2}}}
+ self.setup_bundle_download_mocks(
+ mock_url_parse, mock_bundle, mock_resolve, mock_get_model
+ )
+
+ with self.assertRaises(JujuError):
+ self.loop.run_until_complete(
+ self.libjuju.deploy(
+ self.uri,
+ "model1",
+ wait=True,
+ timeout=0,
+ instantiation_params=self.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_not_called()
+ mock_wait_for_model.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ 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)
+
+ 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_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")
mock_get_model,
mock_get_controller,
):
-
mock_get_model.return_value = juju.model.Model()
mock__get_application.return_value = FakeApplication()
output = None
mock_get_model,
mock_get_controller,
):
-
mock_get_application.return_value = FakeApplication()
self.loop.run_until_complete(
mock_get_model,
mock_get_controller,
):
-
mock_get_application.side_effect = Exception()
with self.assertRaises(Exception):
mock_get_model,
mock_get_controller,
):
-
result = {"error": "not found", "response": "response", "request-id": 1}
mock_get_controller.side_effect = JujuAPIError(result)
mock_get_model,
mock_get_controller,
):
-
result = {"error": "not found", "response": "response", "request-id": 1}
mock_get_model.side_effect = JujuAPIError(result)
projects / osm / N2VC.git / blobdiff