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
.definitions
import Offer
, RelationEndpoint
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
152 mock_file_exists
.return_value
= True
153 mock_path
.return_value
= "/path"
154 ee_id
= self
.loop
.run_until_complete(
155 self
.n2vc
.install_k8s_proxy_charm(
157 "nsi-id.ns-id.vnf-id.vdu",
163 self
.n2vc
.libjuju
.add_model
.assert_called_once()
164 self
.n2vc
.libjuju
.deploy_charm
.assert_called_once_with(
165 model_name
="ns-id-k8s",
166 application_name
="app-vnf-vnf-id-vdu-vdu",
170 progress_timeout
=None,
174 self
.assertEqual(ee_id
, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
176 def test_no_artifact_path(
181 with self
.assertRaises(N2VCBadArgumentsException
):
182 ee_id
= self
.loop
.run_until_complete(
183 self
.n2vc
.install_k8s_proxy_charm(
185 "nsi-id.ns-id.vnf-id.vdu",
190 self
.assertIsNone(ee_id
)
197 with self
.assertRaises(N2VCBadArgumentsException
):
198 ee_id
= self
.loop
.run_until_complete(
199 self
.n2vc
.install_k8s_proxy_charm(
201 "nsi-id.ns-id.vnf-id.vdu",
206 self
.assertIsNone(ee_id
)
208 def test_file_not_exists(
213 mock_file_exists
.return_value
= False
214 with self
.assertRaises(N2VCBadArgumentsException
):
215 ee_id
= self
.loop
.run_until_complete(
216 self
.n2vc
.install_k8s_proxy_charm(
218 "nsi-id.ns-id.vnf-id.vdu",
223 self
.assertIsNone(ee_id
)
230 mock_file_exists
.return_value
= True
231 mock_path
.return_value
= "/path"
232 self
.n2vc
.libjuju
.deploy_charm
.side_effect
= Exception()
233 with self
.assertRaises(N2VCException
):
234 ee_id
= self
.loop
.run_until_complete(
235 self
.n2vc
.install_k8s_proxy_charm(
237 "nsi-id.ns-id.vnf-id.vdu",
242 self
.assertIsNone(ee_id
)
245 class AddRelationTest(N2VCJujuConnTestCase
):
247 super(AddRelationTest
, self
).setUp()
248 self
.n2vc
.libjuju
.add_relation
= AsyncMock()
249 self
.n2vc
.libjuju
.offer
= AsyncMock()
250 self
.n2vc
.libjuju
.get_controller
= AsyncMock()
251 self
.n2vc
.libjuju
.consume
= AsyncMock()
253 def test_standard_relation(self
):
254 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
255 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint")
256 self
.loop
.run_until_complete(
257 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
259 self
.n2vc
.libjuju
.add_relation
.assert_called_once_with(
260 model_name
="model-1", endpoint_1
="app1:endpoint", endpoint_2
="app2:endpoint"
262 self
.n2vc
.libjuju
.offer
.assert_not_called()
263 self
.n2vc
.libjuju
.consume
.assert_not_called()
265 def test_cmr_relation_same_controller(self
):
266 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
267 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
268 offer
= Offer("admin/model-1.app1")
269 self
.n2vc
.libjuju
.offer
.return_value
= offer
270 self
.n2vc
.libjuju
.consume
.return_value
= "saas"
271 self
.loop
.run_until_complete(
272 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
274 self
.n2vc
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
275 self
.n2vc
.libjuju
.consume
.assert_called_once()
276 self
.n2vc
.libjuju
.add_relation
.assert_called_once_with(
277 "model-2", "app2:endpoint", "saas"
280 def test_relation_exception(self
):
281 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
282 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
283 self
.n2vc
.libjuju
.offer
.side_effect
= Exception()
284 with self
.assertRaises(N2VCException
):
285 self
.loop
.run_until_complete(
286 self
.n2vc
.add_relation(relation_endpoint_1
, relation_endpoint_2
)