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
):
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"
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
))
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"
61 def test_get_charm_name_successfully(self
):
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"
68 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
69 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
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)
76 def test_get_charm_name_can_not_open_metadata_file(self
):
78 mock_open
= MagicMock(open)
79 mock_open
.side_effect
= IOError
80 mock_yaml
= MagicMock(create_autospec
=True)
82 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
83 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
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)
90 def test_get_charm_name_wrong_metadata_file_format(self
):
92 mock_open
= MagicMock(open)
93 mock_yaml
= MagicMock(create_autospec
=True)
94 mock_yaml
.safe_load
.return_value
= {}
96 with
patch("osm_lcm.lcm_utils.open", mock_open
), patch(
97 "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml
.safe_load
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)
104 def test_get_charm_path_successfully(self
):
105 instance
= self
.my_ns
106 fs
= fslocal
.FsLocal()
107 fs
.path
= "/app/storage"
109 charm_folder_name
= "simple_charm"
111 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
113 result
= instance
._get
_charm
_path
(
114 TestLcmBase
.nsd_package_path
,
115 TestLcmBase
.nsd_package_name
,
118 self
.assertEqual(result
, expected_result
, "wrong_charm_path")
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"
125 mock_zipfile
= MagicMock(create_autospec
=True)
126 charm_folder_name
= "simple_charm"
128 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
132 + TestLcmBase
.test_nsd_id
133 + "/test_nsd/charms/simple_charm/metadata.yaml"
136 with
patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile
):
137 result
= instance
._get
_charm
_metadata
_file
(
139 TestLcmBase
.nsd_package_path
,
140 TestLcmBase
.nsd_package_name
,
141 charm_path
=charm_path
,
143 self
.assertEqual(result
, expected_result
, "wrong charm metadata path")
144 mock_zipfile
.assert_not_called()
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"
151 mock_zipfile
= MagicMock(create_autospec
=True)
152 mock_zipfile
.side_effect
= None
153 charm_folder_name
= "ubuntu_18.04_simple_charm2.charm"
155 "/app/storage/" + TestLcmBase
.test_nsd_id
+ "/test_nsd/charms/simple_charm"
159 + TestLcmBase
.test_nsd_id
160 + "/test_nsd/charms/ubuntu_18.04_simple_charm2/metadata.yaml"
163 with
patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile
):
164 result
= instance
._get
_charm
_metadata
_file
(
166 TestLcmBase
.nsd_package_path
,
167 TestLcmBase
.nsd_package_name
,
168 charm_path
=charm_path
,
170 self
.assertEqual(result
, expected_result
, "wrong charm metadata path")
171 self
.assertEqual(mock_zipfile
.call_count
, 1)
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"
181 mock_charm_name
= MagicMock()
182 mock_charm_name
.return_value
= "test_charm"
183 expected_result
= "test_charm"
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(
194 + TestLcmBase
.test_nsd_id
195 + "/new_test_nsd/charms/simple/metadata.yaml"
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()
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()
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()
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()
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
235 mock_charm_path
= MagicMock()
236 mock_metadata_file
= MagicMock()
237 mock_charm_name
= MagicMock()
238 mock_charm_name
.side_effect
= yaml
.YAMLError
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)
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."""
254 hexdigest
= self
.hexdigest
255 mock_file_hash
= MagicMock()
256 mock_hashlib
.md5
.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
.md5
.call_count
, 1)
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."""
270 hexdigest
= self
.hexdigest
271 mock_file_hash
= MagicMock()
272 mock_hashlib
.md5
.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
.md5
.call_count
, 1)
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."""
287 hexdigest
= self
.hexdigest
288 mock_file_hash
= MagicMock()
289 mock_file_hash
.update
.side_effect
= Exception
290 mock_hashlib
.md5
.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
.md5
.call_count
, 1)
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
304 """Filehash hexdigest raises exception."""
306 mock_file_hash
= MagicMock()
307 mock_hashlib
.md5
.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
.md5
.call_count
, 1)
315 mock_file_hash
.update
.assert_called_once()
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_md5_raises(
320 self
, mock_hashlib
, mocking_open
322 """Filehash hashlib md5 raises exception."""
324 mock_hashlib
.md5
.side_effect
= Exception
325 with self
.assertRaises(Exception):
326 LcmBase
.calculate_charm_hash(charm
)
327 self
.assertEqual(mock_hashlib
.md5
.call_count
, 1)
328 mocking_open
.assert_not_called()
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."""
335 mock_file_hash
= MagicMock()
336 mock_hashlib
.md5
.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
.md5
.call_count
, 1)
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."""
349 charm1
, charm2
= tmpfile
, tmpfile
350 mock_calculate_charm_hash
.side_effect
= [
352 "0dd7d4173747593c5fe5c006f",
354 result
= LcmBase
.compare_charm_hash(charm1
, charm2
)
355 self
.assertEqual(output
, result
)
356 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
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."""
362 charm1
= charm2
= tmpfile
363 mock_calculate_charm_hash
.side_effect
= [
367 result
= LcmBase
.compare_charm_hash(charm1
, charm2
)
368 self
.assertEqual(output
, result
)
369 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
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
= [
380 with self
.assertRaises(FileNotFoundError
):
381 LcmBase
.compare_charm_hash(charm1
, charm2
)
382 self
.assertEqual(mock_calculate_charm_hash
.call_count
, 2)
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
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)
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
= [
402 "031eddtrtr651593c5fe5c006f",
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)
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
= [
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)
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
= [
429 with self
.assertRaises(FileNotFoundError
):
430 LcmBase
.compare_charmdir_hash(charm_dir1
, charm_dir2
)
431 self
.assertEqual(mock_checksum
.dirhash
.call_count
, 2)
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)