Attachment #7218 for bug #37402

View | Details | Raw Unified | Return to bug 37402
Collapse All | Expand All

(-)a/branches/ucs-4.1/ucs-4.1-0/base/univention-config-registry/debian/changelog (+6 lines)

Lines 1-3	Link Here

univention-config-registry (11.0.0-2) unstable; urgency=low

  * Bug #37402: Make boolean handling more robust.

 -- Philipp Hahn <hahn@univention.de>  Wed, 21 Oct 2015 08:09:46 +0200

univention-config-registry (11.0.0-1) unstable; urgency=medium

univention-config-registry (11.0.0-1) unstable; urgency=medium

  * Bump version for UCS 4.1 (Bug #38011)

  * Bump version for UCS 4.1 (Bug #38011)





	def is_true(self, key=None, default=False, value=None):
		"""Return if the strings value of key is considered as true."""
		if value is None:
			value = self.get(key)
			if value is None:
			else:
				return default
		return value.lower() in ('yes', 'true', '1', 'enable', 'enabled', 'on')

	def is_false(self, key=None, default=False, value=None):
		"""Return if the strings value of key is considered as false."""
		if value is None:
			value = self.get(key)
			if value is None:
			else:
				return default
		return value.lower() in ('no', 'false', '0', 'disable', 'disabled', 'off')

	def update(self, changes):
		"""

			changed[key] = (old_value, new_value)
		return changed


class _ConfigRegistry(dict):
	"""
	Persistent value store.




sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.path.pardir, 'python'))
from univention.config_registry.backend import ConfigRegistry


class TestConfigRegistry(unittest.TestCase):
	"""Unit test for univention.config_registry.backend.ConfigRegistry"""
	def setUp(self):

		"""Test valid is_true()."""
		ucr = ConfigRegistry()
		for ucr['foo'] in ('YES', 'yes', 'Yes', 'true', '1', 'enable', 'enabled', 'on'):
			self.assertTrue(ucr.is_true('foo'), 'is_true(%(foo)r)' % ucr)

	def test_is_true_invalid(self):
		"""Test invalid is_true()."""
		ucr = ConfigRegistry()
		for ucr['foo'] in ('yes ', ' yes', ''):
			self.assertFalse(ucr.is_true('foo'), 'is_true(%(foo)r)' % ucr)

	def test_is_true_valid_direct(self):
		"""Test valid is_true(value)."""
		ucr = ConfigRegistry()
		for value in ('YES', 'Yes', 'yes', 'true', '1', 'enable', 'enabled', 'on'):
			self.assertTrue(ucr.is_true(value=value), 'is_true(v=%r)' % value)

	def test_is_true_invalid_direct(self):
		"""Test invalid is_true(value)."""
		ucr = ConfigRegistry()
		for value in ('yes ', ' yes', ''):
			self.assertFalse(ucr.is_true(value=value), 'is_true(v=%r)' % value)

	def test_is_false_unset(self):
		"""Test unset is_false()."""

		"""Test valid is_false()."""
		ucr = ConfigRegistry()
		for ucr['foo'] in ('NO', 'no', 'No', 'false', '0', 'disable', 'disabled', 'off'):
			self.assertTrue(ucr.is_false('foo'), 'is_false(%(foo)r)' % ucr)

	def test_is_false_invalid(self):
		"""Test invalid is_false()."""
		ucr = ConfigRegistry()
		for ucr['foo'] in ('no ', ' no', ''):
			self.assertFalse(ucr.is_false('foo'), 'is_false(%(foo)r)' % ucr)

	def test_is_false_valid_direct(self):
		"""Test valid is_false(value)."""
		ucr = ConfigRegistry()
		for value in ('NO', 'No', 'no', 'false', '0', 'disable', 'disabled', 'off'):
			self.assertTrue(ucr.is_false(value=value), 'is_false(v=%r)' % value)

	def test_is_false_invalid_direct(self):
		"""Test valid is_false(value)."""
		ucr = ConfigRegistry()
		for value in ('no ', ' no', ''):
			self.assertFalse(ucr.is_false(value=value), 'is_false(v=%r)' % value)

	def test_update(self):
		"""Test update()."""
- 
  - UMC doesn't do the locking, but even with it it fails.
  - moving the "assertEqual()" inside the locks makes the problem go
    away, as then a concurrent "clear()" can not get in between.
--
.../base/univention-config-registry/debian/rules   |  1 +
.../python/univention/config_registry/backend.py   | 27 ++++----
.../tests/test_backend_threading.py                | 78 ++++++++++++++++++++++
3 files changed, 92 insertions(+), 14 deletions(-)
create mode 100644 branches/ucs-4.1/ucs-4.1-0/base/univention-config-registry/tests/test_backend_threading.py

(-)a/branches/ucs-4.1/ucs-4.1-0/base/univention-config-registry/debian/rules (+1 lines)

Lines 50-55 override_dh_auto_test:	Link Here

	ucslint

	ucslint

	python tests/test_info_tools.py

	python tests/test_info_tools.py

	python tests/test_backend.py

	python tests/test_backend.py

	python tests/test_backend_threading.py

	python python/univention/debhelper.py

	python python/univention/debhelper.py

	python python/univention/config_registry/misc.py

	python python/univention/config_registry/misc.py

	UNIVENTION_BASECONF=tests/base.conf python python/univention/config_registry/interfaces.py

	UNIVENTION_BASECONF=tests/base.conf python python/univention/config_registry/interfaces.py





	def load(self):
		"""Load sub registry from file."""
		self.clear()
		import_failed = False
		try:
			reg_file = open(self.file, 'r')
		except EnvironmentError:
			import_failed = True
		else:
			import_failed = reg_file.read() == '' and reg_file.read() == ''
				import_failed = True

		if import_failed:
			try:

				return

		reg_file.seek(0)
		new = {}
		for line in reg_file:
			line = re.sub(r'^[^:]*#.*$', "", line)
			if line == '':
				continue

				continue

			key, value = line.split(': ', 1)
			new[key] = value.strip()
			if len(value) == 0:  # if variable was set without an value
				value = ''

			self[key] = value
		reg_file.close()

		self.update(new)
		for key in set(self.keys()) - set(new.keys()):
			self.pop(key, None)

		if import_failed:
			self.__save_file(self.file)

	def __create_base_conf(self):
		"""Create sub registry file."""
		try:
			reg_file = os.open(self.file, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0644)
			os.close(reg_file)
		except EnvironmentError as ex:
			if ex.errno != errno.EEXIST:
				msg = "E: file '%s' does not exist and could not be created"
				print >> sys.stderr, msg % (self.file,)
				exception_occured()




#!/usr/bin/python
"""Unit test for univention.config_registry.backend."""
# pylint: disable-msg=C0103,E0611,R0904
import unittest
import os
import sys
from tempfile import mkdtemp
from shutil import rmtree
from threading import Thread, Lock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.path.pardir, 'python'))
from univention.config_registry.backend import ConfigRegistry


class DummyLock(object):
	def __enter__(self):
		pass

	def __exit__(self, exc_type, exc_value, traceback):
		pass


class TestConfigRegistry(unittest.TestCase):
	"""Unit test for univention.config_registry.backend.ConfigRegistry"""
	def setUp(self):
		"""Create object."""
		self.work_dir = mkdtemp()
		ConfigRegistry.PREFIX = self.work_dir

	def tearDown(self):
		"""Destroy object."""
		# os.kill(os.getpid(), 19)  # signal.SIGSTOP
		rmtree(self.work_dir)

	def test_threading(self):
		"""Multiple threads accessing same registry."""
		DO_LOCKING = True
		THREADS = 10
		ITERATIONS = 1000
		BASE, PRIME = 7, 23
		KEY = 'x' * PRIME

		SKEY, SVALUE = 'always', 'there'
		ucr = ConfigRegistry()
		ucr[SKEY] = SVALUE
		ucr.save()

		lock = Lock() if DO_LOCKING else DummyLock()

		def run(tid):
			for iteration in xrange(ITERATIONS):
				i = tid + iteration
				random = pow(BASE, i, PRIME)
				key = KEY[:random + 1]

				with lock:
					ucr.load()
				self.assertEqual(ucr[SKEY], SVALUE, 'tid=%d iter=%d %r' % (tid, iteration, ucr.items()))

				try:
					del ucr[key]
				except LookupError:
					ucr[key] = '%d %d' % (tid, iteration)
				if i % 10 == 0 and tid % 10 == 0:
					with lock:
						ucr.save()

		threads = []
		for tid in xrange(THREADS):
			thread = Thread(target=run, name='%d' % tid, args=(tid,))
			threads.append(thread)
		for thread in threads:
			thread.start()
		for thread in threads:
			thread.join()


if __name__ == '__main__':
	unittest.main()

Return to bug 37402