2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: alfonso.tiernosepulveda@telefonica.com
21 from osm_lcm
import lcm_helm_conn
22 from osm_lcm
.lcm_helm_conn
import LCMHelmConn
23 from asynctest
.mock
import Mock
24 from osm_lcm
.data_utils
.database
.database
import Database
25 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
26 from osm_lcm
.data_utils
.lcm_config
import VcaConfig
28 __author__
= "Isabel Lloret <illoret@indra.es>"
31 class TestLcmHelmConn(asynctest
.TestCase
):
32 logging
.basicConfig(level
=logging
.DEBUG
)
33 logger
= logging
.getLogger(__name__
)
34 logger
.setLevel(logging
.DEBUG
)
36 async def setUp(self
):
37 Database
.instance
= None
38 self
.db
= Mock(Database({"database": {"driver": "memory"}}).instance
.db
)
39 Database().instance
.db
= self
.db
41 Filesystem
.instance
= None
42 self
.fs
= asynctest
.Mock(
43 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
46 Filesystem
.instance
.fs
= self
.fs
50 "helmpath": "/usr/local/bin/helm",
51 "helm3path": "/usr/local/bin/helm3",
52 "kubectlpath": "/usr/bin/kubectl",
54 lcm_helm_conn
.K8sHelmConnector
= asynctest
.Mock(lcm_helm_conn
.K8sHelmConnector
)
55 lcm_helm_conn
.K8sHelm3Connector
= asynctest
.Mock(
56 lcm_helm_conn
.K8sHelm3Connector
58 vca_config
= VcaConfig(vca_config
)
59 self
.helm_conn
= LCMHelmConn(
60 loop
=self
.loop
, vca_config
=vca_config
, log
=self
.logger
63 @asynctest.fail_on(active_handles
=True)
64 async def test_create_execution_environment(self
):
65 namespace
= "testnamespace"
67 artifact_path
= "helm_sample_charm"
68 chart_model
= "helm_sample_charm"
69 helm_chart_id
= "helm_sample_charm_0001"
70 self
.helm_conn
._k
8sclusterhelm
3.install
= asynctest
.CoroutineMock(
73 self
.helm_conn
._k
8sclusterhelm
3.generate_kdu_instance_name
= Mock()
74 self
.helm_conn
._k
8sclusterhelm
3.generate_kdu_instance_name
.return_value
= (
77 self
.helm_conn
._k
8sclusterhelm
2.generate_kdu_instance_name
= Mock()
78 self
.helm_conn
._k
8sclusterhelm
2.generate_kdu_instance_name
.return_value
= (
82 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}}
83 ee_id
, _
= await self
.helm_conn
.create_execution_environment(
86 artifact_path
=artifact_path
,
87 chart_model
=chart_model
,
92 "{}:{}.{}".format("helm-v3", "osm", helm_chart_id
),
93 "Check ee_id format: <helm-version>:<default namespace>.<helm_chart-id>",
95 self
.helm_conn
._k
8sclusterhelm
3.install
.assert_called_once_with(
97 kdu_model
="/helm_sample_charm",
98 kdu_instance
=helm_chart_id
,
105 @asynctest.fail_on(active_handles
=True)
106 async def test_get_ee_ssh_public__key(self
):
107 ee_id
= "osm.helm_sample_charm_0001"
109 mock_pub_key
= "ssh-rsapubkey"
110 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
111 self
.helm_conn
._get
_ssh
_key
= asynctest
.CoroutineMock(return_value
=mock_pub_key
)
112 pub_key
= await self
.helm_conn
.get_ee_ssh_public__key(
113 ee_id
=ee_id
, db_dict
=db_dict
115 self
.assertEqual(pub_key
, mock_pub_key
)
117 @asynctest.fail_on(active_handles
=True)
118 async def test_execute_primitive(self
):
119 ee_id
= "osm.helm_sample_charm_0001"
120 primitive_name
= "sleep"
122 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
123 self
.helm_conn
._execute
_primitive
_internal
= asynctest
.CoroutineMock(
124 return_value
=("OK", "test-ok")
126 message
= await self
.helm_conn
.exec_primitive(ee_id
, primitive_name
, params
)
127 self
.assertEqual(message
, "test-ok")
129 @asynctest.fail_on(active_handles
=True)
130 async def test_execute_config_primitive(self
):
131 self
.logger
.debug("Execute config primitive")
132 ee_id
= "osm.helm_sample_charm_0001"
133 primitive_name
= "config"
134 params
= {"ssh-host-name": "host1"}
135 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
136 self
.helm_conn
._execute
_primitive
_internal
= asynctest
.CoroutineMock(
137 return_value
=("OK", "CONFIG OK")
139 message
= await self
.helm_conn
.exec_primitive(ee_id
, primitive_name
, params
)
140 self
.assertEqual(message
, "CONFIG OK")
142 @asynctest.fail_on(active_handles
=True)
143 async def test_delete_execution_environment(self
):
144 ee_id
= "helm-v3:osm.helm_sample_charm_0001"
145 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart-v3": {"id": "myk8s_id"}}}
146 self
.helm_conn
._k
8sclusterhelm
3.uninstall
= asynctest
.CoroutineMock(
149 await self
.helm_conn
.delete_execution_environment(ee_id
)
150 self
.helm_conn
._k
8sclusterhelm
3.uninstall
.assert_called_once_with(
151 "myk8s_id", "helm_sample_charm_0001"
155 if __name__
== "__main__":