Add a new AsynchronousLock class that uses the portage.locks module to

acquire a lock asynchronously, using a background thread.

3 files changed, 31 insertions, 0 deletions

diff --git a/pym/portage/tests/locks/__init__.py b/pym/portage/tests/locks/__init__.py
new file mode 100644
index 000000000..21a391aee
--- /dev/null
+++ b/pym/portage/tests/locks/__init__.py
@@ -0,0 +1,2 @@
+# Copyright 2010 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
diff --git a/pym/portage/tests/locks/__test__ b/pym/portage/tests/locks/__test__
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/pym/portage/tests/locks/__test__
diff --git a/pym/portage/tests/locks/test_asynchronous_lock.py b/pym/portage/tests/locks/test_asynchronous_lock.py
new file mode 100644
index 000000000..ac38462ed
--- /dev/null
+++ b/pym/portage/tests/locks/test_asynchronous_lock.py
@@ -0,0 +1,29 @@
+# Copyright 2010 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import shutil
+import tempfile
+
+from portage import os
+from portage.tests import TestCase
+from _emerge.AsynchronousLock import AsynchronousLock
+from _emerge.PollScheduler import PollScheduler
+
+class AsynchronousLockTestCase(TestCase):
+
+	def testAsynchronousLock(self):
+		scheduler = PollScheduler().sched_iface
+		tempdir = tempfile.mkdtemp()
+		try:
+			path = os.path.join(tempdir, 'lock_me')
+			for force_thread in (True, False):
+				for force_dummy in (True, False):
+					async_lock = AsynchronousLock(path=path,
+						scheduler=scheduler, _force_dummy=force_dummy,
+						_force_thread=force_thread)
+					async_lock.start()
+					async_lock.wait()
+					async_lock.unlock()
+					self.assertEqual(async_lock.returncode, os.EX_OK)
+		finally:
+			shutil.rmtree(tempdir)