self.db = Mock(DbMemory())
self.fs = asynctest.Mock(FsLocal())
self.fs.path = "/app/storage"
- vca_config = {}
+ 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)
+ 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)
@asynctest.fail_on(active_handles=True)
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: <default namespace>.<helm_chart-id>")
- 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=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: <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",
+ 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):
@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__':