+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.list_models")
+class ModelsExistTest(LibjujuTestCase):
+ def setUp(self):
+ super(ModelsExistTest, self).setUp()
+
+ def test_model_names_none(self, mock_list_models):
+ mock_list_models.return_value = []
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.models_exist(None))
+
+ def test_model_names_empty(self, mock_list_models):
+ mock_list_models.return_value = []
+ with self.assertRaises(Exception):
+ (exist, non_existing_models) = self.loop.run_until_complete(
+ self.libjuju.models_exist([])
+ )
+
+ def test_model_names_not_existing(self, mock_list_models):
+ mock_list_models.return_value = ["prometheus", "grafana"]
+ (exist, non_existing_models) = self.loop.run_until_complete(
+ self.libjuju.models_exist(["prometheus2", "grafana"])
+ )
+ self.assertFalse(exist)
+ self.assertEqual(non_existing_models, ["prometheus2"])
+
+ def test_model_names_exist(self, mock_list_models):
+ mock_list_models.return_value = ["prometheus", "grafana"]
+ (exist, non_existing_models) = self.loop.run_until_complete(
+ self.libjuju.models_exist(["prometheus", "grafana"])
+ )
+ self.assertTrue(exist)
+ self.assertEqual(non_existing_models, [])
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("juju.controller.Controller.list_offers")
+class ListOffers(LibjujuTestCase):
+ def setUp(self):
+ super(ListOffers, self).setUp()
+
+ def test_disconnect_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"))
+ mock_disconnect_controller.assert_called_once()
+
+ def test_empty_list(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ 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,
+ ):
+ 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()
+ 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")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@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,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_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(
+ "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()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
+
+ def test_parsing_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_get_model.return_value = juju.model.Model()
+ mock_consume.side_effect = juju.offerendpoints.ParseError("")
+
+ with self.assertRaises(juju.offerendpoints.ParseError):
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
+
+ def test_juju_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_get_model.return_value = juju.model.Model()
+ mock_consume.side_effect = juju.errors.JujuError("")
+
+ with self.assertRaises(juju.errors.JujuError):
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
+
+ def test_juju_api_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ 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": ""}
+ )
+
+ with self.assertRaises(juju.errors.JujuAPIError):
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
+
+
+@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_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()