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 osm_common
.fslocal
import FsLocal
24 from asynctest
.mock
import Mock
25 from osm_common
.dbmemory
import DbMemory
27 __author__
= "Isabel Lloret <illoret@indra.es>"
30 class TestLcmHelmConn(asynctest
.TestCase
):
31 logging
.basicConfig(level
=logging
.DEBUG
)
32 logger
= logging
.getLogger(__name__
)
33 logger
.setLevel(logging
.DEBUG
)
35 async def setUp(self
):
36 self
.db
= Mock(DbMemory())
37 self
.fs
= asynctest
.Mock(FsLocal())
38 self
.fs
.path
= "/app/storage"
40 lcm_helm_conn
.K8sHelmConnector
= asynctest
.Mock(lcm_helm_conn
.K8sHelmConnector
)
41 self
.helm_conn
= LCMHelmConn(self
.db
, self
.fs
, loop
=self
.loop
, vca_config
=vca_config
, log
=self
.logger
)
43 @asynctest.fail_on(active_handles
=True)
44 async def test_create_execution_environment(self
):
45 namespace
= "testnamespace"
47 artifact_path
= "helm_sample_charm"
48 helm_chart_id
= "helm_sample_charm_0001"
49 self
.helm_conn
._k
8sclusterhelm
.install
= asynctest
.CoroutineMock(return_value
=helm_chart_id
)
50 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
51 ee_id
, _
= await self
.helm_conn
.create_execution_environment(namespace
, db_dict
, artifact_path
=artifact_path
)
52 self
.assertEqual(ee_id
, "{}.{}".format("osm", helm_chart_id
),
53 "Check ee_id format: <default namespace>.<helm_chart-id>")
54 self
.helm_conn
._k
8sclusterhelm
.install
.assert_called_once_with("myk8s_id",
55 kdu_model
="/app/storage/helm_sample_charm",
56 namespace
="osm", db_dict
=db_dict
,
57 params
=None, timeout
=None)
59 @asynctest.fail_on(active_handles
=True)
60 async def test_get_ee_ssh_public__key(self
):
61 ee_id
= "osm.helm_sample_charm_0001"
63 lcm_helm_conn
.socket
.gethostbyname
= asynctest
.Mock()
64 mock_pub_key
= "ssh-rsapubkey"
65 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
66 self
.helm_conn
._get
_ssh
_key
= asynctest
.CoroutineMock(return_value
=mock_pub_key
)
67 pub_key
= await self
.helm_conn
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
68 self
.assertEqual(pub_key
, mock_pub_key
)
70 @asynctest.fail_on(active_handles
=True)
71 async def test_execute_primitive(self
):
72 lcm_helm_conn
.socket
.gethostbyname
= asynctest
.Mock()
73 ee_id
= "osm.helm_sample_charm_0001"
74 primitive_name
= "sleep"
76 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
77 self
.helm_conn
._execute
_primitive
_internal
= asynctest
.CoroutineMock(return_value
=("OK", "test-ok"))
78 message
= await self
.helm_conn
.exec_primitive(ee_id
, primitive_name
, params
)
79 self
.assertEqual(message
, "test-ok")
81 @asynctest.fail_on(active_handles
=True)
82 async def test_execute_config_primitive(self
):
83 self
.logger
.debug("Execute config primitive")
84 lcm_helm_conn
.socket
.gethostbyname
= asynctest
.Mock()
85 ee_id
= "osm.helm_sample_charm_0001"
86 primitive_name
= "config"
87 params
= {"ssh-host-name": "host1"}
88 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
89 self
.helm_conn
._execute
_primitive
_internal
= asynctest
.CoroutineMock(return_value
=("OK", "CONFIG OK"))
90 message
= await self
.helm_conn
.exec_primitive(ee_id
, primitive_name
, params
)
91 self
.assertEqual(message
, "CONFIG OK")
93 @asynctest.fail_on(active_handles
=True)
94 async def test_delete_execution_environment(self
):
95 ee_id
= "osm.helm_sample_charm_0001"
96 self
.db
.get_one
.return_value
= {"_admin": {"helm-chart": {"id": "myk8s_id"}}}
97 self
.helm_conn
._k
8sclusterhelm
.uninstall
= asynctest
.CoroutineMock()
98 await self
.helm_conn
.delete_execution_environment(ee_id
)
99 self
.helm_conn
._k
8sclusterhelm
.uninstall
.assert_called_once_with("myk8s_id", "helm_sample_charm_0001")
102 if __name__
== '__main__':