X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_libjuju.py;h=454b87f7288a1d3d57e6af788429ba070bdc8a9f;hp=ae390788b45e2a9b0abe6cfe957b8c21e97330e5;hb=9a63e8d905b4b98f567acccce745b39919fa6860;hpb=12b29244e5d333341166ea92760b8eb245c16b27 diff --git a/n2vc/tests/unit/test_libjuju.py b/n2vc/tests/unit/test_libjuju.py index ae39078..454b87f 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, ) @@ -68,6 +73,91 @@ class LibjujuTestCase(asynctest.TestCase): loop.run_until_complete(self.libjuju.disconnect()) +@asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db") +class LibjujuInitTestCase(asynctest.TestCase): + def setUp(self): + self.loop = asyncio.get_event_loop() + self.n2vc = FakeN2VC() + self.endpoint = "192.168.100.100:17070" + self.username = "admin" + self.password = "secret" + self.cacert = """ + -----BEGIN CERTIFICATE----- + SOMECERT + -----END CERTIFICATE-----""" + + def test_endpoint_not_in_db( + self, + mock__get_api_endpoints_db, + mock_update_endpoints, + mock_create_health_check_task, + ): + mock__get_api_endpoints_db.return_value = ["another_ip"] + Libjuju( + self.endpoint, + "192.168.0.155:17070", + self.username, + self.password, + self.cacert, + self.loop, + log=None, + db={"get_one": []}, + n2vc=self.n2vc, + apt_mirror="192.168.0.100", + enable_os_upgrade=True, + ) + mock_update_endpoints.assert_called_once_with([self.endpoint]) + mock__get_api_endpoints_db.assert_called_once() + + def test_endpoint_in_db( + self, + mock__get_api_endpoints_db, + mock_update_endpoints, + mock_create_health_check_task, + ): + mock__get_api_endpoints_db.return_value = [self.endpoint, "another_ip"] + Libjuju( + self.endpoint, + "192.168.0.155:17070", + self.username, + self.password, + self.cacert, + self.loop, + log=None, + db={"get_one": []}, + n2vc=self.n2vc, + apt_mirror="192.168.0.100", + enable_os_upgrade=True, + ) + mock_update_endpoints.assert_not_called() + mock__get_api_endpoints_db.assert_called_once() + + def test_no_db_endpoints( + self, + mock__get_api_endpoints_db, + mock_update_endpoints, + mock_create_health_check_task, + ): + mock__get_api_endpoints_db.return_value = None + Libjuju( + self.endpoint, + "192.168.0.155:17070", + self.username, + self.password, + self.cacert, + self.loop, + log=None, + db={"get_one": []}, + n2vc=self.n2vc, + apt_mirror="192.168.0.100", + enable_os_upgrade=True, + ) + mock_update_endpoints.assert_called_once_with([self.endpoint]) + mock__get_api_endpoints_db.assert_called_once() + + @asynctest.mock.patch("juju.controller.Controller.connect") @asynctest.mock.patch( "juju.controller.Controller.api_endpoints", @@ -363,6 +453,71 @@ class CreateMachineTest(LibjujuTestCase): # 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") @@ -575,7 +730,35 @@ class ExecuteActionTest(LibjujuTestCase): 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, @@ -652,6 +835,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") @@ -769,37 +1016,37 @@ class AddRelationTest(LibjujuTestCase): # TODO destroy_model testcase -@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( - self, mock_debug, mock_get_machines, - ): - mock_get_machines.side_effect = [ - {"machine": FakeMachine()}, - {"machine": FakeMachine()}, - {}, - ] - 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("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") @@ -1048,49 +1295,50 @@ class ConsumeTest(LibjujuTestCase): 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.auth_data = { - "server": "https://192.168.0.21:16443", - "token": "1234", - "cacert": "cacert", - } + self.configuration = kubernetes.client.configuration.Configuration() - def test_add_k8s(self, mock_add_cloud): + def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential): self.loop.run_until_complete( - self.libjuju.add_k8s("cloud", self.auth_data, "storage_class") + 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): + 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.auth_data, "storage_class") + 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): + 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.auth_data, "storage_class") + 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): + 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.auth_data, "") + self.libjuju.add_k8s("cloud", self.configuration, "") ) mock_add_cloud.assert_not_called() - def test_add_k8s_missing_auth_data_keys(self, mock_add_cloud): + 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", {}, "") - ) + self.loop.run_until_complete(self.libjuju.add_k8s("cloud", None, "")) mock_add_cloud.assert_not_called() @@ -1205,3 +1453,171 @@ class RemoveCloudTest(LibjujuTestCase): 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)