import asynctest
import logging
-from asynctest.mock import Mock
+from asynctest.mock import Mock, patch
from osm_common.dbmemory import DbMemory
from osm_common.fslocal import FsLocal
-from n2vc.k8s_helm3_conn import K8sHelm3Connector
+from n2vc.k8s_helm3_conn import K8sHelm3Connector, K8sException
__author__ = "Isabel Lloret <illoret@indra.es>"
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
- async def setUp(self):
+ @patch("n2vc.k8s_helm_base_conn.EnvironConfig")
+ async def setUp(self, mock_env):
+ mock_env.return_value = {
+ "stablerepourl": "https://charts.helm.sh/stable"
+ }
self.db = Mock(DbMemory())
self.fs = asynctest.Mock(FsLocal())
self.fs.path = "./tmp/"
self.kdu_instance = "stable-openldap-0005399828"
self.helm_conn.generate_kdu_instance_name = Mock(return_value=self.kdu_instance)
self.helm_conn._get_namespaces = asynctest.CoroutineMock(return_value=[])
+ self.helm_conn._namespace_exists = asynctest.CoroutineMock(side_effect=self.helm_conn._namespace_exists)
self.helm_conn._create_namespace = asynctest.CoroutineMock()
await self.helm_conn.install(
db_dict=db_dict,
)
+ self.helm_conn._namespace_exists.assert_called_once()
self.helm_conn._get_namespaces.assert_called_once()
self.helm_conn._create_namespace.assert_called_once_with(
self.cluster_id, self.namespace
command=command, env=self.env, raise_exception_on_error=False
)
+ # Exception test if namespace could not being created for some reason
+ self.helm_conn._namespace_exists.return_value = False
+ self.helm_conn._create_namespace.side_effect = Exception()
+
+ with self.assertRaises(K8sException):
+ await self.helm_conn.install(
+ self.cluster_uuid,
+ kdu_model,
+ self.kdu_instance,
+ atomic=True,
+ namespace=self.namespace,
+ db_dict=db_dict,
+ )
+
+ @asynctest.fail_on(active_handles=True)
+ async def test_namespace_exists(self):
+ self.helm_conn._get_namespaces = asynctest.CoroutineMock()
+
+ self.helm_conn._get_namespaces.return_value = ['testk8s', 'kube-system']
+ result = await self.helm_conn._namespace_exists(self.cluster_id, self.namespace)
+ self.helm_conn._get_namespaces.assert_called_once()
+ self.assertEqual(result, True)
+
+ self.helm_conn._get_namespaces.reset_mock()
+ result = await self.helm_conn._namespace_exists(self.cluster_id, 'none-exists-namespace')
+ self.helm_conn._get_namespaces.assert_called_once()
+ self.assertEqual(result, False)
+
@asynctest.fail_on(active_handles=True)
async def test_upgrade(self):
kdu_model = "stable/openldap:1.2.3"