14aa5a582e97390e0aaeed3e3485873a8de73e7f
[osm/LCM.git] / osm_lcm / tests / test_lcm_utils.py
1 # Copyright 2022 Canonical Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
4 # not use this file except in compliance with the License. You may obtain
5 # 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, WITHOUT
11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 # License for the specific language governing permissions and limitations
13 # under the License.
14 #
15 # For those usages not covered by the Apache License, Version 2.0 please
16 # contact: alfonso.tiernosepulveda@telefonica.com
17 ##
18 import logging
19 import tempfile
20 from unittest.mock import Mock, patch, MagicMock, mock_open
21 from unittest import TestCase
22
23 from osm_common.msgkafka import MsgKafka
24 from osm_common import fslocal
25 from osm_lcm.data_utils.database.database import Database
26 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
27 from osm_lcm.lcm_utils import LcmBase, LcmException, vld_to_ro_ip_profile
28 from osm_lcm.tests import test_db_descriptors as descriptors
29 import yaml
30 from zipfile import BadZipfile
31
32
33 tmpdir = tempfile.mkdtemp()[1]
34 tmpfile = tempfile.mkstemp()[1]
35
36
37 class TestLcmBase(TestCase):
38 test_nsr_id = "f48163a6-c807-47bc-9682-f72caef5af85"
39 test_nsd_id = "8c2f8b95-bb1b-47ee-8001-36dc090678da"
40 nsd_package_path = "/" + test_nsd_id
41 nsd_package_name = "test_nsd"
42 charm_metadata_file = "/path/charm/metadata.yaml"
43
44 def setUp(self):
45 # DB
46 Database.instance = None
47 self.db = Database({"database": {"driver": "memory"}}).instance.db
48 self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text))
49 self.db.create_list("nsds_revisions", yaml.safe_load(descriptors.db_nsds_text))
50 self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text))
51 # Filesystem
52 self.fs = Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
53 # Create LCMBase class
54 self.msg = Mock(MsgKafka())
55 self.logger = Mock(logging)
56 self.my_ns = LcmBase(self.msg, self.logger)
57 self.my_ns.fs = self.fs
58 self.my_ns.db = self.db
59 self.hexdigest = "031edd7d41651593c5fe5c006f"
60
61 def test_get_charm_name_successfully(self):
62 instance = self.my_ns
63 mock_open = MagicMock(open)
64 mock_yaml = MagicMock(yaml)
65 mock_yaml.safe_load.return_value = {"name": "test_charm"}
66 expected_result = "test_charm"
67
68 with patch("osm_lcm.lcm_utils.open", mock_open), patch(
69 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
70 ):
71 result = instance.get_charm_name(TestLcmBase.charm_metadata_file)
72 self.assertEqual(result, expected_result, "wrong charm name")
73 self.assertEqual(mock_yaml.safe_load.call_count, 1)
74 self.assertEqual(mock_open.call_count, 1)
75
76 def test_get_charm_name_can_not_open_metadata_file(self):
77 instance = self.my_ns
78 mock_open = MagicMock(open)
79 mock_open.side_effect = IOError
80 mock_yaml = MagicMock(create_autospec=True)
81
82 with patch("osm_lcm.lcm_utils.open", mock_open), patch(
83 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
84 ):
85 with self.assertRaises(IOError):
86 instance.get_charm_name(TestLcmBase.charm_metadata_file)
87 mock_yaml.safe_load.assert_not_called()
88 self.assertEqual(mock_open.call_count, 1)
89
90 def test_get_charm_name_wrong_metadata_file_format(self):
91 instance = self.my_ns
92 mock_open = MagicMock(open)
93 mock_yaml = MagicMock(create_autospec=True)
94 mock_yaml.safe_load.return_value = {}
95
96 with patch("osm_lcm.lcm_utils.open", mock_open), patch(
97 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
98 ):
99 with self.assertRaises(KeyError):
100 instance.get_charm_name(TestLcmBase.charm_metadata_file)
101 self.assertEqual(mock_open.call_count, 1)
102 self.assertEqual(mock_yaml.safe_load.call_count, 1)
103
104 def test_get_charm_path_successfully(self):
105 instance = self.my_ns
106 fs = fslocal.FsLocal()
107 fs.path = "/app/storage"
108 instance.fs = fs
109 charm_folder_name = "simple_charm"
110 expected_result = (
111 "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
112 )
113 result = instance._get_charm_path(
114 TestLcmBase.nsd_package_path,
115 TestLcmBase.nsd_package_name,
116 charm_folder_name,
117 )
118 self.assertEqual(result, expected_result, "wrong_charm_path")
119
120 def test_get_charm_metadata_file_charm_is_not_zipped(self):
121 instance = self.my_ns
122 fs = fslocal.FsLocal()
123 fs.path = "/app/storage"
124 instance.fs = fs
125 mock_zipfile = MagicMock(create_autospec=True)
126 charm_folder_name = "simple_charm"
127 charm_path = (
128 "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
129 )
130 expected_result = (
131 "/app/storage/"
132 + TestLcmBase.test_nsd_id
133 + "/test_nsd/charms/simple_charm/metadata.yaml"
134 )
135
136 with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
137 result = instance._get_charm_metadata_file(
138 charm_folder_name,
139 TestLcmBase.nsd_package_path,
140 TestLcmBase.nsd_package_name,
141 charm_path=charm_path,
142 )
143 self.assertEqual(result, expected_result, "wrong charm metadata path")
144 mock_zipfile.assert_not_called()
145
146 def test_get_charm_metadata_file_charm_is_zipped(self):
147 instance = self.my_ns
148 fs = fslocal.FsLocal()
149 fs.path = "/app/storage"
150 instance.fs = fs
151 mock_zipfile = MagicMock(create_autospec=True)
152 mock_zipfile.side_effect = None
153 charm_folder_name = "ubuntu_18.04_simple_charm2.charm"
154 charm_path = (
155 "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
156 )
157 expected_result = (
158 "/app/storage/"
159 + TestLcmBase.test_nsd_id
160 + "/test_nsd/charms/ubuntu_18.04_simple_charm2/metadata.yaml"
161 )
162
163 with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
164 result = instance._get_charm_metadata_file(
165 charm_folder_name,
166 TestLcmBase.nsd_package_path,
167 TestLcmBase.nsd_package_name,
168 charm_path=charm_path,
169 )
170 self.assertEqual(result, expected_result, "wrong charm metadata path")
171 self.assertEqual(mock_zipfile.call_count, 1)
172
173 def test_find_charm_name_successfully(self):
174 db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
175 instance = self.my_ns
176 mock_charm_path = MagicMock()
177 mock_metadata_file = MagicMock()
178 mock_metadata_file.return_value = (
179 "/" + TestLcmBase.test_nsd_id + "/new_test_nsd/charms/simple/metadata.yaml"
180 )
181 mock_charm_name = MagicMock()
182 mock_charm_name.return_value = "test_charm"
183 expected_result = "test_charm"
184
185 with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
186 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
187 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
188 result = instance.find_charm_name(db_nsr, "simple")
189 self.assertEqual(result, expected_result, "Wrong charm name")
190 mock_charm_path.assert_called_once()
191 mock_metadata_file.assert_called_once()
192 mock_charm_name.assert_called_once_with(
193 "/"
194 + TestLcmBase.test_nsd_id
195 + "/new_test_nsd/charms/simple/metadata.yaml"
196 )
197
198 def test_find_charm_name_charm_bad_zipfile(self):
199 db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
200 instance = self.my_ns
201 mock_charm_path = MagicMock()
202 mock_metadata_file = MagicMock()
203 mock_metadata_file.side_effect = BadZipfile
204 mock_charm_name = MagicMock()
205
206 with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
207 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
208 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
209 with self.assertRaises(LcmException):
210 instance.find_charm_name(db_nsr, "simple")
211 self.assertEqual(mock_charm_path.call_count, 1)
212 self.assertEqual(mock_metadata_file.call_count, 1)
213 mock_charm_name.assert_not_called()
214
215 def test_find_charm_name_missing_input_charm_folder_name(self):
216 db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
217 instance = self.my_ns
218 mock_metadata_file = MagicMock()
219 mock_charm_name = MagicMock()
220 mock_charm_path = MagicMock()
221
222 with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
223 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
224 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
225 with self.assertRaises(LcmException):
226 instance.find_charm_name(db_nsr, "")
227 mock_charm_path.assert_not_called()
228 mock_metadata_file.assert_not_called()
229 mock_charm_name.assert_not_called()
230
231 def test_find_charm_name_can_not_open_metadata_file(self):
232 db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
233 instance = self.my_ns
234
235 mock_charm_path = MagicMock()
236 mock_metadata_file = MagicMock()
237 mock_charm_name = MagicMock()
238 mock_charm_name.side_effect = yaml.YAMLError
239
240 with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
241 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
242 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
243 with self.assertRaises(LcmException):
244 instance.find_charm_name(db_nsr, "simple")
245 self.assertEqual(mock_charm_path.call_count, 1)
246 self.assertEqual(mock_metadata_file.call_count, 1)
247 self.assertEqual(mock_charm_name.call_count, 1)
248
249 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
250 @patch("osm_lcm.lcm_utils.hashlib")
251 def test_calculate_charm_hash_sucessfully(self, mock_hashlib, mocking_open):
252 """Calculate charm hash successfully."""
253 charm = tmpfile
254 hexdigest = self.hexdigest
255 mock_file_hash = MagicMock()
256 mock_hashlib.sha256.return_value = mock_file_hash
257 mock_file_hash.hexdigest.return_value = hexdigest
258 result = LcmBase.calculate_charm_hash(charm)
259 self.assertEqual(result, hexdigest)
260 self.assertEqual(mocking_open.call_count, 1)
261 self.assertEqual(mock_file_hash.update.call_count, 1)
262 self.assertEqual(mock_file_hash.hexdigest.call_count, 1)
263 self.assertEqual(mock_hashlib.sha256.call_count, 1)
264
265 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
266 @patch("osm_lcm.lcm_utils.hashlib")
267 def test_calculate_charm_hash_open_raises(self, mock_hashlib, mocking_open):
268 """builtins.open raises exception."""
269 charm = tmpfile
270 hexdigest = self.hexdigest
271 mock_file_hash = MagicMock()
272 mock_hashlib.sha256.return_value = mock_file_hash
273 mock_file_hash.hexdigest.return_value = hexdigest
274 mocking_open.side_effect = IOError
275 with self.assertRaises(IOError):
276 LcmBase.calculate_charm_hash(charm)
277 self.assertEqual(mocking_open.call_count, 1)
278 mock_file_hash.update.assert_not_called()
279 mock_file_hash.hexdigest.assert_not_called()
280 self.assertEqual(mock_hashlib.sha256.call_count, 1)
281
282 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
283 @patch("osm_lcm.lcm_utils.hashlib")
284 def test_calculate_charm_filehash_update_raises(self, mock_hashlib, mocking_open):
285 """Filehash update raises exception."""
286 charm = tmpfile
287 hexdigest = self.hexdigest
288 mock_file_hash = MagicMock()
289 mock_file_hash.update.side_effect = Exception
290 mock_hashlib.sha256.return_value = mock_file_hash
291 mock_file_hash.hexdigest.return_value = hexdigest
292 with self.assertRaises(Exception):
293 LcmBase.calculate_charm_hash(charm)
294 self.assertEqual(mocking_open.call_count, 1)
295 self.assertEqual(mock_file_hash.update.call_count, 1)
296 mock_file_hash.hexdigest.assert_not_called()
297 self.assertEqual(mock_hashlib.sha256.call_count, 1)
298
299 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
300 @patch("osm_lcm.lcm_utils.hashlib")
301 def test_calculate_charm_filehash_hexdigest_raises(
302 self, mock_hashlib, mocking_open
303 ):
304 """Filehash hexdigest raises exception."""
305 charm = tmpfile
306 mock_file_hash = MagicMock()
307 mock_hashlib.sha256.return_value = mock_file_hash
308 mock_file_hash.hexdigest.side_effect = Exception
309 with self.assertRaises(Exception):
310 LcmBase.calculate_charm_hash(charm)
311 self.assertEqual(mocking_open.call_count, 1)
312 self.assertEqual(mock_file_hash.update.call_count, 1)
313 mock_file_hash.hexdigest.assert_called_once()
314 self.assertEqual(mock_hashlib.sha256.call_count, 1)
315 mock_file_hash.update.assert_called_once()
316
317 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
318 @patch("osm_lcm.lcm_utils.hashlib")
319 def test_calculate_charm_filehash_hashlib_sha256_raises(
320 self, mock_hashlib, mocking_open
321 ):
322 """Filehash hashlib sha256 raises exception."""
323 charm = tmpfile
324 mock_hashlib.sha256.side_effect = Exception
325 with self.assertRaises(Exception):
326 LcmBase.calculate_charm_hash(charm)
327 self.assertEqual(mock_hashlib.sha256.call_count, 1)
328 mocking_open.assert_not_called()
329
330 @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
331 @patch("osm_lcm.lcm_utils.hashlib")
332 def test_calculate_charm_hash_file_does_not_exist(self, mock_hashlib, mocking_open):
333 """Calculate charm hash, charm file does not exist."""
334 file = None
335 mock_file_hash = MagicMock()
336 mock_hashlib.sha256.return_value = mock_file_hash
337 mocking_open.side_effect = FileNotFoundError
338 with self.assertRaises(FileNotFoundError):
339 LcmBase.calculate_charm_hash(file)
340 self.assertEqual(mocking_open.call_count, 1)
341 mock_file_hash.update.assert_not_called()
342 mock_file_hash.hexdigest.assert_not_called()
343 self.assertEqual(mock_hashlib.sha256.call_count, 1)
344
345 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
346 def test_compare_charm_hash_charm_changed(self, mock_calculate_charm_hash):
347 """Compare charm hash, charm files are different."""
348 output = True
349 charm1, charm2 = tmpfile, tmpfile
350 mock_calculate_charm_hash.side_effect = [
351 self.hexdigest,
352 "0dd7d4173747593c5fe5c006f",
353 ]
354 result = LcmBase.compare_charm_hash(charm1, charm2)
355 self.assertEqual(output, result)
356 self.assertEqual(mock_calculate_charm_hash.call_count, 2)
357
358 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
359 def test_compare_charm_hash_charm_is_same(self, mock_calculate_charm_hash):
360 """Compare charm hash, charm files are same."""
361 output = False
362 charm1 = charm2 = tmpfile
363 mock_calculate_charm_hash.side_effect = [
364 self.hexdigest,
365 self.hexdigest,
366 ]
367 result = LcmBase.compare_charm_hash(charm1, charm2)
368 self.assertEqual(output, result)
369 self.assertEqual(mock_calculate_charm_hash.call_count, 2)
370
371 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
372 def test_compare_charm_hash_one_charm_is_not_valid(self, mock_calculate_charm_hash):
373 """Compare charm hash, one charm file is not valid."""
374 charm1, charm2 = tmpdir, None
375 mock_calculate_charm_hash.side_effect = [
376 self.hexdigest,
377 FileNotFoundError,
378 ]
379
380 with self.assertRaises(FileNotFoundError):
381 LcmBase.compare_charm_hash(charm1, charm2)
382 self.assertEqual(mock_calculate_charm_hash.call_count, 2)
383
384 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
385 def test_compare_charm_hash_both_charms_are_not_valid(
386 self, mock_calculate_charm_hash
387 ):
388 """Compare charm hash, both charm files are not valid."""
389 charm1, charm2 = None, None
390 mock_calculate_charm_hash.side_effect = [IOError, IOError]
391 with self.assertRaises(IOError):
392 LcmBase.compare_charm_hash(charm1, charm2)
393 self.assertEqual(mock_calculate_charm_hash.call_count, 1)
394
395 @patch("osm_lcm.lcm_utils.checksumdir")
396 def test_compare_charmdir_charm_changed(self, mock_checksum):
397 """Compare charm directory hash, charms are changed."""
398 expected_output = True
399 charm_dir1, charm_dir2 = tmpdir, tmpdir
400 mock_checksum.dirhash.side_effect = [
401 self.hexdigest,
402 "031eddtrtr651593c5fe5c006f",
403 ]
404 result = LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
405 self.assertEqual(expected_output, result)
406 self.assertEqual(mock_checksum.dirhash.call_count, 2)
407
408 @patch("osm_lcm.lcm_utils.checksumdir")
409 def test_compare_charmdir_charm_is_same(self, mock_checksum):
410 """Compare charm directory hash, charms are same."""
411 expected_output = False
412 charm_dir1 = charm_dir2 = tmpdir
413 mock_checksum.dirhash.side_effect = [
414 self.hexdigest,
415 self.hexdigest,
416 ]
417 result = LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
418 self.assertEqual(expected_output, result)
419 self.assertEqual(mock_checksum.dirhash.call_count, 2)
420
421 @patch("osm_lcm.lcm_utils.checksumdir")
422 def test_compare_charmdir_one_charmdir_is_not_valid(self, mock_checksum):
423 """Compare charm directory hash, one charm directory is not valid."""
424 charm_dir1, charm_dir2 = tmpdir, None
425 mock_checksum.dirhash.side_effect = [
426 self.hexdigest,
427 FileNotFoundError,
428 ]
429 with self.assertRaises(FileNotFoundError):
430 LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
431 self.assertEqual(mock_checksum.dirhash.call_count, 2)
432
433 @patch("osm_lcm.lcm_utils.checksumdir")
434 def test_compare_charmdir_both_charmdirs_are_not_valid(self, mock_checksum):
435 """Compare charm directory hash, both charm directories are not valid."""
436 charm_dir1, charm_dir2 = None, None
437 mock_checksum.dirhash.side_effect = [FileNotFoundError, FileNotFoundError]
438 with self.assertRaises(FileNotFoundError):
439 LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
440 self.assertEqual(mock_checksum.dirhash.call_count, 1)
441
442
443 class TestLcmUtils(TestCase):
444 def setUp(self):
445 pass
446
447 def test__vld_to_ro_ip_profile_with_none(self):
448 vld_data = None
449
450 result = vld_to_ro_ip_profile(
451 source_data=vld_data,
452 )
453
454 self.assertIsNone(result)
455
456 def test__vld_to_ro_ip_profile_with_empty_profile(self):
457 vld_data = {}
458
459 result = vld_to_ro_ip_profile(
460 source_data=vld_data,
461 )
462
463 self.assertIsNone(result)
464
465 def test__vld_to_ro_ip_profile_with_wrong_profile(self):
466 vld_data = {
467 "no-profile": "here",
468 }
469 expected_result = {
470 "ip_version": "IPv4",
471 "subnet_address": None,
472 "gateway_address": None,
473 "dns_address": None,
474 "dhcp_enabled": False,
475 "dhcp_start_address": None,
476 "dhcp_count": None,
477 "ipv6_address_mode": None,
478 }
479
480 result = vld_to_ro_ip_profile(
481 source_data=vld_data,
482 )
483
484 self.assertDictEqual(expected_result, result)
485
486 def test__vld_to_ro_ip_profile_with_ipv4_profile(self):
487 vld_data = {
488 "ip-version": "ipv4",
489 "cidr": "192.168.0.0/24",
490 "gateway-ip": "192.168.0.254",
491 "dhcp-enabled": True,
492 "dns-server": [{"address": "8.8.8.8"}],
493 }
494 expected_result = {
495 "ip_version": "IPv4",
496 "subnet_address": "192.168.0.0/24",
497 "gateway_address": "192.168.0.254",
498 "dns_address": "8.8.8.8",
499 "dhcp_enabled": True,
500 "dhcp_start_address": None,
501 "dhcp_count": None,
502 "ipv6_address_mode": None,
503 }
504
505 result = vld_to_ro_ip_profile(
506 source_data=vld_data,
507 )
508
509 self.assertDictEqual(expected_result, result)
510
511 def test__vld_to_ro_ip_profile_with_ipv6_profile(self):
512 vld_data = {
513 "ip-version": "ipv6",
514 "cidr": "2001:0200:0001::/48",
515 "gateway-ip": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
516 "dhcp-enabled": True,
517 }
518 expected_result = {
519 "ip_version": "IPv6",
520 "subnet_address": "2001:0200:0001::/48",
521 "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
522 "dns_address": None,
523 "dhcp_enabled": True,
524 "dhcp_start_address": None,
525 "dhcp_count": None,
526 "ipv6_address_mode": None,
527 }
528
529 result = vld_to_ro_ip_profile(
530 source_data=vld_data,
531 )
532
533 self.assertDictEqual(expected_result, result)