import asyncio
import asynctest
+import tempfile
+from unittest.mock import Mock, patch
import juju
+import kubernetes
from juju.errors import JujuAPIError
import logging
-from .utils import FakeN2VC, FakeMachine, FakeApplication
+from .utils import FakeMachine, FakeApplication
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)
+ self.loop.run_until_complete(self.libjuju.disconnect())
@asynctest.mock.patch("juju.controller.Controller.connect")
"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()
+ self.assertIsNone(controller)
+ mock_disconnect_controller.assert_called_once()
- 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)
):
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()
+ mock_disconnect_controller.assert_called()
# TODO Check two job executing at the same time and one returning without doing anything.
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()
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(
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(
@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()
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(
self.assertEqual(status, {"status"})
- def test_excpetion(
+ def test_exception(
self,
mock_get_status,
mock_disconnect_controller,
):
mock_get_model.return_value = juju.model.Model()
mock_get_status.side_effect = Exception()
-
+ status = None
with self.assertRaises(Exception):
status = self.loop.run_until_complete(
self.libjuju.get_model_status("model")
)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
- self.assertIsNone(status)
+ self.assertIsNone(status)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
mock_get_model,
mock_get_controller,
):
+ machine = None
+ bool_res = None
mock_get_model.return_value = juju.model.Model()
with self.assertRaises(JujuMachineNotFound):
machine, bool_res = self.loop.run_until_complete(
self.libjuju.create_machine("model", "non_existing_machine")
)
- self.assertIsNone(machine)
- self.assertIsNone(bool_res)
+ self.assertIsNone(machine)
+ self.assertIsNone(bool_res)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_no_machine(
self,
# TODO test provision 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.juju_watcher.JujuModelWatcher.wait_for_model")
+@asynctest.mock.patch("juju.model.Model.deploy")
+class DeployTest(LibjujuTestCase):
+ def setUp(self):
+ super(DeployTest, self).setUp()
+
+ def test_deploy(
+ self,
+ mock_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ self.loop.run_until_complete(
+ self.libjuju.deploy("cs:osm", "model", wait=True, timeout=0)
+ )
+ mock_deploy.assert_called_once()
+ 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_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ self.loop.run_until_complete(
+ self.libjuju.deploy("cs:osm", "model", wait=False, timeout=0)
+ )
+ 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_exception(
+ self,
+ mock_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ 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"))
+ mock_deploy.assert_called_once()
+ mock_wait_for_model.assert_not_called()
+ 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")
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.libjuju.deploy_charm(
+ "existing_app",
+ "path",
+ "model",
+ "machine",
+ )
)
- self.assertIsNone(application)
+ self.assertIsNone(application)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_non_existing_machine(
self,
):
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.libjuju.deploy_charm(
+ "app",
+ "path",
+ "model",
+ "machine",
+ )
)
- self.assertIsNone(application)
+ self.assertIsNone(application)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_2_units(
self,
mock_deploy.return_value = FakeApplication()
application = self.loop.run_until_complete(
self.libjuju.deploy_charm(
- "app", "path", "model", "existing_machine", num_units=2,
+ "app",
+ "path",
+ "model",
+ "existing_machine",
+ num_units=2,
)
)
super(GetApplicationTest, self).setUp()
def test_existing_application(
- self, mock_applications,
+ self,
+ mock_applications,
):
mock_applications.return_value = {"existing_app": "exists"}
model = juju.model.Model()
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()
):
mock__get_application.return_value = None
mock_get_model.return_value = juju.model.Model()
-
+ output = None
+ 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)
+ self.assertIsNone(output)
+ self.assertIsNone(status)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_no_action(
self,
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)
+ self.assertIsNone(output)
+ self.assertIsNone(status)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
- # TODO no leader unit found exception
+ @asynctest.mock.patch("asyncio.sleep")
+ @asynctest.mock.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
+ def test_no_leader(
+ self,
+ mock_is_leader_from_status,
+ mock_sleep,
+ mock_get_action_status,
+ mock_get_action_output,
+ mock_wait_for,
+ 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_is_leader_from_status.return_value = False
+ output = None
+ status = None
+ with self.assertRaises(JujuLeaderUnitNotFound):
+ output, status = self.loop.run_until_complete(
+ self.libjuju.execute_action(
+ "app",
+ "model",
+ "action",
+ )
+ )
+ self.assertIsNone(output)
+ self.assertIsNone(status)
+
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_succesful_exec(
self,
mock_get_controller,
):
mock_get_application.side_effect = Exception()
-
+ actions = None
with self.assertRaises(Exception):
actions = self.loop.run_until_complete(
self.libjuju.get_actions("app", "model")
)
- self.assertIsNone(actions)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ self.assertIsNone(actions)
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
def test_success(
self,
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("juju.application.Application.get_metrics")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+class GetMetricsTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetMetricsTest, self).setUp()
+
+ def test_get_metrics_success(
+ self,
+ mock_get_application,
+ mock_get_metrics,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_application.return_value = FakeApplication()
+ mock_get_model.return_value = juju.model.Model()
+
+ self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
+
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_get_metrics_exception(
+ self,
+ mock_get_application,
+ mock_get_metrics,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_metrics.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
+
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_missing_args_exception(
+ self,
+ mock_get_application,
+ mock_get_metrics,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.get_metrics("", ""))
+
+ mock_get_controller.assert_not_called()
+ mock_get_model.assert_not_called()
+ mock_disconnect_controller.assert_not_called()
+ mock_disconnect_model.assert_not_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")
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_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")
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()
- mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
def test_success(
self,
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"
- )
+ mock_add_relation.assert_called_with("app1:relation1", "app2:relation2")
mock_disconnect_controller.assert_called_once()
mock_disconnect_model.assert_called_once()
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"
- )
+ mock_add_relation.assert_called_with("app1:relation1", "saas_name")
mock_disconnect_controller.assert_called_once()
mock_disconnect_model.assert_called_once()
# TODO destroy_model testcase
-@asynctest.mock.patch("juju.model.Model.get_machines")
-@asynctest.mock.patch("logging.Logger.debug")
-class DestroyMachineTest(LibjujuTestCase):
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+class DestroyApplicationTest(LibjujuTestCase):
def setUp(self):
- super(DestroyMachineTest, self).setUp()
+ super(DestroyApplicationTest, self).setUp()
def test_success(
- self, mock_debug, mock_get_machines,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
- mock_get_machines.side_effect = [
- {"machine": FakeMachine()},
- {"machine": FakeMachine()},
- {},
- ]
+ mock_get_application.return_value = FakeApplication()
+ mock_get_model.return_value = None
self.loop.run_until_complete(
- self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
+ self.libjuju.destroy_application(
+ "existing_model",
+ "existing_app",
+ 3600,
+ )
)
- calls = [
- asynctest.call("Waiting for machine machine is destroyed"),
- asynctest.call("Machine destroyed: machine"),
- ]
- mock_debug.assert_has_calls(calls)
+ mock_get_application.assert_called()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
- def test_no_machine(
- self, mock_debug, mock_get_machines,
+ def test_no_application(
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
- mock_get_machines.return_value = {}
+ mock_get_model.return_value = None
+ mock_get_application.return_value = None
+
self.loop.run_until_complete(
- self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
+ self.libjuju.destroy_application(
+ "existing_model",
+ "existing_app",
+ 3600,
+ )
)
- mock_debug.assert_called_with("Machine not found: machine")
+ 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")
mock_get_application.return_value = FakeApplication()
self.loop.run_until_complete(
- self.libjuju.configure_application("model", "app", {"config"},)
+ self.libjuju.configure_application(
+ "model",
+ "app",
+ {"config"},
+ )
)
mock_get_application.assert_called_once()
mock_disconnect_controller.assert_called_once()
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()
+ 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
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"]
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"]
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"]
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()
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 = []
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()
mock_list_offers.return_value = ["offer"]
mock_get_controller.return_value = juju.controller.Controller()
mock_get_model.return_value = juju.model.Model()
- self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
- )
+ self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
mock_disconnect_controller.assert_called_once()
):
mock_get_controller.return_value = juju.controller.Controller()
mock_get_model.return_value = juju.model.Model()
- mock_consume.side_effect = juju.errors.JujuAPIError({
- "error": "",
- "response": "",
- "request-id": "",
-
- })
+ mock_consume.side_effect = juju.errors.JujuAPIError(
+ {"error": "", "response": "", "request-id": ""}
+ )
with self.assertRaises(juju.errors.JujuAPIError):
self.loop.run_until_complete(
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.add_cloud")
+class AddK8sTest(LibjujuTestCase):
+ def setUp(self):
+ super(AddK8sTest, self).setUp()
+ 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(**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(**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._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(**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(**self._add_k8s_args))
+ mock_add_cloud.assert_not_called()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("juju.controller.Controller.add_cloud")
+@asynctest.mock.patch("juju.controller.Controller.add_credential")
+class AddCloudTest(LibjujuTestCase):
+ def setUp(self):
+ super(AddCloudTest, self).setUp()
+ self.cloud = juju.client.client.Cloud()
+ self.credential = juju.client.client.CloudCredential()
+
+ def test_add_cloud_with_credential(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ cloud = self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+ self.assertEqual(cloud, self.cloud)
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_called_once_with(
+ "cloud", credential=self.credential, cloud="cloud"
+ )
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_cloud_no_credential(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ cloud = self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud)
+ )
+ self.assertEqual(cloud, self.cloud)
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_cloud_exception(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_add_cloud.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_credential_exception(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_add_credential.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_called_once_with(
+ "cloud", credential=self.credential, cloud="cloud"
+ )
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("juju.controller.Controller.remove_cloud")
+class RemoveCloudTest(LibjujuTestCase):
+ def setUp(self):
+ super(RemoveCloudTest, self).setUp()
+
+ def test_remove_cloud(
+ self,
+ mock_remove_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
+ mock_remove_cloud.assert_called_once_with("cloud")
+ mock_disconnect_controller.assert_called_once()
+
+ def test_remove_cloud_exception(
+ self,
+ mock_remove_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_remove_cloud.side_effect = Exception()
+
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
+ mock_remove_cloud.assert_called_once_with("cloud")
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("kubernetes.client.configuration.Configuration")
+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):
+ mock_configuration.username = ""
+ mock_configuration.password = ""
+ mock_configuration.ssl_ca_cert = None
+ 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.cert_data,
+ self.token,
+ )
+ except JujuInvalidK8sConfiguration as e:
+ exception_raised = True
+ self.assertEqual(
+ e.message,
+ "authentication method not supported",
+ )
+ self.assertTrue(exception_raised)
+
+ def test_user_pass(self, mock_configuration):
+ mock_configuration.username = "admin"
+ mock_configuration.password = "admin"
+ mock_configuration.ssl_ca_cert = None
+ mock_configuration.cert_file = None
+ mock_configuration.key_file = None
+ 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(
+ attrs={"username": "admin", "password": "admin"}, auth_type="userpass"
+ ),
+ )
+
+ def test_user_pass_with_cert(self, mock_configuration):
+ mock_configuration.username = "admin"
+ mock_configuration.password = "admin"
+ mock_configuration.ssl_ca_cert = None
+ mock_configuration.cert_file = None
+ mock_configuration.key_file = None
+ 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",
+ },
+ 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 = ""
+ mock_configuration.api_key = {"authorization": "Bearer Token"}
+ 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.cert_file = None
+ mock_configuration.key_file = 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, "Token": self.token},
+ auth_type="certificate",
+ ),
+ )
+
+ # 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"
+ mock_configuration.password = "pass"
+ mock_configuration.api_key = {"authorization": "No_bearer_token"}
+ 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,
+ self.cert_data,
+ self.token,
+ )
+ except JujuInvalidK8sConfiguration as e:
+ exception_raised = True
+ self.assertEqual(
+ 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_succesful_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)
projects / osm / N2VC.git / blobdiff