Differential D1016 Diff 3236 swh/core/tests/test_config.py

Changeset View

Standalone View

swh/core/tests/test_config.py

	# Copyright (C) 2015 The Software Heritage developers			# Copyright (C) 2015 The Software Heritage developers
	# See the AUTHORS file at the top-level directory of this distribution			# See the AUTHORS file at the top-level directory of this distribution
	# License: GNU General Public License version 3, or any later version			# License: GNU General Public License version 3, or any later version
	# See top-level LICENSE file for more information			# See top-level LICENSE file for more information

	import os			import os
	import shutil			import shutil
	import tempfile			import pytest
	import unittest			import pkg_resources.extern.packaging.version

	from swh.core import config			from swh.core import config

				pytest_v = pkg_resources.get_distribution("pytest").parsed_version
				if pytest_v < pkg_resources.extern.packaging.version.parse('3.9'):
				@pytest.fixture
				def tmp_path(request):
				import tempfile
				import pathlib
				with tempfile.TemporaryDirectory() as tmpdir:
				yield pathlib.Path(tmpdir)

	class ConfReaderTest(unittest.TestCase):

	@classmethod
	def setUpClass(cls):
	# create a temporary folder
	cls.tmpdir = tempfile.mkdtemp(prefix='test-swh-core.')
	cls.conffile = os.path.join(cls.tmpdir, 'config.ini')
	conf_contents = """[main]
	a = 1
	b = this is a string
	c = true
	h = false
	ls = list, of, strings
	li = 1, 2, 3, 4
	"""
	with open(cls.conffile, 'w') as conf:
	conf.write(conf_contents)

	cls.non_existing_conffile = os.path.join(cls.tmpdir,
	'config-nonexisting.ini')

	# Create an unreadable, proper configuration file
	cls.perms_broken_file = os.path.join(cls.tmpdir, 'unreadable.ini')
	with open(cls.perms_broken_file, 'w') as conf:
	conf.write(conf_contents)
	os.chmod(cls.perms_broken_file, 0o000)

	# Create a proper configuration file in an unreadable directory
	cls.perms_broken_dir = os.path.join(cls.tmpdir, 'unreadabledir')
	cls.file_in_broken_dir = os.path.join(cls.perms_broken_dir,
	'unreadable.ini')
	os.makedirs(cls.perms_broken_dir)
	with open(cls.file_in_broken_dir, 'w') as conf:
	conf.write(conf_contents)
	os.chmod(cls.perms_broken_dir, 0o000)

	cls.empty_conffile = os.path.join(cls.tmpdir, 'empty.ini')
	open(cls.empty_conffile, 'w').close()

	cls.default_conf = {			default_conf = {
	'a': ('int', 2),			'a': ('int', 2),
	'b': ('string', 'default-string'),			'b': ('string', 'default-string'),
	'c': ('bool', True),			'c': ('bool', True),
	'd': ('int', 10),			'd': ('int', 10),
	'e': ('int', None),			'e': ('int', None),
	'f': ('bool', None),			'f': ('bool', None),
	'g': ('string', None),			'g': ('string', None),
	'h': ('bool', True),			'h': ('bool', True),
	'i': ('bool', True),			'i': ('bool', True),
	'ls': ('list[str]', ['a', 'b', 'c']),			'ls': ('list[str]', ['a', 'b', 'c']),
	'li': ('list[int]', [42, 43]),			'li': ('list[int]', [42, 43]),
	}			}

	cls.other_default_conf = {			other_default_conf = {
	'a': ('int', 3),			'a': ('int', 3),
	}			}

	cls.full_default_conf = cls.default_conf.copy()			full_default_conf = default_conf.copy()
	cls.full_default_conf['a'] = cls.other_default_conf['a']			full_default_conf['a'] = other_default_conf['a']

	cls.parsed_default_conf = {			parsed_default_conf = {
	key: value			key: value
	for key, (type, value)			for key, (type, value)
	in cls.default_conf.items()			in default_conf.items()
	}			}

	cls.parsed_conffile = {			parsed_conffile = {
	'a': 1,			'a': 1,
	'b': 'this is a string',			'b': 'this is a string',
	'c': True,			'c': True,
	'd': 10,			'd': 10,
	'e': None,			'e': None,
	'f': None,			'f': None,
	'g': None,			'g': None,
	'h': False,			'h': False,
	'i': True,			'i': True,
	'ls': ['list', 'of', 'strings'],			'ls': ['list', 'of', 'strings'],
	'li': [1, 2, 3, 4],			'li': [1, 2, 3, 4],
	}			}

	@classmethod
	def tearDownClass(cls):			@pytest.fixture
				def swh_config(tmp_path):
				# create a temporary folder
				ardumontUnsubmitted Not Done Inline Actions out of curiosity, from where does that writing come from? ardumont: out of curiosity, from where does that writing come from?
				conffile = tmp_path / 'config.ini'
				conf_contents = """[main]
				a = 1
				b = this is a string
				c = true
				h = false
				ls = list, of, strings
				li = 1, 2, 3, 4
				"""
				conffile.open('w').write(conf_contents)
				return conffile


				@pytest.fixture
				def swh_config_unreadable(swh_config):
				# Create an unreadable, proper configuration file
				os.chmod(str(swh_config), 0o000)
				yield swh_config
				# Make the broken perms file readable again to be able to remove them
				os.chmod(str(swh_config), 0o644)


				@pytest.fixture
				def swh_config_unreadable_dir(swh_config):
				# Create a proper configuration file in an unreadable directory
				perms_broken_dir = swh_config.parent / 'unreadabledir'
				perms_broken_dir.mkdir()
				shutil.move(str(swh_config), str(perms_broken_dir))
				os.chmod(str(perms_broken_dir), 0o000)
				yield perms_broken_dir / swh_config.name
	# Make the broken perms items readable again to be able to remove them			# Make the broken perms items readable again to be able to remove them
	os.chmod(cls.perms_broken_dir, 0o755)			os.chmod(str(perms_broken_dir), 0o755)
	os.chmod(cls.perms_broken_file, 0o644)
	shutil.rmtree(cls.tmpdir)
				@pytest.fixture
				def swh_config_empty(tmp_path):
				# create a temporary folder
				conffile = tmp_path / 'config.ini'
				conffile.touch()
				return conffile


	def test_read(self):			def test_read(swh_config):
	# when			# when
	res = config.read(self.conffile, self.default_conf)			res = config.read(str(swh_config), default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_conffile)			assert res == parsed_conffile


	def test_read_empty_file(self):			def test_read_empty_file():
	# when			# when
	res = config.read(None, self.default_conf)			res = config.read(None, default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_default_conf)			assert res == parsed_default_conf

	def test_support_non_existing_conffile(self):
				def test_support_non_existing_conffile(tmp_path):
	# when			# when
	res = config.read(self.non_existing_conffile, self.default_conf)			res = config.read(str(tmp_path / 'void.ini'), default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_default_conf)			assert res == parsed_default_conf


	def test_support_empty_conffile(self):			def test_support_empty_conffile(swh_config_empty):
	# when			# when
	res = config.read(self.empty_conffile, self.default_conf)			res = config.read(str(swh_config_empty), default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_default_conf)			assert res == parsed_default_conf


	def test_raise_on_broken_directory_perms(self):			def test_raise_on_broken_directory_perms(swh_config_unreadable_dir):
	with self.assertRaises(PermissionError):			with pytest.raises(PermissionError):
	config.read(self.file_in_broken_dir, self.default_conf)			config.read(str(swh_config_unreadable_dir), default_conf)

	def test_raise_on_broken_file_perms(self):
	with self.assertRaises(PermissionError):
	config.read(self.perms_broken_file, self.default_conf)

	def test_merge_default_configs(self):			def test_raise_on_broken_file_perms(swh_config_unreadable):
				with pytest.raises(PermissionError):
				config.read(str(swh_config_unreadable), default_conf)


				def test_merge_default_configs():
	# when			# when
	res = config.merge_default_configs(self.default_conf,			res = config.merge_default_configs(default_conf, other_default_conf)
	self.other_default_conf)

	# then			# then
	self.assertEqual(res, self.full_default_conf)			assert res == full_default_conf


	def test_priority_read_nonexist_conf(self):			def test_priority_read_nonexist_conf(swh_config):
				noexist = str(swh_config.parent / 'void.ini')
	# when			# when
	res = config.priority_read([self.non_existing_conffile, self.conffile],			res = config.priority_read([noexist, str(swh_config)], default_conf)
	self.default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_conffile)			assert res == parsed_conffile


				def test_priority_read_conf_nonexist_empty(swh_config):
				noexist = swh_config.parent / 'void.ini'
				empty = swh_config.parent / 'empty.ini'
				empty.touch()

	def test_priority_read_conf_nonexist_empty(self):
	# when			# when
	res = config.priority_read([			res = config.priority_read([str(p) for p in (
	self.conffile,			swh_config, noexist, empty)], default_conf)
	self.non_existing_conffile,
	self.empty_conffile,
	], self.default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_conffile)			assert res == parsed_conffile


				def test_priority_read_empty_conf_nonexist(swh_config):
				noexist = swh_config.parent / 'void.ini'
				empty = swh_config.parent / 'empty.ini'
				empty.touch()

	def test_priority_read_empty_conf_nonexist(self):
	# when			# when
	res = config.priority_read([			res = config.priority_read([str(p) for p in (
	self.empty_conffile,			empty, swh_config, noexist)], default_conf)
	self.conffile,
	self.non_existing_conffile,
	], self.default_conf)

	# then			# then
	self.assertEqual(res, self.parsed_default_conf)			assert res == parsed_default_conf


	def test_swh_config_paths(self):			def test_swh_config_paths():
	res = config.swh_config_paths('foo/bar.ini')			res = config.swh_config_paths('foo/bar.ini')

	self.assertEqual(res, [			assert res == [
	'~/.config/swh/foo/bar.ini',			'~/.config/swh/foo/bar.ini',
	'~/.swh/foo/bar.ini',			'~/.swh/foo/bar.ini',
	'/etc/softwareheritage/foo/bar.ini',			'/etc/softwareheritage/foo/bar.ini',
	])			]


	def test_prepare_folder(self):			def test_prepare_folder(tmp_path):
	# given			# given
	conf = {'path1': os.path.join(self.tmpdir, 'path1'),			conf = {'path1': str(tmp_path / 'path1'),
	'path2': os.path.join(self.tmpdir, 'path2', 'depth1')}			'path2': str(tmp_path / 'path2' / 'depth1')}

	# the folders does not exists			# the folders does not exists
	self.assertFalse(os.path.exists(conf['path1']),			assert not os.path.exists(conf['path1']), "path1 should not exist."
	"path1 should not exist.")			assert not os.path.exists(conf['path2']), "path2 should not exist."
	self.assertFalse(os.path.exists(conf['path2']),
	"path2 should not exist.")

	# when			# when
	config.prepare_folders(conf, 'path1')			config.prepare_folders(conf, 'path1')

	# path1 exists but not path2			# path1 exists but not path2
	self.assertTrue(os.path.exists(conf['path1']),			assert os.path.exists(conf['path1']), "path1 should now exist!"
	"path1 should now exist!")			assert not os.path.exists(conf['path2']), "path2 should not exist."
	self.assertFalse(os.path.exists(conf['path2']),
	"path2 should not exist.")

	# path1 already exists, skips it but creates path2			# path1 already exists, skips it but creates path2
	config.prepare_folders(conf, 'path1', 'path2')			config.prepare_folders(conf, 'path1', 'path2')

	self.assertTrue(os.path.exists(conf['path1']),			assert os.path.exists(conf['path1']), "path1 should still exist!"
	"path1 should still exist!")			assert os.path.exists(conf['path2']), "path2 should now exist."
	self.assertTrue(os.path.exists(conf['path2']),
	"path2 should now exist.")