import asyncio
import logging
import asynctest
-from n2vc.k8s_juju_conn import K8sJujuConnector
+from unittest.mock import Mock
+from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
-from .utils import kubeconfig, FakeModel, FakeFileWrapper
+from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock
from n2vc.exceptions import (
MethodNotImplemented,
K8sException,
N2VCBadArgumentsException,
)
-from unittest.mock import Mock
-from .utils import AsyncMock
class K8sJujuConnTestCase(asynctest.TestCase):
)
+@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._get_secret_data = AsyncMock()
+ self.k8s_juju_conn._get_secret_data.return_value = ("token", "cacert")
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
def test_with_cluster_uuid(
self,
mock_get_default_storage_class,
mock_get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
- def test_with_no_cluster_uuid(self, mock_get_default_storage_class):
+ 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)
)
mock_get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
- def test_init_env_exception(self, mock_get_default_storage_class):
+ def test_init_env_exception(
+ self,
+ mock_get_default_storage_class,
+ ):
self.k8s_juju_conn.libjuju.add_k8s.side_effect = Exception()
created = None
uuid = None
def setUp(self):
super(ResetTest, self).setUp()
self.k8s_juju_conn.libjuju.remove_cloud = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_cloud_credentials = AsyncMock()
+ 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
def test_success(self):
removed = self.loop.run_until_complete(self.k8s_juju_conn.reset("uuid"))