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
JujuApplicationNotFound,
JujuActionNotFound,
JujuApplicationExists,
+ JujuInvalidK8sConfiguration,
+ JujuLeaderUnitNotFound,
)
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",
# 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_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,
# 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")
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()
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)