X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_libjuju.py;h=123da4a03692a5ffe0efab1e28e817c201efce3e;hp=f517afc5da58988bb03eca8704af1460db7c883a;hb=59f520da90fb12b9d9871889dfbc5d57aa14c591;hpb=42f328a34a3ec7c066350de81e1331632a3fee92 diff --git a/n2vc/tests/unit/test_libjuju.py b/n2vc/tests/unit/test_libjuju.py index f517afc..123da4a 100644 --- a/n2vc/tests/unit/test_libjuju.py +++ b/n2vc/tests/unit/test_libjuju.py @@ -14,7 +14,10 @@ import asyncio import asynctest +import tempfile +from unittest import mock import juju +import kubernetes from juju.errors import JujuAPIError import logging from .utils import FakeN2VC, FakeMachine, FakeApplication @@ -26,6 +29,8 @@ from n2vc.exceptions import ( JujuApplicationNotFound, JujuActionNotFound, JujuApplicationExists, + JujuInvalidK8sConfiguration, + JujuLeaderUnitNotFound, ) @@ -96,10 +101,11 @@ class GetControllerTest(LibjujuTestCase): ): self.libjuju.endpoints = [] mock__update_api_endpoints_db.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 @@ -151,7 +157,7 @@ class AddModelTest(LibjujuTestCase): 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. @@ -254,7 +260,7 @@ class GetModelStatusTest(LibjujuTestCase): self.assertEqual(status, {"status"}) - def test_excpetion( + def test_exception( self, mock_get_status, mock_disconnect_controller, @@ -264,16 +270,16 @@ class GetModelStatusTest(LibjujuTestCase): ): 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") @@ -319,16 +325,18 @@ class CreateMachineTest(LibjujuTestCase): 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, @@ -390,14 +398,15 @@ class DeployCharmTest(LibjujuTestCase): 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) + 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, @@ -413,15 +422,16 @@ class DeployCharmTest(LibjujuTestCase): ): 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.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, @@ -532,16 +542,17 @@ class ExecuteActionTest(LibjujuTestCase): ): 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.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, @@ -557,17 +568,47 @@ class ExecuteActionTest(LibjujuTestCase): 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.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, @@ -614,15 +655,15 @@ class GetActionTest(LibjujuTestCase): 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, @@ -644,6 +685,70 @@ class GetActionTest(LibjujuTestCase): 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") @@ -670,9 +775,7 @@ class AddRelationTest(LibjujuTestCase): mock_add_relation.side_effect = JujuAPIError(result) self.loop.run_until_complete( - self.libjuju.add_relation( - "model", "app1", "app2", "relation1", "relation2", - ) + self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) ) mock_warning.assert_called_with("Relation not found: not found") @@ -696,9 +799,7 @@ class AddRelationTest(LibjujuTestCase): mock_add_relation.side_effect = JujuAPIError(result) self.loop.run_until_complete( - self.libjuju.add_relation( - "model", "app1", "app2", "relation1", "relation2", - ) + self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) ) mock_warning.assert_called_with("Relation already exists: already exists") @@ -719,13 +820,11 @@ class AddRelationTest(LibjujuTestCase): with self.assertRaises(JujuAPIError): self.loop.run_until_complete( - self.libjuju.add_relation( - "model", "app1", "app2", "relation1", "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, @@ -738,14 +837,28 @@ class AddRelationTest(LibjujuTestCase): mock_get_model.return_value = juju.model.Model() self.loop.run_until_complete( - self.libjuju.add_relation( - "model", "app1", "app2", "relation1", "relation2", - ) + self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",) ) - mock_add_relation.assert_called_with( - relation1="app1:relation1", relation2="app2:relation2" + mock_add_relation.assert_called_with("app1:relation1", "app2:relation2") + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_saas( + self, + mock_add_relation, + 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.add_relation("model", "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() @@ -828,8 +941,8 @@ class ConfigureApplicationTest(LibjujuTestCase): self.loop.run_until_complete( 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() # TODO _get_api_endpoints_db test case @@ -907,3 +1020,454 @@ class ModelsExistTest(LibjujuTestCase): ) 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() + mock_list_offers.return_value = [] + 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() + mock_list_offers.return_value = ["offer"] + offers = self.loop.run_until_complete(self.libjuju.list_offers("model")) + self.assertEqual(offers, ["offer"]) + 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("juju.model.Model.consume") +class ConsumeTest(LibjujuTestCase): + def setUp(self): + super(ConsumeTest, self).setUp() + + def test_consume( + 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() + + 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() + + 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("offer_url", "model_name") + ) + mock_consume.assert_called_once() + mock_disconnect_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + + 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("offer_url", "model_name") + ) + mock_consume.assert_called_once() + mock_disconnect_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + + 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("offer_url", "model_name") + ) + 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() + self.configuration = kubernetes.client.configuration.Configuration() + + def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential): + self.loop.run_until_complete( + self.libjuju.add_k8s("cloud", self.configuration, "storage_class") + ) + 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("cloud", self.configuration, "storage_class") + ) + 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): + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.libjuju.add_k8s("", self.configuration, "storage_class") + ) + mock_add_cloud.assert_not_called() + + def test_add_k8s_missing_storage_name( + self, mock_add_cloud, mock_get_k8s_cloud_credential + ): + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.libjuju.add_k8s("cloud", self.configuration, "") + ) + mock_add_cloud.assert_not_called() + + def test_add_k8s_missing_configuration_keys( + self, mock_add_cloud, mock_get_k8s_cloud_credential + ): + with self.assertRaises(Exception): + self.loop.run_until_complete(self.libjuju.add_k8s("cloud", None, "")) + 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() + + @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 + try: + _ = self.libjuju.get_k8s_cloud_credential(mock_configuration) + 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 + credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) + self.assertEqual( + credential, + juju.client._definitions.CloudCredential( + attrs={"username": "admin", "password": "admin"}, auth_type="userpass" + ), + ) + + 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 + with mock.patch.object(self.libjuju.log, "debug") as mock_debug: + credential = self.libjuju.get_k8s_cloud_credential(mock_configuration) + 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_user_pass_with_cert(self, mock_configuration): + mock_configuration.username = "admin" + mock_configuration.password = "admin" + 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.assertEqual( + credential, + juju.client._definitions.CloudCredential( + attrs={ + "username": "admin", + "password": "admin", + "ClientCertificateData": "cacert", + }, + auth_type="userpasswithcert", + ), + ) + + 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.assertEqual( + credential, + juju.client._definitions.CloudCredential( + attrs={"ClientCertificateData": "cacert", "Token": "Token"}, + auth_type="certificate", + ), + ) + + 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.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) + except JujuInvalidK8sConfiguration as e: + exception_raised = True + self.assertEqual( + e.message, "missing token for auth type oauth2", + ) + self.assertTrue(exception_raised) + + def test_unknown_api_key(self, mock_configuration): + mock_configuration.username = "" + mock_configuration.password = "" + mock_configuration.api_key = {"authorization": "Bearer Token Wrong"} + 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) + except JujuInvalidK8sConfiguration as e: + exception_raised = True + self.assertEqual( + e.message, "unknown format of api_key", + ) + 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) + except JujuInvalidK8sConfiguration as e: + exception_raised = True + self.assertEqual( + e.message, "Cannot set both token and user/pass", + ) + self.assertTrue(exception_raised)