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.
16 from base64
import b64decode
17 from unittest
import TestCase
18 from unittest
.mock
import Mock
, patch
21 from n2vc
.store
import DbMongoStore
, MotorStore
22 from n2vc
.vca
.connection_data
import ConnectionData
23 from n2vc
.tests
.unit
.utils
import AsyncMock
24 from osm_common
.dbmongo
import DbException
27 class TestDbMongoStore(TestCase
):
29 self
.store
= DbMongoStore(Mock())
30 self
.loop
= asyncio
.get_event_loop()
32 @patch("n2vc.vca.connection_data.base64_to_cacert")
33 def test_get_vca_connection_data(self
, mock_base64_to_cacert
):
34 mock_base64_to_cacert
.return_value
= "cacert"
36 "endpoints": ["1.2.3.4:17070"],
41 "lxd-cloud": "lxd-cloud",
42 "lxd-credentials": "lxd-credentials",
43 "k8s-cloud": "k8s-cloud",
44 "k8s-credentials": "k8s-credentials",
48 db_get_one
= conn_data
.copy()
49 db_get_one
.update({"schema_version": "1.1", "_id": "id"})
50 self
.store
.db
.get_one
.return_value
= db_get_one
51 connection_data
= self
.loop
.run_until_complete(
52 self
.store
.get_vca_connection_data("vca_id")
56 connection_data
.__dict
__[k
.replace("-", "_")] == v
57 for k
, v
in conn_data
.items()
61 def test_update_vca_endpoints(self
):
62 endpoints
= ["1.2.3.4:17070"]
63 self
.store
.db
.get_one
.side_effect
= [None, {"api_endpoints": []}]
64 self
.store
.db
.create
.side_effect
= DbException("already exists")
65 self
.loop
.run_until_complete(self
.store
.update_vca_endpoints(endpoints
))
66 self
.assertEqual(self
.store
.db
.get_one
.call_count
, 2)
68 self
.store
.db
.set_one
.assert_called_once_with(
69 "vca", {"_id": "juju"}, {"api_endpoints": endpoints
}
72 def test_update_vca_endpoints_exception(self
):
73 endpoints
= ["1.2.3.4:17070"]
74 self
.store
.db
.get_one
.side_effect
= [None, None]
75 self
.store
.db
.create
.side_effect
= DbException("already exists")
76 with self
.assertRaises(DbException
):
77 self
.loop
.run_until_complete(self
.store
.update_vca_endpoints(endpoints
))
78 self
.assertEqual(self
.store
.db
.get_one
.call_count
, 2)
79 self
.store
.db
.set_one
.assert_not_called()
81 def test_update_vca_endpoints_with_vca_id(self
):
82 endpoints
= ["1.2.3.4:17070"]
83 self
.store
.db
.get_one
.return_value
= {}
84 self
.loop
.run_until_complete(
85 self
.store
.update_vca_endpoints(endpoints
, "vca_id")
87 self
.store
.db
.get_one
.assert_called_once_with("vca", q_filter
={"_id": "vca_id"})
88 self
.store
.db
.replace
.assert_called_once_with(
89 "vca", "vca_id", {"endpoints": endpoints
}
92 def test_get_vca_endpoints(self
):
93 endpoints
= ["1.2.3.4:17070"]
94 db_data
= {"api_endpoints": endpoints
}
95 db_returns
= [db_data
, None]
96 expected_returns
= [endpoints
, []]
98 self
.store
._get
_juju
_info
= Mock()
99 self
.store
._get
_juju
_info
.side_effect
= db_returns
100 for _
in range(len(db_returns
)):
101 e
= self
.loop
.run_until_complete(self
.store
.get_vca_endpoints())
103 self
.assertEqual(expected_returns
, returns
)
105 @patch("n2vc.vca.connection_data.base64_to_cacert")
106 def test_get_vca_endpoints_with_vca_id(self
, mock_base64_to_cacert
):
107 expected_endpoints
= ["1.2.3.4:17070"]
108 mock_base64_to_cacert
.return_value
= "cacert"
109 self
.store
.get_vca_connection_data
= Mock()
110 self
.store
.get_vca_connection_data
.return_value
= ConnectionData(
112 "endpoints": expected_endpoints
,
118 endpoints
= self
.loop
.run_until_complete(self
.store
.get_vca_endpoints("vca_id"))
119 self
.store
.get_vca_connection_data
.assert_called_with("vca_id")
120 self
.assertEqual(expected_endpoints
, endpoints
)
122 def test_get_vca_id(self
):
123 self
.assertIsNone(self
.loop
.run_until_complete(self
.store
.get_vca_id()))
125 def test_get_vca_id_with_vim_id(self
):
126 self
.store
.db
.get_one
.return_value
= {"vca": "vca_id"}
127 vca_id
= self
.loop
.run_until_complete(self
.store
.get_vca_id("vim_id"))
128 self
.store
.db
.get_one
.assert_called_once_with(
129 "vim_accounts", q_filter
={"_id": "vim_id"}, fail_on_empty
=False
131 self
.assertEqual(vca_id
, "vca_id")
134 class TestMotorStore(TestCase
):
136 self
.store
= MotorStore("uri")
137 self
.vca_collection
= Mock()
138 self
.vca_collection
.find_one
= AsyncMock()
139 self
.vca_collection
.insert_one
= AsyncMock()
140 self
.vca_collection
.replace_one
= AsyncMock()
141 self
.admin_collection
= Mock()
142 self
.admin_collection
.find_one
= AsyncMock()
143 self
.admin_collection
.insert_one
= AsyncMock()
144 self
.admin_collection
.replace_one
= AsyncMock()
145 self
.vim_accounts_collection
= Mock()
146 self
.vim_accounts_collection
.find_one
= AsyncMock()
147 self
.store
._client
= {
149 "vca": self
.vca_collection
,
150 "admin": self
.admin_collection
,
151 "vim_accounts": self
.vim_accounts_collection
,
154 self
.store
._config
= {"database_commonkey": "osm"}
155 # self.store.decrypt_fields = Mock()
156 self
.loop
= asyncio
.get_event_loop()
158 @patch("n2vc.vca.connection_data.base64_to_cacert")
159 def test_get_vca_connection_data(self
, mock_base64_to_cacert
):
160 mock_base64_to_cacert
.return_value
= "cacert"
162 "endpoints": ["1.2.3.4:17070"],
167 "lxd-cloud": "lxd-cloud",
168 "lxd-credentials": "lxd-credentials",
169 "k8s-cloud": "k8s-cloud",
170 "k8s-credentials": "k8s-credentials",
174 db_find_one
= conn_data
.copy()
175 db_find_one
.update({"schema_version": "1.1", "_id": "id"})
176 self
.vca_collection
.find_one
.return_value
= db_find_one
177 self
.store
.decrypt_fields
= AsyncMock()
178 connection_data
= self
.loop
.run_until_complete(
179 self
.store
.get_vca_connection_data("vca_id")
183 connection_data
.__dict
__[k
.replace("-", "_")] == v
184 for k
, v
in conn_data
.items()
188 @patch("n2vc.vca.connection_data.base64_to_cacert")
189 def test_get_vca_connection_data_exception(self
, mock_base64_to_cacert
):
190 mock_base64_to_cacert
.return_value
= "cacert"
191 self
.vca_collection
.find_one
.return_value
= None
192 with self
.assertRaises(Exception):
193 self
.loop
.run_until_complete(self
.store
.get_vca_connection_data("vca_id"))
195 def test_update_vca_endpoints(self
):
196 endpoints
= ["1.2.3.4:17070"]
197 self
.admin_collection
.find_one
.side_effect
= [None, {"api_endpoints": []}]
198 self
.admin_collection
.insert_one
.side_effect
= DbException("already exists")
199 self
.loop
.run_until_complete(self
.store
.update_vca_endpoints(endpoints
))
200 self
.assertEqual(self
.admin_collection
.find_one
.call_count
, 2)
201 self
.admin_collection
.replace_one
.assert_called_once_with(
202 {"_id": "juju"}, {"api_endpoints": ["1.2.3.4:17070"]}
205 def test_get_vca_connection_data_with_id(self
):
206 secret
= "e7b253af37785045d1ca08b8d929e556"
207 encrypted_secret
= "kI46kRJh828ExSNpr16OG/q5a5/qTsE0bsHrv/W/2/g="
208 cacert
= "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUQ4ekNDQWx1Z0F3SUJBZ0lVRWlzTTBoQWxiYzQ0Z1ZhZWh6bS80ZUsyNnRZd0RRWUpLb1pJaHZjTkFRRUwKQlFBd0lURU5NQXNHQTFVRUNoTUVTblZxZFRFUU1BNEdBMVVFQXhNSGFuVnFkUzFqWVRBZUZ3MHlNVEEwTWpNeApNRFV3TXpSYUZ3MHpNVEEwTWpNeE1EVTFNelJhTUNFeERUQUxCZ05WQkFvVEJFcDFhblV4RURBT0JnTlZCQU1UCkIycDFhblV0WTJFd2dnR2lNQTBHQ1NxR1NJYjNEUUVCQVFVQUE0SUJqd0F3Z2dHS0FvSUJnUUNhTmFvNGZab2gKTDJWYThtdy9LdCs3RG9tMHBYTlIvbEUxSHJyVmZvbmZqZFVQV01zSHpTSjJZZXlXcUNSd3BiaHlLaE82N1c1dgpUY2RsV3Y3WGFLTGtsdVkraDBZY3BQT3BFTmZZYmxrNGk0QkV1L0wzYVY5MFFkUFFrMG94S01CS2R5QlBNZVNNCkJmS2pPWXdyOGgzM0ZWUWhmVkJnMXVGZ2tGaDdTamNuNHczUFdvc1BCMjNiVHBCbGR3VE9zemN4Qm9TaDNSVTkKTzZjb3lQdDdEN0drOCtHRlA3RGRUQTdoV1RkaUM4cDBkeHp2RUNmY0psMXNFeFEyZVprS1QvVzZyelNtVDhUTApCM0ErM1FDRDhEOEVsQU1IVy9zS25SeHphYU8welpNVmVlQnRnNlFGZ1F3M0dJMGo2ZTY0K2w3VExoOW8wSkZVCjdpUitPY01xUzVDY0NROGpWV3JPSk9Xc2dEbDZ4T2FFREczYnR5SVJHY29jbVcvcEZFQjNZd1A2S1BRTUIrNXkKWDdnZExEWmFGRFBVakZmblhkMnhHdUZlMnpRTDNVbXZEUkZuUlBBaW02QlpQbWo1OFh2emFhZXROa3lyaUZLZwp4Z0Z1dVpTcDUwV2JWdjF0MkdzOTMrRE53NlhFZHRFYnlWWUNBa28xTTY0MkozczFnN3NoQnRFQ0F3RUFBYU1qCk1DRXdEZ1lEVlIwUEFRSC9CQVFEQWdLa01BOEdBMVVkRXdFQi93UUZNQU1CQWY4d0RRWUpLb1pJaHZjTkFRRUwKQlFBRGdnR0JBRXYxM2o2ZGFVbDBqeERPSnNTV1ZJZS9JdXNXVTRpN2ZXSWlqMHAwRU1GNS9LTE8yemRndTR5SQoreVd2T3N5aVFPanEzMlRYVlo2bTRDSnBkR1dGVE5HK2lLdXVOU3M0N3g3Q3dmVUNBWm5VVzhyamd3ZWJyS3BmCkJMNEVQcTZTcW0rSmltN0VPankyMWJkY2cyUXdZb3A3eUhvaHcveWEvL0l6RTMzVzZxNHlJeEFvNDBVYUhPTEMKTGtGbnNVYitjcFZBeFlPZGp6bjFzNWhnclpuWXlETEl3WmtIdFdEWm94alUzeC9jdnZzZ1FzLytzTWYrRFU4RgpZMkJKRHJjQ1VQM2xzclc0QVpFMFplZkEwOTlncFEvb3dSN0REYnMwSjZUeFM4NGt6Tldjc1FuWnRraXZheHJNClkyVHNnaWVndFExVFdGRWpxLy9sUFV4emJCdmpnd1FBZm5CQXZGeVNKejdTa0VuVm5rUXJGaUlUQVArTHljQVIKMlg4UFI2ZGI1bEt0SitBSENDM3kvZmNQS2k0ZzNTL3djeXRRdmdvOXJ6ODRFalp5YUNTaGJXNG9jNzNrMS9RcAowQWtHRDU0ZGVDWWVPYVJNbW96c0w3ZzdxWkpFekhtODdOcVBYSy9EZFoweWNxaVFhMXY2T3QxNjdXNUlzMUkzCjBWb0IzUzloSlE9PQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCgo=" # noqa: E501
209 encrypted_cacert
= "QeV4evTLXzcKwZZvmXQ/OvSHToXH3ISwfoLmU+Q9JlQWAFUHSJ9IhO0ewaQrJmx3NkfFb7NCxsQhh+wE57zDW4rWgn4w/SWkzvwSi1h2xYOO3ECEHzzVqgUm15Sk0xaj1Fv9Ed4hipf6PRijeOZ7A1G9zekr1w9WIvebMyJZrK+f6QJ8AP20NUZqG/3k+MeJr3kjrl+8uwU5aPOrHAexSQGAqSKTkWzW7glmlyMWTjwkuSgNVgFg0ctdWTZ5JnNwxXbpjwIKrC4E4sIHcxko2vsTeLF8pZFPk+3QUZIg8BrgtyM3lJC2kO1g3emPQhCIk3VDb5GBgssc/GyFyRXNS651d5BNgcABOKZ4Rv/gGnprB35zP7TKJKkST44XJTEBiugWMkSZg+T9H98/l3eE34O6thfTZXgIyG+ZM6uGlW2XOce0OoEIyJiEL039WJe3izjbD3b9sCCdgQc0MgS+hTaayJI6oCUWPsJLmRji19jLi/wjOsU5gPItCFWw3pBye/A4Zf8Hxm+hShvqBnk8R2yx1fPTiyw/Zx4Jn8m49XQJyjDSZnhIck0PVHR9xWzKCr++PKljLMLdkdFxVRVPFQk/FBbesqofjSXsq9DASY6ACTL3Jmignx2OXD6ac4SlBqCTjV2dIM0yEgZF7zwMNCtppRdXTV8S29JP4W2mfaiqXCUSRTggv8EYU+9diCE+8sPB6HjuLrsfiySbFlYR2m4ysDGXjsVx5CDAf0Nh4IRfcSceYnnBGIQ2sfgGcJFOZoJqr/QeE2NWz6jlWYbWT7MjS/0decpKxP7L88qrR+F48WXQvfsvjWgKjlMKw7lHmFF8FeY836VWWICTRZx+y6IlY1Ys2ML4kySF27Hal4OPhOOoBljMNMVwUEvBulOnKUWw4BGz8eGCl8Hw6tlyJdC7kcBj/aCyNCR/NnuDk4Wck6e//He8L6mS83OJi/hIFc8vYQxnCJMXj9Ou7wr5hxtBnvxXzZM3kFHxCDO24Cd5UyBV9GD8TiQJfBGAy7a2BCBMb5ESVX8NOkyyv2hXMHOjpnKhUM9yP3Ke4CBImO7mCKJNHdFVtAmuyVKJ+jT6ooAAArkX2xwEAvBEpvGNmW2jgs6wxSuKY0h5aUm0rA4v/s8fqSZhzdInB54sMldyAnt9G+9e+g933DfyA/tkc56Ed0vZ/XEvTkThVHyUbfYR/Gjsoab1RpnDBi4aZ2E7iceoBshy+L6NXdL0jlWEs4ZubiWlbVNWlN/MqJcjV/quLU7q4HtkG0MDEFm6To3o48x7xpv8otih6YBduNqBFnwQ6Qz9rM2chFgOR4IgNSZKPxHO0AGCi1gnK/CeCvrSfWYAMn+2rmw0hMZybqKMStG28+rXsKDdqmy6vAwL/+dJwkAW+ix68rWRXpeqHlWidu4SkIBELuwEkFIC/GJU/DRvcN2GG9uP1m+VFifCIS2UdiO4OVrP6PVoW1O+jBJvFH3K1YT7CRqevb9OzjS9fO1wjkOff0W8zZyJK9Mp25aynpf0k3oMpZDpjnlOsFXFUb3N6SvXD1Yi95szIlmsr5yRYaeGUJH7/SAmMr8R6RqsCR0ANptL2dtRoGPi/qcDQE15vnjJ+QMYCg9KbCdV+Qq5di93XAjmwPj6tKZv0aXQuaTZgYR7bdLmAnJaFLbHWcQG1k6F/vdKNEb7llLsoAD9KuKXPZT/LErIyKcI0RZySy9yvhTZb4jQWn17b83yfvqfd5/2NpcyaY4gNERhDRJHw7VhoS5Leai5ZnFaO3C1vU9tIJ85XgCUASTsBLoQWVCKPSQZGxzF7PVLnHui3YA5OsOQpVqAPtgGZ12tP9XkEKj+u2/Atj2bgYrqBF7zUL64X/AQpwr/UElWDhJLSD/KStVeDOUx3AwAVVi9eTUJr6NiNMutCE1sqUf9XVIddgZ/BaG5t3NV2L+T+11QzAl+Xrh8wH/XeUCTmnU3NGkvCz/9Y7PMS+qQL7T7WeGdYmEhb5s/5p/yjSYeqybr5sANOHs83OdeSXbop9cLWW+JksHmS//rHHcrrJhZgCb3P0EOpEoEMCarT6sJq0V1Hwf/YNFdJ9V7Ac654ALS+a9ffNthMUEJeY21QMtNOrEg3QH5RWBPn+yOYN/f38tzwlT1k6Ec94y/sBmeQVv8rRzkkiMSXeAL5ATdJntq8NQq5JbvLQDNnZnHQthZt+uhcUf08mWlRrxxBUaE6xLppgMqFdYSjLGvgn/d8FZ9y7UCg5ZBhgP1rrRQL1COpNKKlJLf5laqwiGAucIDmzSbhO+MidSauDLWuv+fsdd2QYk98PHxqNrPYLrlAlABFi3JEApBm4IlrGbHxKg6dRiy7L1c9xWnAD7E3XrZrSc6DXvGRsjMXWoQdlp4CX5H3cdH9sjIE6akWqiwwrOP6QTbJcxmJGv/MVhsDVrVKmrKSn2H0/Us1fyYCHCOyCSc2L96uId8i9wQO1NXj+1PJmUq3tJ8U0TUwTblOEQdYej99xEI8EzsXLjNJHCgbDygtHBYd/SHToXH3ISwfoLmU+Q9JlS1woaUpVa5sdvbsr4BXR6J" # noqa: E501
211 self
.vca_collection
.find_one
.return_value
= {
212 "_id": "2ade7f0e-9b58-4dbd-93a3-4ec076185d39",
213 "schema_version": "1.11",
216 "secret": encrypted_secret
,
217 "cacert": encrypted_cacert
,
219 self
.admin_collection
.find_one
.return_value
= {
220 "serial": b
"l+U3HDp9td+UjQ+AN+Ypj/Uh7n3C+rMJueQNNxkIpWI="
222 connection_data
= self
.loop
.run_until_complete(
223 self
.store
.get_vca_connection_data("vca_id")
225 self
.assertEqual(connection_data
.endpoints
, [])
226 self
.assertEqual(connection_data
.user
, "admin")
227 self
.assertEqual(connection_data
.secret
, secret
)
229 connection_data
.cacert
, b64decode(cacert
.encode("utf-8")).decode("utf-8")
232 def test_update_vca_endpoints_exception(self
):
233 endpoints
= ["1.2.3.4:17070"]
234 self
.admin_collection
.find_one
.side_effect
= [None, None]
235 self
.admin_collection
.insert_one
.side_effect
= DbException("already exists")
236 with self
.assertRaises(DbException
):
237 self
.loop
.run_until_complete(self
.store
.update_vca_endpoints(endpoints
))
238 self
.assertEqual(self
.admin_collection
.find_one
.call_count
, 2)
239 self
.admin_collection
.replace_one
.assert_not_called()
241 def test_update_vca_endpoints_with_vca_id(self
):
242 endpoints
= ["1.2.3.4:17070"]
243 self
.vca_collection
.find_one
.return_value
= {}
244 self
.loop
.run_until_complete(
245 self
.store
.update_vca_endpoints(endpoints
, "vca_id")
247 self
.vca_collection
.find_one
.assert_called_once_with({"_id": "vca_id"})
248 self
.vca_collection
.replace_one
.assert_called_once_with(
249 {"_id": "vca_id"}, {"endpoints": endpoints
}
252 def test_get_vca_endpoints(self
):
253 endpoints
= ["1.2.3.4:17070"]
254 db_data
= {"api_endpoints": endpoints
}
255 db_returns
= [db_data
, None]
256 expected_returns
= [endpoints
, []]
258 self
.admin_collection
.find_one
.side_effect
= db_returns
259 for _
in range(len(db_returns
)):
260 e
= self
.loop
.run_until_complete(self
.store
.get_vca_endpoints())
262 self
.assertEqual(expected_returns
, returns
)
264 @patch("n2vc.vca.connection_data.base64_to_cacert")
265 def test_get_vca_endpoints_with_vca_id(self
, mock_base64_to_cacert
):
266 expected_endpoints
= ["1.2.3.4:17070"]
267 mock_base64_to_cacert
.return_value
= "cacert"
268 self
.store
.get_vca_connection_data
= AsyncMock()
269 self
.store
.get_vca_connection_data
.return_value
= ConnectionData(
271 "endpoints": expected_endpoints
,
277 endpoints
= self
.loop
.run_until_complete(self
.store
.get_vca_endpoints("vca_id"))
278 self
.store
.get_vca_connection_data
.assert_called_with("vca_id")
279 self
.assertEqual(expected_endpoints
, endpoints
)
281 def test_get_vca_id(self
):
282 self
.assertIsNone(self
.loop
.run_until_complete((self
.store
.get_vca_id())))
284 def test_get_vca_id_with_vim_id(self
):
285 self
.vim_accounts_collection
.find_one
.return_value
= {"vca": "vca_id"}
286 vca_id
= self
.loop
.run_until_complete(self
.store
.get_vca_id("vim_id"))
287 self
.vim_accounts_collection
.find_one
.assert_called_once_with({"_id": "vim_id"})
288 self
.assertEqual(vca_id
, "vca_id")