Feature 10974: Add juju instantiation params.
[osm/N2VC.git] / n2vc / tests / unit / test_n2vc_juju_conn.py
1 # Copyright 2020 Canonical Ltd.
2 #
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
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
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.
14
15
16 import asyncio
17 import logging
18 from unittest.mock import Mock
19
20
21 import asynctest
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,
27 N2VCException,
28 )
29 from n2vc.tests.unit.utils import AsyncMock
30 from n2vc.vca.connection_data import ConnectionData
31
32
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")
37 def setUp(
38 self,
39 mock_base64_to_cacert=None,
40 mock_get_connection=None,
41 mock_store=None,
42 ):
43 self.loop = asyncio.get_event_loop()
44 self.db = Mock()
45 mock_base64_to_cacert.return_value = """
46 -----BEGIN CERTIFICATE-----
47 SOMECERT
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(
53 **{
54 "endpoints": ["1.2.3.4:17070"],
55 "user": "user",
56 "secret": "secret",
57 "cacert": "cacert",
58 "pubkey": "pubkey",
59 "lxd-cloud": "cloud",
60 "lxd-credentials": "credentials",
61 "k8s-cloud": "k8s_cloud",
62 "k8s-credentials": "k8s_credentials",
63 "model-config": {},
64 "api-proxy": "api_proxy",
65 }
66 )
67 logging.disable(logging.CRITICAL)
68
69 N2VCJujuConnector.get_public_key = Mock()
70 self.n2vc = N2VCJujuConnector(
71 db=self.db,
72 fs=fslocal.FsLocal(),
73 log=None,
74 loop=self.loop,
75 on_update_db=None,
76 )
77 N2VCJujuConnector.get_public_key.assert_not_called()
78 self.n2vc.libjuju = Mock()
79
80
81 class GetMetricssTest(N2VCJujuConnTestCase):
82 def setUp(self):
83 super(GetMetricssTest, self).setUp()
84 self.n2vc.libjuju.get_metrics = AsyncMock()
85
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()
89
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")
95 )
96 self.n2vc.libjuju.get_metrics.assert_called_once()
97
98
99 class UpdateVcaStatusTest(N2VCJujuConnTestCase):
100 def setUp(self):
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()
108
109 def test_success(
110 self,
111 ):
112 self.loop.run_until_complete(
113 self.n2vc.update_vca_status(
114 {"model": {"applications": {"app": {"actions": {}}}}}
115 )
116 )
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()
120
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": {}}}}}
128 )
129 )
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()
133
134
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
138 )
139 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
140 def setUp(self):
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
146
147 def test_success(
148 self,
149 mock_path,
150 mock_file_exists,
151 ):
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(
156 "charm",
157 "nsi-id.ns-id.vnf-id.vdu",
158 "////path/",
159 {},
160 )
161 )
162
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",
167 path="/path/path/",
168 machine_id=None,
169 db_dict={},
170 progress_timeout=None,
171 total_timeout=None,
172 config=None,
173 )
174 self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
175
176 def test_no_artifact_path(
177 self,
178 mock_path,
179 mock_file_exists,
180 ):
181 with self.assertRaises(N2VCBadArgumentsException):
182 ee_id = self.loop.run_until_complete(
183 self.n2vc.install_k8s_proxy_charm(
184 "charm",
185 "nsi-id.ns-id.vnf-id.vdu",
186 "",
187 {},
188 )
189 )
190 self.assertIsNone(ee_id)
191
192 def test_no_db(
193 self,
194 mock_path,
195 mock_file_exists,
196 ):
197 with self.assertRaises(N2VCBadArgumentsException):
198 ee_id = self.loop.run_until_complete(
199 self.n2vc.install_k8s_proxy_charm(
200 "charm",
201 "nsi-id.ns-id.vnf-id.vdu",
202 "/path/",
203 None,
204 )
205 )
206 self.assertIsNone(ee_id)
207
208 def test_file_not_exists(
209 self,
210 mock_path,
211 mock_file_exists,
212 ):
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(
217 "charm",
218 "nsi-id.ns-id.vnf-id.vdu",
219 "/path/",
220 {},
221 )
222 )
223 self.assertIsNone(ee_id)
224
225 def test_exception(
226 self,
227 mock_path,
228 mock_file_exists,
229 ):
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(
236 "charm",
237 "nsi-id.ns-id.vnf-id.vdu",
238 "path/",
239 {},
240 )
241 )
242 self.assertIsNone(ee_id)
243
244
245 class AddRelationTest(N2VCJujuConnTestCase):
246 def setUp(self):
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()
252
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)
258 )
259 self.n2vc.libjuju.add_relation.assert_called_once_with(
260 model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
261 )
262 self.n2vc.libjuju.offer.assert_not_called()
263 self.n2vc.libjuju.consume.assert_not_called()
264
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)
273 )
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"
278 )
279
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)
287 )