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
.definitions
import Offer
, RelationEndpoint
24 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
25 from osm_common
import fslocal
26 from n2vc
.exceptions
import (
27 N2VCBadArgumentsException
,
30 from n2vc
.tests
.unit
.utils
import AsyncMock
31 from n2vc
.vca
.connection_data
import ConnectionData
34 class N2VCJujuConnTestCase(asynctest
.TestCase
):
35 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.MotorStore")
36 @asynctest.mock
.patch("n2vc.n2vc_juju_conn.get_connection")
37 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
40 mock_base64_to_cacert
=None,
41 mock_get_connection
=None,
44 self
.loop
= asyncio
.get_event_loop()
46 mock_base64_to_cacert
.return_value
= """
47 -----BEGIN CERTIFICATE-----
49 -----END CERTIFICATE-----"""
50 mock_store
.return_value
= AsyncMock()
51 mock_vca_connection
= Mock()
52 mock_get_connection
.return_value
= mock_vca_connection
53 mock_vca_connection
.data
.return_value
= ConnectionData(
55 "endpoints": ["1.2.3.4:17070"],
61 "lxd-credentials": "credentials",
62 "k8s-cloud": "k8s_cloud",
63 "k8s-credentials": "k8s_credentials",
65 "api-proxy": "api_proxy",
68 logging
.disable(logging
.CRITICAL
)
70 N2VCJujuConnector
.get_public_key
= Mock()
71 self
.n2vc
= N2VCJujuConnector(
78 N2VCJujuConnector
.get_public_key
.assert_not_called()
79 self
.n2vc
.libjuju
= Mock()
82 class GetMetricssTest(N2VCJujuConnTestCase
):
84 super(GetMetricssTest
, self
).setUp()
85 self
.n2vc
.libjuju
.get_metrics
= AsyncMock()
87 def test_success(self
):
88 _
= self
.loop
.run_until_complete(self
.n2vc
.get_metrics("model", "application"))
89 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
91 def test_except(self
):
92 self
.n2vc
.libjuju
.get_metrics
.side_effect
= Exception()
93 with self
.assertRaises(Exception):
94 _
= self
.loop
.run_until_complete(
95 self
.n2vc
.get_metrics("model", "application")
97 self
.n2vc
.libjuju
.get_metrics
.assert_called_once()
100 class UpdateVcaStatusTest(N2VCJujuConnTestCase
):
102 super(UpdateVcaStatusTest
, self
).setUp()
103 self
.n2vc
.libjuju
.get_controller
= AsyncMock()
104 self
.n2vc
.libjuju
.get_model
= AsyncMock()
105 self
.n2vc
.libjuju
.get_executed_actions
= AsyncMock()
106 self
.n2vc
.libjuju
.get_actions
= AsyncMock()
107 self
.n2vc
.libjuju
.get_application_configs
= AsyncMock()
108 self
.n2vc
.libjuju
._get
_application
= AsyncMock()
113 self
.loop
.run_until_complete(
114 self
.n2vc
.update_vca_status(
115 {"model": {"applications": {"app": {"actions": {}}}}}
118 self
.n2vc
.libjuju
.get_executed_actions
.assert_called_once()
119 self
.n2vc
.libjuju
.get_actions
.assert_called_once()
120 self
.n2vc
.libjuju
.get_application_configs
.assert_called_once()
122 def test_exception(self
):
123 self
.n2vc
.libjuju
.get_model
.return_value
= None
124 self
.n2vc
.libjuju
.get_executed_actions
.side_effect
= Exception()
125 with self
.assertRaises(Exception):
126 self
.loop
.run_until_complete(
127 self
.n2vc
.update_vca_status(
128 {"model": {"applications": {"app": {"actions": {}}}}}
131 self
.n2vc
.libjuju
.get_executed_actions
.assert_not_called()
132 self
.n2vc
.libjuju
.get_actions
.assert_not_called_once()
133 self
.n2vc
.libjuju
.get_application_configs
.assert_not_called_once()
136 @asynctest.mock
.patch("osm_common.fslocal.FsLocal.file_exists")
137 @asynctest.mock
.patch(
138 "osm_common.fslocal.FsLocal.path", new_callable
=asynctest
.PropertyMock
, create
=True
140 class K8sProxyCharmsTest(N2VCJujuConnTestCase
):
142 super(K8sProxyCharmsTest
, self
).setUp()
143 self
.n2vc
.libjuju
.model_exists
= AsyncMock()
144 self
.n2vc
.libjuju
.add_model
= AsyncMock()
145 self
.n2vc
.libjuju
.deploy_charm
= AsyncMock()
146 self
.n2vc
.libjuju
.model_exists
.return_value
= False
149 "n2vc.n2vc_juju_conn.generate_random_alfanum_string",
150 **{"return_value": "random"}
154 mock_generate_random_alfanum_string
,
158 mock_file_exists
.return_value
= True
159 mock_path
.return_value
= "/path"
160 ee_id
= self
.loop
.run_until_complete(
161 self
.n2vc
.install_k8s_proxy_charm(
163 "nsi-id.ns-id.vnf-id.vdu",
169 self
.n2vc
.libjuju
.add_model
.assert_called_once()
170 self
.n2vc
.libjuju
.deploy_charm
.assert_called_once_with(
171 model_name
="ns-id-k8s",
172 application_name
="app-vnf-vnf-id-vdu-vdu-random",
176 progress_timeout
=None,
180 self
.assertEqual(ee_id
, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s")
182 def test_no_artifact_path(
187 with self
.assertRaises(N2VCBadArgumentsException
):
188 ee_id
= self
.loop
.run_until_complete(
189 self
.n2vc
.install_k8s_proxy_charm(
191 "nsi-id.ns-id.vnf-id.vdu",
196 self
.assertIsNone(ee_id
)
203 with self
.assertRaises(N2VCBadArgumentsException
):
204 ee_id
= self
.loop
.run_until_complete(
205 self
.n2vc
.install_k8s_proxy_charm(
207 "nsi-id.ns-id.vnf-id.vdu",
212 self
.assertIsNone(ee_id
)
214 def test_file_not_exists(
219 mock_file_exists
.return_value
= False
220 with self
.assertRaises(N2VCBadArgumentsException
):
221 ee_id
= self
.loop
.run_until_complete(
222 self
.n2vc
.install_k8s_proxy_charm(
224 "nsi-id.ns-id.vnf-id.vdu",
229 self
.assertIsNone(ee_id
)
236 mock_file_exists
.return_value
= True
237 mock_path
.return_value
= "/path"
238 self
.n2vc
.libjuju
.deploy_charm
.side_effect
= Exception()
239 with self
.assertRaises(N2VCException
):
240 ee_id
= self
.loop
.run_until_complete(
241 self
.n2vc
.install_k8s_proxy_charm(
243 "nsi-id.ns-id.vnf-id.vdu",
248 self
.assertIsNone(ee_id
)
251 class AddRelationTest(N2VCJujuConnTestCase
):
253 super(AddRelationTest
, self
).setUp()
254 self
.n2vc
.libjuju
.add_relation
= AsyncMock()
255 self
.n2vc
.libjuju
.offer
= AsyncMock()
256 self
.n2vc
.libjuju
.get_controller
= AsyncMock()
257 self
.n2vc
.libjuju
.consume
= AsyncMock()
259 def test_standard_relation(self
):
260 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
261 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint")
262 self
.loop
.run_until_complete(
263 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
265 self
.n2vc
.libjuju
.add_relation
.assert_called_once_with(
266 model_name
="model-1", endpoint_1
="app1:endpoint", endpoint_2
="app2:endpoint"
268 self
.n2vc
.libjuju
.offer
.assert_not_called()
269 self
.n2vc
.libjuju
.consume
.assert_not_called()
271 def test_cmr_relation_same_controller(self
):
272 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
273 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
274 offer
= Offer("admin/model-1.app1")
275 self
.n2vc
.libjuju
.offer
.return_value
= offer
276 self
.n2vc
.libjuju
.consume
.return_value
= "saas"
277 self
.loop
.run_until_complete(
278 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
280 self
.n2vc
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
281 self
.n2vc
.libjuju
.consume
.assert_called_once()
282 self
.n2vc
.libjuju
.add_relation
.assert_called_once_with(
283 "model-2", "app2:endpoint", "saas"
286 def test_relation_exception(self
):
287 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
288 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
289 self
.n2vc
.libjuju
.offer
.side_effect
= Exception()
290 with self
.assertRaises(N2VCException
):
291 self
.loop
.run_until_complete(
292 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)