import asyncio
import logging
+import os
import typing
+import yaml
import time
import juju.errors
+from juju.bundle import BundleHandler
from juju.model import Model
from juju.machine import Machine
from juju.application import Application
from juju.unit import Unit
+from juju.url import URL
+from juju.version import DEFAULT_ARCHITECTURE
from juju.client._definitions import (
FullStatus,
QueryApplicationOffersResults,
return machine_id
async def deploy(
- self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
+ self,
+ uri: str,
+ model_name: str,
+ wait: bool = True,
+ timeout: float = 3600,
+ instantiation_params: dict = None,
):
"""
Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
- :param: uri: Path or Charm Store uri in which the charm or bundle can be found
- :param: model_name: Model name
- :param: wait: Indicates whether to wait or not until all applications are active
- :param: timeout: Time in seconds to wait until all applications are active
+ :param uri: Path or Charm Store uri in which the charm or bundle can be found
+ :param model_name: Model name
+ :param wait: Indicates whether to wait or not until all applications are active
+ :param timeout: Time in seconds to wait until all applications are active
+ :param instantiation_params: To be applied as overlay bundle over primary bundle.
"""
controller = await self.get_controller()
model = await self.get_model(controller, model_name)
+ overlays = []
try:
- await model.deploy(uri, trust=True)
+ await self._validate_instantiation_params(uri, model, instantiation_params)
+ overlays = self._get_overlays(model_name, instantiation_params)
+ await model.deploy(uri, trust=True, overlays=overlays)
if wait:
await JujuModelWatcher.wait_for_model(model, timeout=timeout)
self.log.debug("All units active in model {}".format(model_name))
finally:
+ self._remove_overlay_file(overlays)
await self.disconnect_model(model)
await self.disconnect_controller(controller)
+ async def _validate_instantiation_params(
+ self, uri: str, model, instantiation_params: dict
+ ) -> None:
+ """Checks if all the applications in instantiation_params
+ exist ins the original bundle.
+
+ Raises:
+ JujuApplicationNotFound if there is an invalid app in
+ the instantiation params.
+ """
+ overlay_apps = self._get_apps_in_instantiation_params(instantiation_params)
+ if not overlay_apps:
+ return
+ original_apps = await self._get_apps_in_original_bundle(uri, model)
+ if not all(app in original_apps for app in overlay_apps):
+ raise JujuApplicationNotFound(
+ "Cannot find application {} in original bundle {}".format(
+ overlay_apps, original_apps
+ )
+ )
+
+ async def _get_apps_in_original_bundle(self, uri: str, model) -> set:
+ """Bundle is downloaded in BundleHandler.fetch_plan.
+ That method takes care of opening and exception handling.
+
+ Resolve method gets all the information regarding the channel,
+ track, revision, type, source.
+
+ Returns:
+ Set with the names of the applications in original bundle.
+ """
+ url = URL.parse(uri)
+ architecture = DEFAULT_ARCHITECTURE # only AMD64 is allowed
+ res = await model.deploy_types[str(url.schema)].resolve(
+ url, architecture, entity_url=uri
+ )
+ handler = BundleHandler(model, trusted=True, forced=False)
+ await handler.fetch_plan(url, res.origin)
+ return handler.applications
+
+ def _get_apps_in_instantiation_params(self, instantiation_params: dict) -> list:
+ """Extract applications key in instantiation params.
+
+ Returns:
+ List with the names of the applications in instantiation params.
+
+ Raises:
+ JujuError if applications key is not found.
+ """
+ if not instantiation_params:
+ return []
+ try:
+ return [key for key in instantiation_params.get("applications")]
+ except Exception as e:
+ raise JujuError("Invalid overlay format. {}".format(str(e)))
+
+ def _get_overlays(self, model_name: str, instantiation_params: dict) -> list:
+ """Creates a temporary overlay file which includes the instantiation params.
+ Only one overlay file is created.
+
+ Returns:
+ List with one overlay filename. Empty list if there are no instantiation params.
+ """
+ if not instantiation_params:
+ return []
+ file_name = model_name + "-overlay.yaml"
+ self._write_overlay_file(file_name, instantiation_params)
+ return [file_name]
+
+ def _write_overlay_file(self, file_name: str, instantiation_params: dict) -> None:
+ with open(file_name, "w") as file:
+ yaml.dump(instantiation_params, file)
+
+ def _remove_overlay_file(self, overlay: list) -> None:
+ """Overlay contains either one or zero file names."""
+ if not overlay:
+ return
+ try:
+ filename = overlay[0]
+ os.remove(filename)
+ except OSError as e:
+ self.log.warning(
+ "Overlay file {} could not be removed: {}".format(filename, e)
+ )
+
async def add_unit(
self,
application_name: str,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params=None,
)
)
self.assertEqual(mock_chdir.call_count, 2)
model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
def test_success_cs(self, mock_chdir):
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params={},
)
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
+ self.cs_bundle,
+ model_name=self.default_namespace,
+ wait=True,
+ timeout=1800,
+ instantiation_params=None,
)
def test_success_http(self, mock_chdir):
+ params = {"overlay": {"applications": {"squid": {"scale": 2}}}}
self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params=params,
)
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.http_bundle, model_name=self.default_namespace, wait=True, timeout=1800
+ self.http_bundle,
+ model_name=self.default_namespace,
+ wait=True,
+ timeout=1800,
+ instantiation_params=params.get("overlay"),
)
def test_success_not_kdu_name(self, mock_chdir):
+ params = {"some_key": {"applications": {"squid": {"scale": 2}}}}
self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
atomic=True,
db_dict=self.db_dict,
timeout=1800,
+ params=params,
)
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
+ self.cs_bundle,
+ model_name=self.default_namespace,
+ wait=True,
+ timeout=1800,
+ instantiation_params=None,
)
def test_missing_db_dict(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
+ self.cs_bundle,
+ model_name=self.default_namespace,
+ wait=True,
+ timeout=1800,
+ instantiation_params=None,
)
def test_missing_bundle(self, mock_chdir):
model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
# 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")
projects / osm / N2VC.git / commitdiff