X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Ftests%2Ftest_lcm_helm_conn.py;h=051fff855dbea20dfaf40332b3584eb8a243acf1;hb=b6049d378301d13e0ddaadb8433b8ad20f4f8a23;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..051fff8 100644 --- a/osm_lcm/tests/test_lcm_helm_conn.py +++ b/osm_lcm/tests/test_lcm_helm_conn.py @@ -20,9 +20,10 @@ 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 +from osm_lcm.data_utils.lcm_config import VcaConfig __author__ = "Isabel Lloret " @@ -33,71 +34,116 @@ 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 = {} - 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) + 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.K8sHelm3Connector = asynctest.Mock( + lcm_helm_conn.K8sHelm3Connector + ) + vca_config = VcaConfig(vca_config) + self.helm_conn = LCMHelmConn(vca_config=vca_config, log=self.logger) @asynctest.fail_on(active_handles=True) async def test_create_execution_environment(self): namespace = "testnamespace" db_dict = {} artifact_path = "helm_sample_charm" + chart_model = "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.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, + chart_model=chart_model, + vca_type="helm-v3", + ) + self.assertEqual( + ee_id, + "{}:{}.{}".format("helm-v3", namespace, 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=namespace, + db_dict=db_dict, + params=None, + timeout=None, + ) @asynctest.fail_on(active_handles=True) async def test_get_ee_ssh_public__key(self): ee_id = "osm.helm_sample_charm_0001" db_dict = {} - lcm_helm_conn.socket.gethostbyname = asynctest.Mock() 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) async def test_execute_primitive(self): - lcm_helm_conn.socket.gethostbyname = asynctest.Mock() ee_id = "osm.helm_sample_charm_0001" 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") @asynctest.fail_on(active_handles=True) async def test_execute_config_primitive(self): self.logger.debug("Execute config primitive") - lcm_helm_conn.socket.gethostbyname = asynctest.Mock() ee_id = "osm.helm_sample_charm_0001" 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()