from typing import List
from juju.controller import Controller
-from juju.errors import JujuError
from juju.model import Model
from n2vc.exceptions import (
- JujuApplicationExists,
JujuControllerFailedConnecting,
JujuModelAlreadyExists,
)
finally:
await self.disconnect_controller(controller)
- async def deploy_charm(
- self,
- application_name: str,
- path: str,
- model_name: str,
- config: dict = None,
- series: str = None,
- num_units: int = 1,
- channel: str = "stable",
- ):
- """
- Args:
- application_name (str): Application name.
- path (str): Local path to the charm.
- model_name (str): Model name.
- config (dict): Config for the charm.
- series (str): Series of the charm.
- num_units (str): Number of units to deploy.
- channel (str): Charm store channel from which to retrieve the charm.
-
- Returns:
- (juju.application.Application): Juju application
- """
- self.logger.debug(
- "Deploying charm {} in model {}".format(application_name, model_name)
- )
- self.logger.debug("charm: {}".format(path))
- controller = None
- model = None
- try:
- controller = await self.get_controller()
- model = await self.get_model(controller, model_name)
- if application_name in model.applications:
- raise JujuApplicationExists(
- "Application {} exists".format(application_name)
- )
- application = await model.deploy(
- entity_url=path,
- application_name=application_name,
- channel=channel,
- num_units=num_units,
- series=series,
- config=config,
- )
-
- self.logger.debug(
- "Wait until application {} is ready in model {}".format(
- application_name, model_name
- )
- )
- await self.wait_app_deployment_completion(application_name, model_name)
-
- except JujuError as e:
- if "already exists" in e.message:
- raise JujuApplicationExists(
- "Application {} exists".format(application_name)
- )
- else:
- raise e
- finally:
- await self.disconnect_model(model)
- await self.disconnect_controller(controller)
-
- return application
-
- async def wait_app_deployment_completion(
- self, application_name: str, model_name: str
- ) -> None:
- self.logger.debug(
- "Application {} is ready in model {}".format(application_name, model_name)
- )
-
async def destroy_model(self, model_name: str, force=False) -> None:
controller = None
model = None
import asynctest
import logging
import juju
-from unittest.mock import Mock
from n2vc.temporal_libjuju import Libjuju, ConnectionInfo
from n2vc.exceptions import (
- JujuApplicationExists,
JujuControllerFailedConnecting,
JujuModelAlreadyExists,
)
self.assertEquals(models, expected_list)
-@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
-@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_model")
-@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
-@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
-@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.wait_app_deployment_completion")
-@asynctest.mock.patch(
- "juju.model.Model.applications", new_callable=asynctest.PropertyMock
-)
-@asynctest.mock.patch("juju.model.Model.deploy")
-class DeployCharmTest(LibjujuTestCase):
- def setUp(self):
- super(DeployCharmTest, self).setUp()
-
- def test_existing_app_is_not_deployed(
- self,
- mock_deploy,
- mock_applications,
- mock_wait_app_deployment,
- 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_app_is_deployed(
- self,
- mock_deploy,
- mock_applications,
- mock_wait_app_deployment,
- mock_disconnect_controller,
- mock_disconnect_model,
- mock_get_model,
- mock_get_controller,
- ):
- mock_get_model.return_value = juju.model.Model()
- mock_deploy.return_value = Mock()
- self.loop.run_until_complete(
- self.libjuju.deploy_charm(
- "app",
- "path",
- "model",
- series="series",
- num_units=2,
- )
- )
- mock_deploy.assert_called_once_with(
- entity_url="path",
- application_name="app",
- channel="stable",
- num_units=2,
- series="series",
- config=None,
- )
- mock_wait_app_deployment.assert_called()
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
-
-
@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_model")
@asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.model_exists")