osm_lcm/tests/test_lcm_helm_conn.py

   1 ##
   2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   3 # not use this file except in compliance with the License. You may obtain
   4 # a copy of the License at
   5 #
   6 #         http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  11 # License for the specific language governing permissions and limitations
  12 # under the License.
  13 #
  14 # For those usages not covered by the Apache License, Version 2.0 please
  15 # contact: alfonso.tiernosepulveda@telefonica.com
  16 ##
  17
  18 import asynctest
  19 import logging
  20
  21 from osm_lcm import lcm_helm_conn
  22 from osm_lcm.lcm_helm_conn import LCMHelmConn
  23 from asynctest.mock import Mock
  24 from osm_lcm.data_utils.database.database import Database
  25 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
  26 from osm_lcm.data_utils.lcm_config import VcaConfig
  27
  28 __author__ = "Isabel Lloret <illoret@indra.es>"
  29
  30
  31 class TestLcmHelmConn(asynctest.TestCase):
  32     logging.basicConfig(level=logging.DEBUG)
  33     logger = logging.getLogger(__name__)
  34     logger.setLevel(logging.DEBUG)
  35
  36     async def setUp(self):
  37         Database.instance = None
  38         self.db = Mock(Database({"database": {"driver": "memory"}}).instance.db)
  39         Database().instance.db = self.db
  40
  41         Filesystem.instance = None
  42         self.fs = asynctest.Mock(
  43             Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
  44         )
  45
  46         Filesystem.instance.fs = self.fs
  47         self.fs.path = "/"
  48
  49         vca_config = {
  50             "helmpath": "/usr/local/bin/helm",
  51             "helm3path": "/usr/local/bin/helm3",
  52             "kubectlpath": "/usr/bin/kubectl",
  53         }
  54         lcm_helm_conn.K8sHelm3Connector = asynctest.Mock(
  55             lcm_helm_conn.K8sHelm3Connector
  56         )
  57         vca_config = VcaConfig(vca_config)
  58         self.helm_conn = LCMHelmConn(vca_config=vca_config, log=self.logger)
  59
  60     @asynctest.fail_on(active_handles=True)
  61     async def test_create_execution_environment(self):
  62         namespace = "testnamespace"
  63         db_dict = {}
  64         artifact_path = "helm_sample_charm"
  65         chart_model = "helm_sample_charm"
  66         helm_chart_id = "helm_sample_charm_0001"
  67         self.helm_conn._k8sclusterhelm3.install = asynctest.CoroutineMock(
  68             return_value=None
  69         )
  70         self.helm_conn._k8sclusterhelm3.generate_kdu_instance_name = Mock()
  71         self.helm_conn._k8sclusterhelm3.generate_kdu_instance_name.return_value = (
  72             helm_chart_id
  73         )
  74
  75         self.db.get_one.return_value = {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}}
  76         ee_id, _ = await self.helm_conn.create_execution_environment(
  77             namespace,
  78             db_dict,
  79             artifact_path=artifact_path,
  80             chart_model=chart_model,
  81             vca_type="helm-v3",
  82         )
  83         self.assertEqual(
  84             ee_id,
  85             "{}:{}.{}".format("helm-v3", namespace, helm_chart_id),
  86             "Check ee_id format: <helm-version>:<NS ID>.<helm_chart-id>",
  87         )
  88         self.helm_conn._k8sclusterhelm3.install.assert_called_once_with(
  89             "myk8s_id",
  90             kdu_model="/helm_sample_charm",
  91             kdu_instance=helm_chart_id,
  92             namespace=namespace,
  93             db_dict=db_dict,
  94             params=None,
  95             timeout=None,
  96         )
  97
  98     @asynctest.fail_on(active_handles=True)
  99     async def test_get_ee_ssh_public__key(self):
 100         ee_id = "osm.helm_sample_charm_0001"
 101         db_dict = {}
 102         mock_pub_key = "ssh-rsapubkey"
 103         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
 104         self.helm_conn._get_ssh_key = asynctest.CoroutineMock(return_value=mock_pub_key)
 105         pub_key = await self.helm_conn.get_ee_ssh_public__key(
 106             ee_id=ee_id, db_dict=db_dict
 107         )
 108         self.assertEqual(pub_key, mock_pub_key)
 109
 110     @asynctest.fail_on(active_handles=True)
 111     async def test_execute_primitive(self):
 112         ee_id = "osm.helm_sample_charm_0001"
 113         primitive_name = "sleep"
 114         params = {}
 115         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
 116         self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(
 117             return_value=("OK", "test-ok")
 118         )
 119         message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params)
 120         self.assertEqual(message, "test-ok")
 121
 122     @asynctest.fail_on(active_handles=True)
 123     async def test_execute_config_primitive(self):
 124         self.logger.debug("Execute config primitive")
 125         ee_id = "osm.helm_sample_charm_0001"
 126         primitive_name = "config"
 127         params = {"ssh-host-name": "host1"}
 128         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
 129         self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(
 130             return_value=("OK", "CONFIG OK")
 131         )
 132         message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params)
 133         self.assertEqual(message, "CONFIG OK")
 134
 135     @asynctest.fail_on(active_handles=True)
 136     async def test_delete_execution_environment(self):
 137         ee_id = "helm-v3:osm.helm_sample_charm_0001"
 138         self.db.get_one.return_value = {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}}
 139         self.helm_conn._k8sclusterhelm3.uninstall = asynctest.CoroutineMock(
 140             return_value=""
 141         )
 142         await self.helm_conn.delete_execution_environment(ee_id)
 143         self.helm_conn._k8sclusterhelm3.uninstall.assert_called_once_with(
 144             "myk8s_id", "helm_sample_charm_0001"
 145         )
 146
 147
 148 if __name__ == "__main__":
 149     asynctest.main()