from asynctest.mock import Mock
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.filesystem.filesystem import Filesystem
from asynctest.mock import Mock
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.filesystem.filesystem import Filesystem
- self.helm_conn = LCMHelmConn(
- loop=self.loop, vca_config=vca_config, log=self.logger
- )
+ 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"
@asynctest.fail_on(active_handles=True)
async def test_create_execution_environment(self):
namespace = "testnamespace"
db_dict = {}
artifact_path = "helm_sample_charm"
helm_chart_id = "helm_sample_charm_0001"
self.helm_conn._k8sclusterhelm3.install = asynctest.CoroutineMock(
return_value=None
helm_chart_id = "helm_sample_charm_0001"
self.helm_conn._k8sclusterhelm3.install = asynctest.CoroutineMock(
return_value=None
self.db.get_one.return_value = {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}}
ee_id, _ = await self.helm_conn.create_execution_environment(
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"
+ namespace,
+ db_dict,
+ artifact_path=artifact_path,
+ chart_model=chart_model,
+ vca_type="helm-v3",
- "{}:{}.{}".format("helm-v3", "osm", helm_chart_id),
- "Check ee_id format: <helm-version>:<default namespace>.<helm_chart-id>",
+ "{}:{}.{}".format("helm-v3", namespace, helm_chart_id),
+ "Check ee_id format: <helm-version>:<NS ID>.<helm_chart-id>",
)
self.helm_conn._k8sclusterhelm3.install.assert_called_once_with(
"myk8s_id",
kdu_model="/helm_sample_charm",
kdu_instance=helm_chart_id,
)
self.helm_conn._k8sclusterhelm3.install.assert_called_once_with(
"myk8s_id",
kdu_model="/helm_sample_charm",
kdu_instance=helm_chart_id,
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)
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)
@asynctest.fail_on(active_handles=True)
async def test_execute_config_primitive(self):
self.logger.debug("Execute config primitive")
@asynctest.fail_on(active_handles=True)
async def test_execute_config_primitive(self):
self.logger.debug("Execute config primitive")
ee_id = "osm.helm_sample_charm_0001"
primitive_name = "config"
params = {"ssh-host-name": "host1"}
ee_id = "osm.helm_sample_charm_0001"
primitive_name = "config"
params = {"ssh-host-name": "host1"}