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
19 from unittest
.mock
import patch
23 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
24 from osm_common
import fslocal
25 from n2vc
.exceptions
import (
26 N2VCBadArgumentsException
,
29 from n2vc
.tests
.unit
.utils
import AsyncMock
30 from n2vc
.vca
.connection_data
import ConnectionData
33 class N2VCJujuConnTestCase(asynctest
.TestCase
):
34 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.MotorStore")
35 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.get_connection")
36 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
39 mock_base64_to_cacert
=None,
40 mock_get_connection
=None,
43 self
.loop
= asyncio
.get_event_loop()
45 mock_base64_to_cacert
.return_value
= """
46 -----BEGIN CERTIFICATE-----
48 -----END CERTIFICATE-----"""
49 mock_store
.return_value
= AsyncMock()
50 mock_vca_connection
= Mock()
51 mock_get_connection
.return_value
= mock_vca_connection
52 mock_vca_connection
.data
.return_value
= ConnectionData(
54 "endpoints": ["1.2.3.4:17070"],
60 "lxd-credentials": "credentials",
61 "k8s-cloud": "k8s_cloud",
62 "k8s-credentials": "k8s_credentials",
64 "api-proxy": "api_proxy",
67 logging
.disable(logging
.CRITICAL
)
69 N2VCJujuConnector
.get_public_key
= Mock()
70 self
.n2vc
= N2VCJujuConnector(
77 N2VCJujuConnector
.get_public_key
.assert_not_called()
78 self
.n2vc
.libjuju
= Mock()
81 class GetMetricssTest(N2VCJujuConnTestCase
):
83 super(GetMetricssTest
, self
).setUp()
84 self
.n2vc
.libjuju
.get_metrics
= AsyncMock()
86 def test_success(self
):
87 _
= self
.loop
.run_until_complete(self
.n2vc
.get_metrics("model", "application"))
88 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
90 def test_except(self
):
91 self
.n2vc
.libjuju
.get_metrics
.side_effect
= Exception()
92 with self
.assertRaises(Exception):
93 _
= self
.loop
.run_until_complete(
94 self
.n2vc
.get_metrics("model", "application")
96 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
99 class UpdateVcaStatusTest(N2VCJujuConnTestCase
):
101 super(UpdateVcaStatusTest
, self
).setUp()
102 self
.n2vc
.libjuju
.get_controller
= AsyncMock()
103 self
.n2vc
.libjuju
.get_model
= AsyncMock()
104 self
.n2vc
.libjuju
.get_executed_actions
= AsyncMock()
105 self
.n2vc
.libjuju
.get_actions
= AsyncMock()
106 self
.n2vc
.libjuju
.get_application_configs
= AsyncMock()
107 self
.n2vc
.libjuju
._get
_application
= AsyncMock()
112 self
.loop
.run_until_complete(
113 self
.n2vc
.update_vca_status(
114 {"model": {"applications": {"app": {"actions": {}}}}}
117 self
.n2vc
.libjuju
.get_executed_actions
.assert_called_once()
118 self
.n2vc
.libjuju
.get_actions
.assert_called_once()
119 self
.n2vc
.libjuju
.get_application_configs
.assert_called_once()
121 def test_exception(self
):
122 self
.n2vc
.libjuju
.get_model
.return_value
= None
123 self
.n2vc
.libjuju
.get_executed_actions
.side_effect
= Exception()
124 with self
.assertRaises(Exception):
125 self
.loop
.run_until_complete(
126 self
.n2vc
.update_vca_status(
127 {"model": {"applications": {"app": {"actions": {}}}}}
130 self
.n2vc
.libjuju
.get_executed_actions
.assert_not_called()
131 self
.n2vc
.libjuju
.get_actions
.assert_not_called_once()
132 self
.n2vc
.libjuju
.get_application_configs
.assert_not_called_once()
135 @asynctest.mock
.patch("osm_common.fslocal.FsLocal.file_exists")
136 @asynctest.mock
.patch(
137 "osm_common.fslocal.FsLocal.path", new_callable
=asynctest
.PropertyMock
, create
=True
139 class K8sProxyCharmsTest(N2VCJujuConnTestCase
):
141 super(K8sProxyCharmsTest
, self
).setUp()
142 self
.n2vc
.libjuju
.model_exists
= AsyncMock()
143 self
.n2vc
.libjuju
.add_model
= AsyncMock()
144 self
.n2vc
.libjuju
.deploy_charm
= AsyncMock()
145 self
.n2vc
.libjuju
.model_exists
.return_value
= False
148 "n2vc.n2vc_juju_conn.generate_random_alfanum_string",
149 **{"return_value": "random"}
153 mock_generate_random_alfanum_string
,
157 mock_file_exists
.return_value
= True
158 mock_path
.return_value
= "/path"
159 ee_id
= self
.loop
.run_until_complete(
160 self
.n2vc
.install_k8s_proxy_charm(
162 "nsi-id.ns-id.vnf-id.vdu",
168 self
.n2vc
.libjuju
.add_model
.assert_called_once()
169 self
.n2vc
.libjuju
.deploy_charm
.assert_called_once_with(
170 model_name
="ns-id-k8s",
171 application_name
="app-vnf-vnf-id-vdu-vdu-random",
175 progress_timeout
=None,
179 self
.assertEqual(ee_id
, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s")
181 def test_no_artifact_path(
186 with self
.assertRaises(N2VCBadArgumentsException
):
187 ee_id
= self
.loop
.run_until_complete(
188 self
.n2vc
.install_k8s_proxy_charm(
190 "nsi-id.ns-id.vnf-id.vdu",
195 self
.assertIsNone(ee_id
)
202 with self
.assertRaises(N2VCBadArgumentsException
):
203 ee_id
= self
.loop
.run_until_complete(
204 self
.n2vc
.install_k8s_proxy_charm(
206 "nsi-id.ns-id.vnf-id.vdu",
211 self
.assertIsNone(ee_id
)
213 def test_file_not_exists(
218 mock_file_exists
.return_value
= False
219 with self
.assertRaises(N2VCBadArgumentsException
):
220 ee_id
= self
.loop
.run_until_complete(
221 self
.n2vc
.install_k8s_proxy_charm(
223 "nsi-id.ns-id.vnf-id.vdu",
228 self
.assertIsNone(ee_id
)
235 mock_file_exists
.return_value
= True
236 mock_path
.return_value
= "/path"
237 self
.n2vc
.libjuju
.deploy_charm
.side_effect
= Exception()
238 with self
.assertRaises(N2VCException
):
239 ee_id
= self
.loop
.run_until_complete(
240 self
.n2vc
.install_k8s_proxy_charm(
242 "nsi-id.ns-id.vnf-id.vdu",
247 self
.assertIsNone(ee_id
)