import logging
import asynctest
from unittest.mock import Mock
+from n2vc.definitions import Offer, RelationEndpoint
from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
-from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock
-from n2vc.exceptions import (
- MethodNotImplemented,
- K8sException,
- N2VCBadArgumentsException,
-)
+from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication
+from n2vc.exceptions import MethodNotImplemented, K8sException
+from n2vc.vca.connection_data import ConnectionData
class K8sJujuConnTestCase(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.k8s_juju_conn.base64_to_cacert")
@asynctest.mock.patch("n2vc.k8s_juju_conn.Libjuju")
+ @asynctest.mock.patch("n2vc.k8s_juju_conn.MotorStore")
+ @asynctest.mock.patch("n2vc.k8s_juju_conn.get_connection")
+ @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
def setUp(
self,
- mock_libjuju=None,
mock_base64_to_cacert=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
+ mock_get_connection=None,
+ mock_store=None,
+ mock_libjuju=None,
):
self.loop = asyncio.get_event_loop()
- mock_libjuju.return_value = AsyncMock()
- db = Mock()
- vca_config = {
- "secret": "secret",
- "api_proxy": "api_proxy",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- "user": "user",
- "host": "1.1.1.1",
- "port": 17070,
- "ca_cert": "cacert",
- }
-
+ self.db = Mock()
+ mock_base64_to_cacert.return_value = """
+ -----BEGIN CERTIFICATE-----
+ SOMECERT
+ -----END CERTIFICATE-----"""
+ mock_libjuju.return_value = Mock()
+ mock_store.return_value = AsyncMock()
+ mock_vca_connection = Mock()
+ mock_get_connection.return_value = mock_vca_connection
+ mock_vca_connection.data.return_value = 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",
+ }
+ )
logging.disable(logging.CRITICAL)
+ self.kdu_name = "kdu_name"
+ self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
+ self.default_namespace = self.kdu_instance
+
self.k8s_juju_conn = K8sJujuConnector(
fs=fslocal.FsLocal(),
- db=db,
+ db=self.db,
log=None,
- loop=self.loop,
- vca_config=vca_config,
- on_update_db=None,
- )
-
-
-class K8sJujuConnInitSuccessTestCase(asynctest.TestCase):
- def setUp(
- self,
- ):
- logging.disable(logging.CRITICAL)
-
- @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.k8s_juju_conn.base64_to_cacert")
- @asynctest.mock.patch("n2vc.libjuju.Libjuju.__init__")
- def test_success(
- self,
- mock_libjuju=None,
- mock_base64_to_cacert=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
- ):
- mock_libjuju.return_value = None
- loop = asyncio.get_event_loop()
- log = logging.getLogger()
- db = Mock()
- vca_config = {
- "secret": "secret",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- "user": "user",
- "host": "1.1.1.1",
- "port": 17070,
- "ca_cert": "cacert",
- }
- K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=log,
- loop=self.loop,
- vca_config=vca_config,
on_update_db=None,
)
-
- mock_libjuju.assert_called_once_with(
- endpoint="1.1.1.1:17070",
- api_proxy=None, # Not needed for k8s charms
- model_config={},
- username="user",
- password="secret",
- cacert=mock_base64_to_cacert.return_value,
- loop=loop,
- log=log,
- db=db,
+ self.k8s_juju_conn._store.get_vca_id.return_value = None
+ self.k8s_juju_conn.libjuju = Mock()
+ # Mock Kubectl
+ self.kubectl = Mock()
+ self.kubectl.get_secret_data = AsyncMock()
+ self.kubectl.get_secret_data.return_value = ("token", "cacert")
+ self.kubectl.get_services.return_value = [{}]
+ self.k8s_juju_conn._get_kubectl = Mock()
+ self.k8s_juju_conn._get_kubectl.return_value = self.kubectl
+ self.k8s_juju_conn._obtain_namespace_from_db = Mock(
+ return_value=self.default_namespace
)
-class K8sJujuConnectorInitFailureTestCase(asynctest.TestCase):
- def setUp(
- self,
- ):
- self.loop = asyncio.get_event_loop()
- logging.disable(logging.CRITICAL)
- self.vca_config = {
- "secret": "secret",
- "api_proxy": "api_proxy",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- "user": "user",
- "host": "1.1.1.1",
- "port": 17070,
- "ca_cert": "cacert",
- }
-
- def test_missing_vca_config_host(self):
- db = Mock()
- self.vca_config.pop("host")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_user(self):
- db = Mock()
- self.vca_config.pop("user")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_secret(self):
- db = Mock()
- self.vca_config.pop("secret")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_ca_cert(self):
- db = Mock()
- self.vca_config.pop("ca_cert")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
-
-@asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
class InitEnvTest(K8sJujuConnTestCase):
def setUp(self):
super(InitEnvTest, self).setUp()
self.k8s_juju_conn.libjuju.add_k8s = AsyncMock()
- self.k8s_juju_conn._create_cluster_role = Mock()
- self.k8s_juju_conn._create_service_account = Mock()
- self.k8s_juju_conn._create_cluster_role_binding = Mock()
- self.k8s_juju_conn._delete_cluster_role = Mock()
- self.k8s_juju_conn._delete_service_account = Mock()
- self.k8s_juju_conn._delete_cluster_role_binding = Mock()
- self.k8s_juju_conn._get_secret_data = AsyncMock()
- self.k8s_juju_conn._get_secret_data.return_value = ("token", "cacert")
def test_with_cluster_uuid(
self,
- mock_get_default_storage_class,
):
reuse_cluster_uuid = "uuid"
uuid, created = self.loop.run_until_complete(
self.assertTrue(created)
self.assertEqual(uuid, reuse_cluster_uuid)
- mock_get_default_storage_class.assert_called_once()
+ self.kubectl.get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
def test_with_no_cluster_uuid(
self,
- mock_get_default_storage_class,
):
uuid, created = self.loop.run_until_complete(
self.k8s_juju_conn.init_env(k8s_creds=kubeconfig)
self.assertTrue(created)
self.assertTrue(isinstance(uuid, str))
- mock_get_default_storage_class.assert_called_once()
+ self.kubectl.get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
def test_init_env_exception(
self,
- mock_get_default_storage_class,
):
self.k8s_juju_conn.libjuju.add_k8s.side_effect = Exception()
created = None
uuid, created = self.loop.run_until_complete(
self.k8s_juju_conn.init_env(k8s_creds=kubeconfig)
)
-
self.assertIsNone(created)
self.assertIsNone(uuid)
- mock_get_default_storage_class.assert_called_once()
+ self.kubectl.create_cluster_role.assert_called_once()
+ self.kubectl.create_service_account.assert_called_once()
+ self.kubectl.create_cluster_role_binding.assert_called_once()
+ self.kubectl.get_default_storage_class.assert_called_once()
+ self.kubectl.delete_cluster_role.assert_called_once()
+ self.kubectl.delete_service_account.assert_called_once()
+ self.kubectl.delete_cluster_role_binding.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
cloud_creds = Mock()
cloud_creds.result = {"attrs": {RBAC_LABEL_KEY_NAME: "asd"}}
self.k8s_juju_conn.libjuju.get_cloud_credentials.return_value = [cloud_creds]
- self.k8s_juju_conn._delete_cluster_role_binding = Mock()
- self.k8s_juju_conn._delete_service_account = Mock()
- self.k8s_juju_conn._delete_cluster_role = Mock()
self.k8s_juju_conn.get_credentials = Mock()
self.k8s_juju_conn.get_credentials.return_value = kubeconfig
self.local_bundle = "bundle"
self.cs_bundle = "cs:bundle"
self.http_bundle = "https://example.com/bundle.yaml"
- self.kdu_name = "kdu_name"
self.cluster_uuid = "cluster"
- self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
self.k8s_juju_conn.libjuju.add_model = AsyncMock()
self.k8s_juju_conn.libjuju.deploy = AsyncMock()
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params=None,
)
)
self.assertEqual(mock_chdir.call_count, 2)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
def test_success_cs(self, mock_chdir):
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params={},
)
)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
def test_success_http(self, mock_chdir):
+ params = {"overlay": {"applications": {"squid": {"scale": 2}}}}
self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
+ params=params,
)
)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.http_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=params.get("overlay"),
)
def test_success_not_kdu_name(self, mock_chdir):
+ params = {"some_key": {"applications": {"squid": {"scale": 2}}}}
self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
atomic=True,
db_dict=self.db_dict,
timeout=1800,
+ params=params,
)
)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
def test_missing_db_dict(self, mock_chdir):
timeout=1800,
)
)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
def test_missing_bundle(self, mock_chdir):
timeout=1800,
)
)
- self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=self.kdu_instance,
- cloud_name=self.cluster_uuid,
- credential_name="cred-{}".format(self.cluster_uuid),
- )
+ self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
+ instantiation_params=None,
)
super(ExecPrimitivesTest, self).setUp()
self.action_name = "touch"
self.application_name = "myapp"
- self.model_name = "model"
self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
self.k8s_juju_conn.libjuju.execute_action = AsyncMock()
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertEqual(output, "success")
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
def test_exception(self):
with self.assertRaises(Exception):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
def test_missing_application_name_in_params(self):
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name
+ "cluster", self.kdu_instance, self.action_name
)
)
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, "non-existing-action", params=params
+ "cluster", self.kdu_instance, "non-existing-action", params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
def setUp(self):
super(GetServicesTest, self).setUp()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
@asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
- def test_success(self, mock_get_credentials, mock_get_services):
+ def test_success(self, mock_get_credentials):
mock_get_credentials.return_value = kubeconfig
self.loop.run_until_complete(self.k8s_juju_conn.get_services("", "", ""))
mock_get_credentials.assert_called_once()
- mock_get_services.assert_called_once()
+ self.kubectl.get_services.assert_called_once()
class GetServiceTest(K8sJujuConnTestCase):
def setUp(self):
super(GetServiceTest, self).setUp()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
@asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
- def test_success(self, mock_get_credentials, mock_get_services):
+ def test_success(self, mock_get_credentials):
mock_get_credentials.return_value = kubeconfig
self.loop.run_until_complete(self.k8s_juju_conn.get_service("", "", ""))
mock_get_credentials.assert_called_once()
- mock_get_services.assert_called_once()
+ self.kubectl.get_services.assert_called_once()
class GetCredentialsTest(K8sJujuConnTestCase):
def setUp(self):
super(UpdateVcaStatusTest, self).setUp()
self.vcaStatus = {"model": {"applications": {"app": {"actions": {}}}}}
- self.kdu_name = "kdu_name"
- self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
self.k8s_juju_conn.libjuju.get_executed_actions = AsyncMock()
self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
self.k8s_juju_conn.libjuju.get_application_configs = AsyncMock()
def test_success(self):
- self.loop.run_until_complete(self.k8s_juju_conn.update_vca_status(
- self.vcaStatus, self.kdu_instance))
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
self.k8s_juju_conn.libjuju.get_executed_actions.assert_called_once()
- self.k8s_juju_conn.libjuju.get_actions.assert_called_once()
self.k8s_juju_conn.libjuju.get_application_configs.assert_called_once()
def test_exception(self):
self.k8s_juju_conn.libjuju.get_model.return_value = None
self.k8s_juju_conn.libjuju.get_executed_actions.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.k8s_juju_conn.update_vca_status(
- self.vcaStatus, self.kdu_instance))
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
self.k8s_juju_conn.libjuju.get_executed_actions.assert_not_called()
- self.k8s_juju_conn.libjuju.get_actions.assert_not_called_once()
self.k8s_juju_conn.libjuju.get_application_configs.assert_not_called_once()
+
+
+class ScaleTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(ScaleTest, self).setUp()
+ self.application_name = "app"
+ self.kdu_name = "kdu-instance"
+ self._scale = 2
+ self.k8s_juju_conn.libjuju.scale_application = AsyncMock()
+
+ def test_success(self):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(self.kdu_name, self._scale, self.application_name)
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.scale_application.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(
+ self.kdu_name, self._scale, self.application_name
+ )
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+
+class GetScaleCount(K8sJujuConnTestCase):
+ def setUp(self):
+ super(GetScaleCount, self).setUp()
+ self.k8s_juju_conn.libjuju.get_model_status = AsyncMock()
+
+ def test_success(self):
+ applications = {"app": FakeApplication()}
+ model = FakeModel(applications=applications)
+ self.k8s_juju_conn.libjuju.get_model_status.return_value = model
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.get_scale_count("app", "kdu_instance")
+ )
+ self.assertEqual(status, 2)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.get_model_status.side_effect = Exception()
+ status = None
+ with self.assertRaises(Exception):
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.status_kdu("app", "kdu_instance")
+ )
+ self.assertIsNone(status)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+
+class AddRelationTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(AddRelationTest, self).setUp()
+ self.k8s_juju_conn.libjuju.add_relation = AsyncMock()
+ self.k8s_juju_conn.libjuju.offer = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_controller = AsyncMock()
+ self.k8s_juju_conn.libjuju.consume = AsyncMock()
+
+ def test_standard_relation_same_model_and_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint1")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint2")
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ model_name="model-1",
+ endpoint_1="app1:endpoint1",
+ endpoint_2="app2:endpoint2",
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_not_called()
+ self.k8s_juju_conn.libjuju.consume.assert_not_called()
+
+ def test_cmr_relation_same_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.k8s_juju_conn.libjuju.offer.return_value = offer
+ self.k8s_juju_conn.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.k8s_juju_conn.libjuju.consume.assert_called_once()
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ "model-2", "app2:endpoint", "saas"
+ )
+
+ def test_cmr_relation_different_controller(self):
+ self.k8s_juju_conn._get_libjuju = AsyncMock(
+ return_value=self.k8s_juju_conn.libjuju
+ )
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", "vca-id-1", "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", "vca-id-2", "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.k8s_juju_conn.libjuju.offer.return_value = offer
+ self.k8s_juju_conn.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.k8s_juju_conn.libjuju.consume.assert_called_once()
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ "model-1", "app2:endpoint", "saas"
+ )
+
+ def test_relation_exception(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ self.k8s_juju_conn.libjuju.offer.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(
+ relation_endpoint_1, relation_endpoint_2
+ )
+ )
projects / osm / N2VC.git / blobdiff