test_asynchronous_lock: simulate SIGINT from tty

1 files changed, 33 insertions, 1 deletions

diff --git a/pym/portage/tests/locks/test_asynchronous_lock.py b/pym/portage/tests/locks/test_asynchronous_lock.py
index 50b9c93a5..6e8bc0a07 100644
--- a/pym/portage/tests/locks/test_asynchronous_lock.py
+++ b/pym/portage/tests/locks/test_asynchronous_lock.py
@@ -1,7 +1,8 @@
-# Copyright 2010 Gentoo Foundation
+# Copyright 2010-2011 Gentoo Foundation
 # Distributed under the terms of the GNU General Public License v2
 
 import shutil
+import signal
 import tempfile
 
 from portage import os
@@ -90,3 +91,34 @@ class AsynchronousLockTestCase(TestCase):
 			lock1.unlock()
 		finally:
 			shutil.rmtree(tempdir)
+
+	def testAsynchronousLockWaitKill(self):
+		scheduler = PollScheduler().sched_iface
+		tempdir = tempfile.mkdtemp()
+		try:
+			path = os.path.join(tempdir, 'lock_me')
+			lock1 = AsynchronousLock(path=path, scheduler=scheduler)
+			lock1.start()
+			self.assertEqual(lock1.wait(), os.EX_OK)
+			self.assertEqual(lock1.returncode, os.EX_OK)
+			lock2 = AsynchronousLock(path=path, scheduler=scheduler,
+				_force_async=True, _force_process=True)
+			lock2.start()
+			# lock2 should we waiting for lock1 to release
+			self.assertEqual(lock2.poll(), None)
+			self.assertEqual(lock2.returncode, None)
+
+			# Kill lock2's process and then check wait() and
+			# returncode results. This is intended to simulate
+			# a SIGINT sent via the controlling tty.
+			self.assertEqual(lock2._imp is not None, True)
+			self.assertEqual(lock2._imp._proc is not None, True)
+			self.assertEqual(lock2._imp._proc.pid is not None, True)
+			lock2._imp._kill_test = True
+			os.kill(lock2._imp._proc.pid, signal.SIGTERM)
+			self.assertEqual(lock2.wait() == os.EX_OK, False)
+			self.assertEqual(lock2.returncode == os.EX_OK, False)
+			self.assertEqual(lock2.returncode is None, False)
+			lock1.unlock()
+		finally:
+			shutil.rmtree(tempdir)