X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Ftests%2Ftest_lcm_helm_conn.py;h=89ca891c382ba1a578fd5b2a2fc685e7cc129466;hb=4554a700d861c477360b6360920a5cb8c1f61718;hp=ae92c5eaa1a4575c7b19d940eea6af3a446e5c7c;hpb=6de77721b4b59605d0e79c99561b1731355f5b68;p=osm%2FLCM.git diff --git a/osm_lcm/tests/test_lcm_helm_conn.py b/osm_lcm/tests/test_lcm_helm_conn.py index ae92c5e..89ca891 100644 --- a/osm_lcm/tests/test_lcm_helm_conn.py +++ b/osm_lcm/tests/test_lcm_helm_conn.py @@ -20,9 +20,9 @@ import logging from osm_lcm import lcm_helm_conn from osm_lcm.lcm_helm_conn import LCMHelmConn -from osm_common.fslocal import FsLocal from asynctest.mock import Mock -from osm_common.dbmemory import DbMemory +from osm_lcm.data_utils.database.database import Database +from osm_lcm.data_utils.filesystem.filesystem import Filesystem __author__ = "Isabel Lloret " @@ -33,12 +33,30 @@ class TestLcmHelmConn(asynctest.TestCase): logger.setLevel(logging.DEBUG) async def setUp(self): - self.db = Mock(DbMemory()) - self.fs = asynctest.Mock(FsLocal()) - self.fs.path = "/app/storage" - vca_config = {} + Database.instance = None + self.db = Mock(Database({"database": {"driver": "memory"}}).instance.db) + Database().instance.db = self.db + + Filesystem.instance = None + self.fs = asynctest.Mock( + Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs + ) + + Filesystem.instance.fs = self.fs + self.fs.path = "/" + + vca_config = { + "helmpath": "/usr/local/bin/helm", + "helm3path": "/usr/local/bin/helm3", + "kubectlpath": "/usr/bin/kubectl", + } lcm_helm_conn.K8sHelmConnector = asynctest.Mock(lcm_helm_conn.K8sHelmConnector) - self.helm_conn = LCMHelmConn(self.db, self.fs, loop=self.loop, vca_config=vca_config, log=self.logger) + lcm_helm_conn.K8sHelm3Connector = asynctest.Mock( + lcm_helm_conn.K8sHelm3Connector + ) + self.helm_conn = LCMHelmConn( + loop=self.loop, vca_config=vca_config, log=self.logger + ) @asynctest.fail_on(active_handles=True) async def test_create_execution_environment(self): @@ -46,15 +64,36 @@ class TestLcmHelmConn(asynctest.TestCase): db_dict = {} artifact_path = "helm_sample_charm" helm_chart_id = "helm_sample_charm_0001" - self.helm_conn._k8sclusterhelm.install = asynctest.CoroutineMock(return_value=helm_chart_id) - self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}} - ee_id, _ = await self.helm_conn.create_execution_environment(namespace, db_dict, artifact_path=artifact_path) - self.assertEqual(ee_id, "{}.{}".format("osm", helm_chart_id), - "Check ee_id format: .") - self.helm_conn._k8sclusterhelm.install.assert_called_once_with("myk8s_id", - kdu_model="/app/storage/helm_sample_charm", - namespace="osm", db_dict=db_dict, - params=None, timeout=None) + self.helm_conn._k8sclusterhelm3.install = asynctest.CoroutineMock( + return_value=None + ) + self.helm_conn._k8sclusterhelm3.generate_kdu_instance_name = Mock() + self.helm_conn._k8sclusterhelm3.generate_kdu_instance_name.return_value = ( + helm_chart_id + ) + self.helm_conn._k8sclusterhelm2.generate_kdu_instance_name = Mock() + self.helm_conn._k8sclusterhelm2.generate_kdu_instance_name.return_value = ( + helm_chart_id + ) + + self.db.get_one.return_value = {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}} + ee_id, _ = await self.helm_conn.create_execution_environment( + namespace, db_dict, artifact_path=artifact_path, vca_type="helm-v3" + ) + self.assertEqual( + ee_id, + "{}:{}.{}".format("helm-v3", "osm", helm_chart_id), + "Check ee_id format: :.", + ) + self.helm_conn._k8sclusterhelm3.install.assert_called_once_with( + "myk8s_id", + kdu_model="/helm_sample_charm", + kdu_instance=helm_chart_id, + namespace="osm", + db_dict=db_dict, + params=None, + timeout=None, + ) @asynctest.fail_on(active_handles=True) async def test_get_ee_ssh_public__key(self): @@ -64,7 +103,9 @@ class TestLcmHelmConn(asynctest.TestCase): mock_pub_key = "ssh-rsapubkey" self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}} self.helm_conn._get_ssh_key = asynctest.CoroutineMock(return_value=mock_pub_key) - pub_key = await self.helm_conn.get_ee_ssh_public__key(ee_id=ee_id, db_dict=db_dict) + pub_key = await self.helm_conn.get_ee_ssh_public__key( + ee_id=ee_id, db_dict=db_dict + ) self.assertEqual(pub_key, mock_pub_key) @asynctest.fail_on(active_handles=True) @@ -74,7 +115,9 @@ class TestLcmHelmConn(asynctest.TestCase): primitive_name = "sleep" params = {} self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}} - self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(return_value=("OK", "test-ok")) + self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock( + return_value=("OK", "test-ok") + ) message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params) self.assertEqual(message, "test-ok") @@ -86,18 +129,24 @@ class TestLcmHelmConn(asynctest.TestCase): primitive_name = "config" params = {"ssh-host-name": "host1"} self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}} - self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock(return_value=("OK", "CONFIG OK")) + self.helm_conn._execute_primitive_internal = asynctest.CoroutineMock( + return_value=("OK", "CONFIG OK") + ) message = await self.helm_conn.exec_primitive(ee_id, primitive_name, params) self.assertEqual(message, "CONFIG OK") @asynctest.fail_on(active_handles=True) async def test_delete_execution_environment(self): - ee_id = "osm.helm_sample_charm_0001" - self.db.get_one.return_value = {"_admin": {"helm-chart": {"id": "myk8s_id"}}} - self.helm_conn._k8sclusterhelm.uninstall = asynctest.CoroutineMock() + ee_id = "helm-v3:osm.helm_sample_charm_0001" + self.db.get_one.return_value = {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}} + self.helm_conn._k8sclusterhelm3.uninstall = asynctest.CoroutineMock( + return_value="" + ) await self.helm_conn.delete_execution_environment(ee_id) - self.helm_conn._k8sclusterhelm.uninstall.assert_called_once_with("myk8s_id", "helm_sample_charm_0001") + self.helm_conn._k8sclusterhelm3.uninstall.assert_called_once_with( + "myk8s_id", "helm_sample_charm_0001" + ) -if __name__ == '__main__': +if __name__ == "__main__": asynctest.main()