import asyncio
import asynctest
import tempfile
-from unittest import mock
+from unittest.mock import Mock, patch
import juju
import kubernetes
from juju.errors import JujuAPIError
import logging
-from .utils import FakeN2VC, FakeMachine, FakeApplication
+from .utils import FakeMachine, FakeApplication
from n2vc.libjuju import Libjuju
from n2vc.exceptions import (
JujuControllerFailedConnecting,
JujuApplicationExists,
JujuInvalidK8sConfiguration,
JujuLeaderUnitNotFound,
+ JujuError,
)
from n2vc.k8s_juju_conn import generate_rbac_id
+from n2vc.tests.unit.utils import AsyncMock
+from n2vc.vca.connection import Connection
+from n2vc.vca.connection_data import ConnectionData
+cacert = """-----BEGIN CERTIFICATE-----
+SOMECERT
+-----END CERTIFICATE-----"""
+
+
+@asynctest.mock.patch("n2vc.libjuju.Controller")
class LibjujuTestCase(asynctest.TestCase):
- @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
- @asynctest.mock.patch("juju.client.connector.Connector.connect")
- @asynctest.mock.patch("juju.controller.Controller.connection")
- @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
+ @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
def setUp(
self,
- mock__get_api_endpoints_db=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
- ):
- loop = asyncio.get_event_loop()
- n2vc = FakeN2VC()
- mock__get_api_endpoints_db.return_value = ["127.0.0.1:17070"]
- endpoints = "127.0.0.1:17070"
- username = "admin"
- password = "secret"
- cacert = """
- -----BEGIN CERTIFICATE-----
- SOMECERT
- -----END CERTIFICATE-----"""
- self.libjuju = Libjuju(
- endpoints,
- "192.168.0.155:17070",
- username,
- password,
- cacert,
- loop,
- log=None,
- db={"get_one": []},
- n2vc=n2vc,
- )
- logging.disable(logging.CRITICAL)
- 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):
+ mock_base64_to_cacert=None,
+ ):
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,
+ self.db = Mock()
+ mock_base64_to_cacert.return_value = cacert
+ Connection._load_vca_connection_data = Mock()
+ vca_connection = Connection(AsyncMock())
+ vca_connection._data = ConnectionData(
+ **{
+ "endpoints": ["1.2.3.4:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "pubkey": "pubkey",
+ "lxd-cloud": "cloud",
+ "lxd-credentials": "credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "model-config": {},
+ "api-proxy": "api_proxy",
+ }
)
- 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,
- )
- 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,
- )
- mock_update_endpoints.assert_called_once_with([self.endpoint])
- mock__get_api_endpoints_db.assert_called_once()
+ logging.disable(logging.CRITICAL)
+ self.libjuju = Libjuju(vca_connection, self.loop)
+ self.loop.run_until_complete(self.libjuju.disconnect())
@asynctest.mock.patch("juju.controller.Controller.connect")
"juju.controller.Controller.api_endpoints",
new_callable=asynctest.CoroutineMock(return_value=["127.0.0.1:17070"]),
)
-@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
class GetControllerTest(LibjujuTestCase):
def setUp(self):
super(GetControllerTest, self).setUp()
- def test_diff_endpoint(
- self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
- ):
+ def test_diff_endpoint(self, mock_api_endpoints, mock_connect):
self.libjuju.endpoints = []
controller = self.loop.run_until_complete(self.libjuju.get_controller())
- mock__update_api_endpoints_db.assert_called_once_with(["127.0.0.1:17070"])
self.assertIsInstance(controller, juju.controller.Controller)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
def test_exception(
self,
mock_disconnect_controller,
- mock__update_api_endpoints_db,
mock_api_endpoints,
mock_connect,
):
self.libjuju.endpoints = []
- mock__update_api_endpoints_db.side_effect = Exception()
+
+ mock_connect.side_effect = Exception()
controller = None
with self.assertRaises(JujuControllerFailedConnecting):
controller = self.loop.run_until_complete(self.libjuju.get_controller())
self.assertIsNone(controller)
mock_disconnect_controller.assert_called_once()
- def test_same_endpoint_get_controller(
- self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
- ):
+ def test_same_endpoint_get_controller(self, mock_api_endpoints, mock_connect):
self.libjuju.endpoints = ["127.0.0.1:17070"]
controller = self.loop.run_until_complete(self.libjuju.get_controller())
- mock__update_api_endpoints_db.assert_not_called()
self.assertIsInstance(controller, juju.controller.Controller)
mock_model_exists.return_value = True
# This should not raise an exception
- self.loop.run_until_complete(
- self.libjuju.add_model("existing_model", "cloud")
- )
+ self.loop.run_until_complete(self.libjuju.add_model("existing_model", "cloud"))
mock_disconnect_controller.assert_called()
mock_get_controller.return_value = juju.controller.Controller()
self.loop.run_until_complete(
- self.libjuju.add_model("nonexisting_model", "cloud")
+ self.libjuju.add_model("nonexisting_model", Mock())
)
mock_add_model.assert_called_once()
mock_disconnect_model.assert_called()
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch(
+ "juju.model.Model.applications", new_callable=asynctest.PropertyMock
+)
+@asynctest.mock.patch("juju.model.Model.get_action_status")
+@asynctest.mock.patch("juju.model.Model.get_action_output")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
+class GetExecutedActionsTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetExecutedActionsTest, self).setUp()
+
+ def test_exception(
+ self,
+ mock_get_actions,
+ mock_get_action_output,
+ mock_get_action_status,
+ mock_applications,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = None
+ with self.assertRaises(JujuError):
+ self.loop.run_until_complete(self.libjuju.get_executed_actions("model"))
+
+ mock_get_controller.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_model.assert_not_called()
+
+ def test_success(
+ self,
+ mock_get_actions,
+ mock_get_action_output,
+ mock_get_action_status,
+ mock_applications,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_applications.return_value = {"existing_app"}
+ mock_get_actions.return_value = {"action_name": "description"}
+ mock_get_action_status.return_value = {"id": "status"}
+ mock_get_action_output.return_value = {"output": "completed"}
+
+ executed_actions = self.loop.run_until_complete(
+ self.libjuju.get_executed_actions("model")
+ )
+ expected_result = [{'id': 'id', 'action': 'action_name',
+ 'status': 'status', 'output': 'completed'}]
+ self.assertListEqual(expected_result, executed_actions)
+ self.assertIsInstance(executed_actions, list)
+
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+class GetApplicationConfigsTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetApplicationConfigsTest, self).setUp()
+
+ def test_exception(
+ self,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = None
+ with self.assertRaises(JujuError):
+ self.loop.run_until_complete(
+ self.libjuju.get_application_configs("model", "app"))
+
+ mock_get_controller.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_model.assert_not_called()
+
+ def test_success(
+ self,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_application.return_value = FakeApplication()
+ application_configs = self.loop.run_until_complete(self.libjuju
+ .get_application_configs("model", "app"))
+
+ self.assertEqual(application_configs, ["app_config"])
+
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+
@asynctest.mock.patch("juju.controller.Controller.get_model")
class GetModelTest(LibjujuTestCase):
def setUp(self):
# TODO destroy_model testcase
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+class DestroyApplicationTest(LibjujuTestCase):
+ def setUp(self):
+ super(DestroyApplicationTest, self).setUp()
+
+ def test_success(
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
+ ):
+ mock_get_application.return_value = FakeApplication()
+ mock_get_model.return_value = None
+ self.loop.run_until_complete(
+ self.libjuju.destroy_application(
+ "existing_model",
+ "existing_app",
+ 3600,
+ )
+ )
+ mock_get_application.assert_called()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_application(
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
+ ):
+ mock_get_model.return_value = None
+ mock_get_application.return_value = None
+
+ self.loop.run_until_complete(
+ self.libjuju.destroy_application(
+ "existing_model",
+ "existing_app",
+ 3600,
+ )
+ )
+ mock_get_application.assert_called()
+
+ def test_exception(
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
+ ):
+ mock_get_application.return_value = FakeApplication
+ mock_get_model.return_value = None
+
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.destroy_application(
+ "existing_model",
+ "existing_app",
+ 0,
+ )
+ )
+ mock_get_application.assert_called_once()
+
+
# @asynctest.mock.patch("juju.model.Model.get_machines")
# @asynctest.mock.patch("logging.Logger.debug")
# class DestroyMachineTest(LibjujuTestCase):
mock_configuration.key_file = None
self.token = None
self.cert_data = None
- with mock.patch.object(self.libjuju.log, "debug") as mock_debug:
+ with patch.object(self.libjuju.log, "debug") as mock_debug:
credential = self.libjuju.get_k8s_cloud_credential(
mock_configuration,
self.cert_data,
projects / osm / N2VC.git / blobdiff