Fix Bug 1575

[osm/N2VC.git] / n2vc / tests / unit / test_k8s_helm3_conn.py

diff --git a/n2vc/tests/unit/test_k8s_helm3_conn.py b/n2vc/tests/unit/test_k8s_helm3_conn.py
index b1b5a60..5f43804 100644 (file)
--- a/n2vc/tests/unit/test_k8s_helm3_conn.py
+++ b/n2vc/tests/unit/test_k8s_helm3_conn.py
@@ -21,7 +21,7 @@ import logging
 from asynctest.mock import Mock
 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>"
 
@@ -134,25 +134,26 @@ class TestK8sHelm3Conn(asynctest.TestCase):
         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
         self.helm_conn._status_kdu = asynctest.CoroutineMock(return_value=None)
         self.helm_conn._store_status = asynctest.CoroutineMock()
-        self.helm_conn._generate_release_name = Mock(return_value="stable-openldap-0005399828")
+        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()
 
-        kdu_instance = await self.helm_conn.install(self.cluster_uuid,
-                                                    kdu_model,
-                                                    atomic=True,
-                                                    namespace=self.namespace,
-                                                    db_dict=db_dict)
+        await self.helm_conn.install(
+            self.cluster_uuid,
+            kdu_model,
+            self.kdu_instance,
+            atomic=True,
+            namespace=self.namespace,
+            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)
         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
         self.helm_conn.fs.reverse_sync.assert_called_once_with(from_path=self.cluster_id)
-        self.helm_conn._generate_release_name.assert_called_once_with("stable/openldap")
-        self.helm_conn._status_kdu.assert_called_once_with(cluster_id=self.cluster_id,
-                                                           kdu_instance=kdu_instance,
-                                                           namespace=self.namespace,
-                                                           show_error_log=False)
         self.helm_conn._store_status.assert_called_with(cluster_id=self.cluster_id,
                                                         kdu_instance=kdu_instance,
                                                         namespace=self.namespace,
@@ -166,6 +167,34 @@ class TestK8sHelm3Conn(asynctest.TestCase):
                                                                  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"
@@ -446,7 +475,7 @@ class TestK8sHelm3Conn(asynctest.TestCase):
             if args[0] == "k8sclusters":
                 return {
                     "_admin": {
-                        "helm_chart_repos" : [
+                        "helm_chart_repos": [
                             "4b5550a9-990d-4d95-8a48-1f4614d6ac9c"
                         ]
                     }