1 # Copyright 2020 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
18 from unittest
.mock
import Mock
22 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
23 from osm_common
import fslocal
24 from n2vc
.exceptions
import (
25 N2VCBadArgumentsException
,
28 from n2vc
.tests
.unit
.utils
import AsyncMock
29 from n2vc
.vca
.connection_data
import ConnectionData
32 class N2VCJujuConnTestCase(asynctest
.TestCase
):
33 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.MotorStore")
34 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.get_connection")
35 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
38 mock_base64_to_cacert
=None,
39 mock_get_connection
=None,
42 self
.loop
= asyncio
.get_event_loop()
44 mock_base64_to_cacert
.return_value
= """
45 -----BEGIN CERTIFICATE-----
47 -----END CERTIFICATE-----"""
48 mock_store
.return_value
= AsyncMock()
49 mock_vca_connection
= Mock()
50 mock_get_connection
.return_value
= mock_vca_connection
51 mock_vca_connection
.data
.return_value
= ConnectionData(
53 "endpoints": ["1.2.3.4:17070"],
59 "lxd-credentials": "credentials",
60 "k8s-cloud": "k8s_cloud",
61 "k8s-credentials": "k8s_credentials",
63 "api-proxy": "api_proxy",
66 logging
.disable(logging
.CRITICAL
)
68 N2VCJujuConnector
.get_public_key
= Mock()
69 self
.n2vc
= N2VCJujuConnector(
76 N2VCJujuConnector
.get_public_key
.assert_not_called()
77 self
.n2vc
.libjuju
= Mock()
80 class GetMetricssTest(N2VCJujuConnTestCase
):
82 super(GetMetricssTest
, self
).setUp()
83 self
.n2vc
.libjuju
.get_metrics
= AsyncMock()
85 def test_success(self
):
86 _
= self
.loop
.run_until_complete(self
.n2vc
.get_metrics("model", "application"))
87 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
89 def test_except(self
):
90 self
.n2vc
.libjuju
.get_metrics
.side_effect
= Exception()
91 with self
.assertRaises(Exception):
92 _
= self
.loop
.run_until_complete(
93 self
.n2vc
.get_metrics("model", "application")
95 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
98 class UpdateVcaStatusTest(N2VCJujuConnTestCase
):
100 super(UpdateVcaStatusTest
, self
).setUp()
101 self
.n2vc
.libjuju
.get_controller
= AsyncMock()
102 self
.n2vc
.libjuju
.get_model
= AsyncMock()
103 self
.n2vc
.libjuju
.get_executed_actions
= AsyncMock()
104 self
.n2vc
.libjuju
.get_actions
= AsyncMock()
105 self
.n2vc
.libjuju
.get_application_configs
= AsyncMock()
106 self
.n2vc
.libjuju
._get
_application
= AsyncMock()
111 self
.loop
.run_until_complete(
112 self
.n2vc
.update_vca_status(
113 {"model": {"applications": {"app": {"actions": {}}}}}
116 self
.n2vc
.libjuju
.get_executed_actions
.assert_called_once()
117 self
.n2vc
.libjuju
.get_actions
.assert_called_once()
118 self
.n2vc
.libjuju
.get_application_configs
.assert_called_once()
120 def test_exception(self
):
121 self
.n2vc
.libjuju
.get_model
.return_value
= None
122 self
.n2vc
.libjuju
.get_executed_actions
.side_effect
= Exception()
123 with self
.assertRaises(Exception):
124 self
.loop
.run_until_complete(
125 self
.n2vc
.update_vca_status(
126 {"model": {"applications": {"app": {"actions": {}}}}}
129 self
.n2vc
.libjuju
.get_executed_actions
.assert_not_called()
130 self
.n2vc
.libjuju
.get_actions
.assert_not_called_once()
131 self
.n2vc
.libjuju
.get_application_configs
.assert_not_called_once()
134 @asynctest.mock
.patch("osm_common.fslocal.FsLocal.file_exists")
135 @asynctest.mock
.patch(
136 "osm_common.fslocal.FsLocal.path", new_callable
=asynctest
.PropertyMock
, create
=True
138 class K8sProxyCharmsTest(N2VCJujuConnTestCase
):
140 super(K8sProxyCharmsTest
, self
).setUp()
141 self
.n2vc
.libjuju
.model_exists
= AsyncMock()
142 self
.n2vc
.libjuju
.add_model
= AsyncMock()
143 self
.n2vc
.libjuju
.deploy_charm
= AsyncMock()
144 self
.n2vc
.libjuju
.model_exists
.return_value
= False
151 mock_file_exists
.return_value
= True
152 mock_path
.return_value
= "/path"
153 ee_id
= self
.loop
.run_until_complete(
154 self
.n2vc
.install_k8s_proxy_charm(
156 "nsi-id.ns-id.vnf-id.vdu",
162 self
.n2vc
.libjuju
.add_model
.assert_called_once()
163 self
.n2vc
.libjuju
.deploy_charm
.assert_called_once_with(
164 model_name
="ns-id-k8s",
165 application_name
="app-vnf-vnf-id-vdu-vdu",
169 progress_timeout
=None,
173 self
.assertEqual(ee_id
, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
175 def test_no_artifact_path(
180 with self
.assertRaises(N2VCBadArgumentsException
):
181 ee_id
= self
.loop
.run_until_complete(
182 self
.n2vc
.install_k8s_proxy_charm(
184 "nsi-id.ns-id.vnf-id.vdu",
189 self
.assertIsNone(ee_id
)
196 with self
.assertRaises(N2VCBadArgumentsException
):
197 ee_id
= self
.loop
.run_until_complete(
198 self
.n2vc
.install_k8s_proxy_charm(
200 "nsi-id.ns-id.vnf-id.vdu",
205 self
.assertIsNone(ee_id
)
207 def test_file_not_exists(
212 mock_file_exists
.return_value
= False
213 with self
.assertRaises(N2VCBadArgumentsException
):
214 ee_id
= self
.loop
.run_until_complete(
215 self
.n2vc
.install_k8s_proxy_charm(
217 "nsi-id.ns-id.vnf-id.vdu",
222 self
.assertIsNone(ee_id
)
229 mock_file_exists
.return_value
= True
230 mock_path
.return_value
= "/path"
231 self
.n2vc
.libjuju
.deploy_charm
.side_effect
= Exception()
232 with self
.assertRaises(N2VCException
):
233 ee_id
= self
.loop
.run_until_complete(
234 self
.n2vc
.install_k8s_proxy_charm(
236 "nsi-id.ns-id.vnf-id.vdu",
241 self
.assertIsNone(ee_id
)