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 <illoret@indra.es>"
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",
}
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)
+ 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):
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,
self.assertEqual(ee_id, "{}:{}.{}".format("helm-v3", "osm", helm_chart_id),
"Check ee_id format: <helm-version>:<default namespace>.<helm_chart-id>")
self.helm_conn._k8sclusterhelm3.install.assert_called_once_with("myk8s_id",
- kdu_model="/app/storage/helm_sample_charm",
+ kdu_model="/helm_sample_charm",
+ kdu_instance=helm_chart_id,
namespace="osm", db_dict=db_dict,
params=None, timeout=None)