1 # Copyright 2022 Canonical Ltd.
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
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, WITHOUT
11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 # License for the specific language governing permissions and limitations
15 # For those usages not covered by the Apache License, Version 2.0 please
16 # contact: alfonso.tiernosepulveda@telefonica.com
20 from unittest
.mock
import Mock
, patch
, MagicMock
, mock_open
21 from unittest
import TestCase
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
28 from osm_lcm
.tests
import test_db_descriptors
as descriptors
30 from zipfile
import BadZipfile
33 tmpdir
= tempfile
.mkdtemp()[1]
34 tmpfile
= tempfile
.mkstemp()[1]
37 class TestLcmBase(TestCase
):
39 test_nsr_id
= "f48163a6-c807-47bc-9682-f72caef5af85"
40 test_nsd_id
= "8c2f8b95-bb1b-47ee-8001-36dc090678da"
41 nsd_package_path
= "/" + test_nsd_id
42 nsd_package_name
= "test_nsd"
43 charm_metadata_file
= "/path/charm/metadata.yaml"
47 Database
.instance
= None
48 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
49 self
.db
.create_list("nsds", yaml
.safe_load(descriptors
.db_nsds_text
))
50 self
.db
.create_list("nsds_revisions", yaml
.safe_load(descriptors
.db_nsds_text
))
51 self
.db
.create_list("nsrs", yaml
.safe_load(descriptors
.db_nsrs_text
))
53 self
.fs
= Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
54 # Create LCMBase class
55 self
.msg
= Mock(MsgKafka())
56 self
.logger
= Mock(logging
)
57 self
.my_ns
= LcmBase(self
.msg
, self
.logger
)
58 self
.my_ns
.fs
= self
.fs
59 self
.my_ns
.db
= self
.db
60 self
.hexdigest
= "031edd7d41651593c5fe5c006f"
62 def test_get_charm_name_successfully(self
):
64 mock_open
= MagicMock(open)
65 mock_yaml
= MagicMock(yaml
)
66 mock_yaml
.safe_load
.return_value
= {"name": "test_charm"}
67 expected_result
= "test_charm"
69 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
70 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
72 result
= instance
.get_charm_name(TestLcmBase
.charm_metadata_file
)
73 self
.assertEqual(result
, expected_result
, "wrong charm name")
74 self
.assertEqual(mock_yaml
.safe_load
.call_count
, 1)
75 self
.assertEqual(mock_open
.call_count
, 1)
77 def test_get_charm_name_can_not_open_metadata_file(self
):
79 mock_open
= MagicMock(open)
80 mock_open
.side_effect
= IOError
81 mock_yaml
= MagicMock(create_autospec
=True)
83 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
84 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
86 with self
.assertRaises(IOError):
87 instance
.get_charm_name(TestLcmBase
.charm_metadata_file
)
88 mock_yaml
.safe_load
.assert_not_called()
89 self
.assertEqual(mock_open
.call_count
, 1)
91 def test_get_charm_name_wrong_metadata_file_format(self
):
93 mock_open
= MagicMock(open)
94 mock_yaml
= MagicMock(create_autospec
=True)
95 mock_yaml
.safe_load
.return_value
= {}
97 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
98 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
100 with self
.assertRaises(KeyError):
101 instance
.get_charm_name(TestLcmBase
.charm_metadata_file
)
102 self
.assertEqual(mock_open
.call_count
, 1)
103 self
.assertEqual(mock_yaml
.safe_load
.call_count
, 1)
105 def test_get_charm_path_successfully(self
):
106 instance
= self
.my_ns
107 fs
= fslocal
.FsLocal()
108 fs
.path
= "/app/storage"
110 charm_folder_name
= "simple_charm"
112 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
114 result
= instance
._get
_charm
_path
(
115 TestLcmBase
.nsd_package_path
,
116 TestLcmBase
.nsd_package_name
,
119 self
.assertEqual(result
, expected_result
, "wrong_charm_path")
121 def test_get_charm_metadata_file_charm_is_not_zipped(self
):
122 instance
= self
.my_ns
123 fs
= fslocal
.FsLocal()
124 fs
.path
= "/app/storage"
126 mock_zipfile
= MagicMock(create_autospec
=True)
127 charm_folder_name
= "simple_charm"
129 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
133 + TestLcmBase
.test_nsd_id
134 + "/test_nsd/charms/simple_charm/metadata.yaml"
137 with
patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile
):
138 result
= instance
._get
_charm
_metadata
_file
(
140 TestLcmBase
.nsd_package_path
,
141 TestLcmBase
.nsd_package_name
,
142 charm_path
=charm_path
,
144 self
.assertEqual(result
, expected_result
, "wrong charm metadata path")
145 mock_zipfile
.assert_not_called()
147 def test_get_charm_metadata_file_charm_is_zipped(self
):
148 instance
= self
.my_ns
149 fs
= fslocal
.FsLocal()
150 fs
.path
= "/app/storage"
152 mock_zipfile
= MagicMock(create_autospec
=True)
153 mock_zipfile
.side_effect
= None
154 charm_folder_name
= "ubuntu_18.04_simple_charm2.charm"
156 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
160 + TestLcmBase
.test_nsd_id
161 + "/test_nsd/charms/ubuntu_18.04_simple_charm2/metadata.yaml"
164 with
patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile
):
165 result
= instance
._get
_charm
_metadata
_file
(
167 TestLcmBase
.nsd_package_path
,
168 TestLcmBase
.nsd_package_name
,
169 charm_path
=charm_path
,
171 self
.assertEqual(result
, expected_result
, "wrong charm metadata path")
172 self
.assertEqual(mock_zipfile
.call_count
, 1)
174 def test_find_charm_name_successfully(self
):
175 db_nsr
= self
.db
.get_one("nsrs", {"_id": TestLcmBase
.test_nsr_id
})
176 instance
= self
.my_ns
177 mock_charm_path
= MagicMock()
178 mock_metadata_file
= MagicMock()
179 mock_metadata_file
.return_value
= (
180 "/" + TestLcmBase
.test_nsd_id
+ "/new_test_nsd/charms/simple/metadata.yaml"
182 mock_charm_name
= MagicMock()
183 mock_charm_name
.return_value
= "test_charm"
184 expected_result
= "test_charm"
186 with
patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path
), patch(
187 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
188 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name
):
190 result
= instance
.find_charm_name(db_nsr
, "simple")
191 self
.assertEqual(result
, expected_result
, "Wrong charm name")
192 mock_charm_path
.assert_called_once()
193 mock_metadata_file
.assert_called_once()
194 mock_charm_name
.assert_called_once_with(
196 + TestLcmBase
.test_nsd_id
197 + "/new_test_nsd/charms/simple/metadata.yaml"
200 def test_find_charm_name_charm_bad_zipfile(self
):
201 db_nsr
= self
.db
.get_one("nsrs", {"_id": TestLcmBase
.test_nsr_id
})
202 instance
= self
.my_ns
203 mock_charm_path
= MagicMock()
204 mock_metadata_file
= MagicMock()
205 mock_metadata_file
.side_effect
= BadZipfile
206 mock_charm_name
= MagicMock()
208 with
patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path
), patch(
209 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
210 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name
):
212 with self
.assertRaises(LcmException
):
214 instance
.find_charm_name(db_nsr
, "simple")
215 self
.assertEqual(mock_charm_path
.call_count
, 1)
216 self
.assertEqual(mock_metadata_file
.call_count
, 1)
217 mock_charm_name
.assert_not_called()
219 def test_find_charm_name_missing_input_charm_folder_name(self
):
220 db_nsr
= self
.db
.get_one("nsrs", {"_id": TestLcmBase
.test_nsr_id
})
221 instance
= self
.my_ns
222 mock_metadata_file
= MagicMock()
223 mock_charm_name
= MagicMock()
224 mock_charm_path
= MagicMock()
226 with
patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path
), patch(
227 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
228 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name
):
230 with self
.assertRaises(LcmException
):
231 instance
.find_charm_name(db_nsr
, "")
232 mock_charm_path
.assert_not_called()
233 mock_metadata_file
.assert_not_called()
234 mock_charm_name
.assert_not_called()
236 def test_find_charm_name_can_not_open_metadata_file(self
):
237 db_nsr
= self
.db
.get_one("nsrs", {"_id": TestLcmBase
.test_nsr_id
})
238 instance
= self
.my_ns
240 mock_charm_path
= MagicMock()
241 mock_metadata_file
= MagicMock()
242 mock_charm_name
= MagicMock()
243 mock_charm_name
.side_effect
= yaml
.YAMLError
245 with
patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path
), patch(
246 "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
247 ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name
):
249 with self
.assertRaises(LcmException
):
250 instance
.find_charm_name(db_nsr
, "simple")
251 self
.assertEqual(mock_charm_path
.call_count
, 1)
252 self
.assertEqual(mock_metadata_file
.call_count
, 1)
253 self
.assertEqual(mock_charm_name
.call_count
, 1)
255 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
256 @patch("osm_lcm.lcm_utils.hashlib")
257 def test_calculate_charm_hash_sucessfully(self
, mock_hashlib
, mocking_open
):
258 """Calculate charm hash successfully."""
260 hexdigest
= self
.hexdigest
261 mock_file_hash
= MagicMock()
262 mock_hashlib
.md5
.return_value
= mock_file_hash
263 mock_file_hash
.hexdigest
.return_value
= hexdigest
264 result
= LcmBase
.calculate_charm_hash(charm
)
265 self
.assertEqual(result
, hexdigest
)
266 self
.assertEqual(mocking_open
.call_count
, 1)
267 self
.assertEqual(mock_file_hash
.update
.call_count
, 1)
268 self
.assertEqual(mock_file_hash
.hexdigest
.call_count
, 1)
269 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
271 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
272 @patch("osm_lcm.lcm_utils.hashlib")
273 def test_calculate_charm_hash_open_raises(self
, mock_hashlib
, mocking_open
):
274 """builtins.open raises exception."""
276 hexdigest
= self
.hexdigest
277 mock_file_hash
= MagicMock()
278 mock_hashlib
.md5
.return_value
= mock_file_hash
279 mock_file_hash
.hexdigest
.return_value
= hexdigest
280 mocking_open
.side_effect
= IOError
281 with self
.assertRaises(IOError):
282 LcmBase
.calculate_charm_hash(charm
)
283 self
.assertEqual(mocking_open
.call_count
, 1)
284 mock_file_hash
.update
.assert_not_called()
285 mock_file_hash
.hexdigest
.assert_not_called()
286 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
288 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
289 @patch("osm_lcm.lcm_utils.hashlib")
290 def test_calculate_charm_filehash_update_raises(self
, mock_hashlib
, mocking_open
):
291 """Filehash update raises exception."""
293 hexdigest
= self
.hexdigest
294 mock_file_hash
= MagicMock()
295 mock_file_hash
.update
.side_effect
= Exception
296 mock_hashlib
.md5
.return_value
= mock_file_hash
297 mock_file_hash
.hexdigest
.return_value
= hexdigest
298 with self
.assertRaises(Exception):
299 LcmBase
.calculate_charm_hash(charm
)
300 self
.assertEqual(mocking_open
.call_count
, 1)
301 self
.assertEqual(mock_file_hash
.update
.call_count
, 1)
302 mock_file_hash
.hexdigest
.assert_not_called()
303 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
305 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
306 @patch("osm_lcm.lcm_utils.hashlib")
307 def test_calculate_charm_filehash_hexdigest_raises(
308 self
, mock_hashlib
, mocking_open
310 """Filehash hexdigest raises exception."""
312 mock_file_hash
= MagicMock()
313 mock_hashlib
.md5
.return_value
= mock_file_hash
314 mock_file_hash
.hexdigest
.side_effect
= Exception
315 with self
.assertRaises(Exception):
316 LcmBase
.calculate_charm_hash(charm
)
317 self
.assertEqual(mocking_open
.call_count
, 1)
318 self
.assertEqual(mock_file_hash
.update
.call_count
, 1)
319 mock_file_hash
.hexdigest
.assert_called_once()
320 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
321 mock_file_hash
.update
.assert_called_once()
323 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
324 @patch("osm_lcm.lcm_utils.hashlib")
325 def test_calculate_charm_filehash_hashlib_md5_raises(
326 self
, mock_hashlib
, mocking_open
328 """Filehash hashlib md5 raises exception."""
330 mock_hashlib
.md5
.side_effect
= Exception
331 with self
.assertRaises(Exception):
332 LcmBase
.calculate_charm_hash(charm
)
333 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
334 mocking_open
.assert_not_called()
336 @patch("builtins.open", new_callable
=mock_open(read_data
="charm content"))
337 @patch("osm_lcm.lcm_utils.hashlib")
338 def test_calculate_charm_hash_file_does_not_exist(self
, mock_hashlib
, mocking_open
):
339 """Calculate charm hash, charm file does not exist."""
341 mock_file_hash
= MagicMock()
342 mock_hashlib
.md5
.return_value
= mock_file_hash
343 mocking_open
.side_effect
= FileNotFoundError
344 with self
.assertRaises(FileNotFoundError
):
345 LcmBase
.calculate_charm_hash(file)
346 self
.assertEqual(mocking_open
.call_count
, 1)
347 mock_file_hash
.update
.assert_not_called()
348 mock_file_hash
.hexdigest
.assert_not_called()
349 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
351 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
352 def test_compare_charm_hash_charm_changed(self
, mock_calculate_charm_hash
):
353 """Compare charm hash, charm files are different."""
355 charm1
, charm2
= tmpfile
, tmpfile
356 mock_calculate_charm_hash
.side_effect
= [
358 "0dd7d4173747593c5fe5c006f",
360 result
= LcmBase
.compare_charm_hash(charm1
, charm2
)
361 self
.assertEqual(output
, result
)
362 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
364 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
365 def test_compare_charm_hash_charm_is_same(self
, mock_calculate_charm_hash
):
366 """Compare charm hash, charm files are same."""
368 charm1
= charm2
= tmpfile
369 mock_calculate_charm_hash
.side_effect
= [
373 result
= LcmBase
.compare_charm_hash(charm1
, charm2
)
374 self
.assertEqual(output
, result
)
375 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
377 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
378 def test_compare_charm_hash_one_charm_is_not_valid(self
, mock_calculate_charm_hash
):
379 """Compare charm hash, one charm file is not valid."""
380 charm1
, charm2
= tmpdir
, None
381 mock_calculate_charm_hash
.side_effect
= [
386 with self
.assertRaises(FileNotFoundError
):
387 LcmBase
.compare_charm_hash(charm1
, charm2
)
388 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
390 @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
391 def test_compare_charm_hash_both_charms_are_not_valid(
392 self
, mock_calculate_charm_hash
394 """Compare charm hash, both charm files are not valid."""
395 charm1
, charm2
= None, None
396 mock_calculate_charm_hash
.side_effect
= [IOError, IOError]
397 with self
.assertRaises(IOError):
398 LcmBase
.compare_charm_hash(charm1
, charm2
)
399 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 1)
401 @patch("osm_lcm.lcm_utils.checksumdir")
402 def test_compare_charmdir_charm_changed(self
, mock_checksum
):
403 """Compare charm directory hash, charms are changed."""
404 expected_output
= True
405 charm_dir1
, charm_dir2
= tmpdir
, tmpdir
406 mock_checksum
.dirhash
.side_effect
= [
408 "031eddtrtr651593c5fe5c006f",
410 result
= LcmBase
.compare_charmdir_hash(charm_dir1
, charm_dir2
)
411 self
.assertEqual(expected_output
, result
)
412 self
.assertEqual(mock_checksum
.dirhash
.call_count
, 2)
414 @patch("osm_lcm.lcm_utils.checksumdir")
415 def test_compare_charmdir_charm_is_same(self
, mock_checksum
):
416 """Compare charm directory hash, charms are same."""
417 expected_output
= False
418 charm_dir1
= charm_dir2
= tmpdir
419 mock_checksum
.dirhash
.side_effect
= [
423 result
= LcmBase
.compare_charmdir_hash(charm_dir1
, charm_dir2
)
424 self
.assertEqual(expected_output
, result
)
425 self
.assertEqual(mock_checksum
.dirhash
.call_count
, 2)
427 @patch("osm_lcm.lcm_utils.checksumdir")
428 def test_compare_charmdir_one_charmdir_is_not_valid(self
, mock_checksum
):
429 """Compare charm directory hash, one charm directory is not valid."""
430 charm_dir1
, charm_dir2
= tmpdir
, None
431 mock_checksum
.dirhash
.side_effect
= [
435 with self
.assertRaises(FileNotFoundError
):
436 LcmBase
.compare_charmdir_hash(charm_dir1
, charm_dir2
)
437 self
.assertEqual(mock_checksum
.dirhash
.call_count
, 2)
439 @patch("osm_lcm.lcm_utils.checksumdir")
440 def test_compare_charmdir_both_charmdirs_are_not_valid(self
, mock_checksum
):
441 """Compare charm directory hash, both charm directories are not valid."""
442 charm_dir1
, charm_dir2
= None, None
443 mock_checksum
.dirhash
.side_effect
= [FileNotFoundError
, FileNotFoundError
]
444 with self
.assertRaises(FileNotFoundError
):
445 LcmBase
.compare_charmdir_hash(charm_dir1
, charm_dir2
)
446 self
.assertEqual(mock_checksum
.dirhash
.call_count
, 1)