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=8f777e188e76faa9d6e11351fe81c81312dad594;hpb=18ebc3ad0e546dfa7ce649d336bd1c838b38d94c;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 8f777e1..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,17 +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" + 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" + "kubectlpath": "/usr/bin/kubectl", } lcm_helm_conn.K8sHelmConnector = asynctest.Mock(lcm_helm_conn.K8sHelmConnector) - lcm_helm_conn.K8sHelm3Connector = asynctest.Mock(lcm_helm_conn.K8sHelm3Connector) - 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): @@ -51,18 +64,36 @@ class TestLcmHelmConn(asynctest.TestCase): db_dict = {} artifact_path = "helm_sample_charm" helm_chart_id = "helm_sample_charm_0001" - self.helm_conn._k8sclusterhelm3.install = asynctest.CoroutineMock(return_value=helm_chart_id) + 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="/app/storage/helm_sample_charm", - namespace="osm", db_dict=db_dict, - params=None, timeout=None) + 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): @@ -72,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) @@ -82,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") @@ -94,7 +129,9 @@ 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") @@ -102,10 +139,14 @@ class TestLcmHelmConn(asynctest.TestCase): async def test_delete_execution_environment(self): 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="") + self.helm_conn._k8sclusterhelm3.uninstall = asynctest.CoroutineMock( + return_value="" + ) await self.helm_conn.delete_execution_environment(ee_id) - self.helm_conn._k8sclusterhelm3.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()