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 osm_common.fslocal import FsLocal
  24 from asynctest.mock import Mock
  25 from osm_common.dbmemory import DbMemory
  26
  27 __author__ = "Isabel Lloret <illoret@indra.es>"
  28
  29
  30 class TestLcmHelmConn(asynctest.TestCase):
  31     logging.basicConfig(level=logging.DEBUG)
  32     logger = logging.getLogger(__name__)
  33     logger.setLevel(logging.DEBUG)
  34
  35     async def setUp(self):
  36         self.db = Mock(DbMemory())
  37         self.fs = asynctest.Mock(FsLocal())
  38         self.fs.path = "/app/storage"
  39         vca_config = {}
  40         lcm_helm_conn.K8sHelmConnector = asynctest.Mock(lcm_helm_conn.K8sHelmConnector)
  41         self.helm_conn = LCMHelmConn(self.db, self.fs, loop=self.loop, vca_config=vca_config, log=self.logger)
  42
  43     @asynctest.fail_on(active_handles=True)
  44     async def test_create_execution_environment(self):
  45         namespace = "testnamespace"
  46         db_dict = {}
  47         artifact_path = "helm_sample_charm"
  48         helm_chart_id = "helm_sample_charm_0001"
  49         self.helm_conn._k8sclusterhelm.install = asynctest.CoroutineMock(return_value=helm_chart_id)
  50         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
  51         ee_id, _ = await self.helm_conn.create_execution_environment(namespace, db_dict, artifact_path=artifact_path)
  52         self.assertEqual(ee_id, "{}.{}".format("osm", helm_chart_id),
  53                          "Check ee_id format: <default namespace>.<helm_chart-id>")
  54         self.helm_conn._k8sclusterhelm.install.assert_called_once_with("myk8s_id",
  55                                                                        kdu_model="/app/storage/helm_sample_charm",
  56                                                                        namespace="osm", db_dict=db_dict,
  57                                                                        params=None, timeout=None)
  58
  59     @asynctest.fail_on(active_handles=True)
  60     async def test_get_ee_ssh_public__key(self):
  61         ee_id = "osm.helm_sample_charm_0001"
  62         db_dict = {}
  63         lcm_helm_conn.socket.gethostbyname = asynctest.Mock()
  64         mock_pub_key = "ssh-rsapubkey"
  65         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
  66         self.helm_conn._get_ssh_key = asynctest.CoroutineMock(return_value=mock_pub_key)
  67         pub_key = await self.helm_conn.get_ee_ssh_public__key(ee_id=ee_id, db_dict=db_dict)
  68         self.assertEqual(pub_key, mock_pub_key)
  69
  70     @asynctest.fail_on(active_handles=True)
  71     async def test_execute_primitive(self):
  72         lcm_helm_conn.socket.gethostbyname = asynctest.Mock()
  73         ee_id = "osm.helm_sample_charm_0001"
  74         primitive_name = "sleep"
  75         params = {}
  76         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
  77         self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(return_value=("OK", "test-ok"))
  78         message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params)
  79         self.assertEqual(message, "test-ok")
  80
  81     @asynctest.fail_on(active_handles=True)
  82     async def test_execute_config_primitive(self):
  83         self.logger.debug("Execute config primitive")
  84         lcm_helm_conn.socket.gethostbyname = asynctest.Mock()
  85         ee_id = "osm.helm_sample_charm_0001"
  86         primitive_name = "config"
  87         params = {"ssh-host-name": "host1"}
  88         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
  89         self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(return_value=("OK", "CONFIG OK"))
  90         message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params)
  91         self.assertEqual(message, "CONFIG OK")
  92
  93     @asynctest.fail_on(active_handles=True)
  94     async def test_delete_execution_environment(self):
  95         ee_id = "osm.helm_sample_charm_0001"
  96         self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
  97         self.helm_conn._k8sclusterhelm.uninstall = asynctest.CoroutineMock()
  98         await self.helm_conn.delete_execution_environment(ee_id)
  99         self.helm_conn._k8sclusterhelm.uninstall.assert_called_once_with("myk8s_id", "helm_sample_charm_0001")
 100
 101
 102 if __name__ == '__main__':
 103     asynctest.main()