Package portage :: Package tests :: Package locks :: Module test_asynchronous_lock
[hide private]

Source Code for Module portage.tests.locks.test_asynchronous_lock

  1  # Copyright 2010-2011 Gentoo Foundation 
  2  # Distributed under the terms of the GNU General Public License v2 
  3   
  4  import signal 
  5  import tempfile 
  6   
  7  from portage import os 
  8  from portage import shutil 
  9  from portage.tests import TestCase 
 10  from portage.util._eventloop.global_event_loop import global_event_loop 
 11  from _emerge.AsynchronousLock import AsynchronousLock 
 12   
13 -class AsynchronousLockTestCase(TestCase):
14
15 - def _testAsynchronousLock(self):
16 scheduler = global_event_loop() 17 tempdir = tempfile.mkdtemp() 18 try: 19 path = os.path.join(tempdir, 'lock_me') 20 for force_async in (True, False): 21 for force_dummy in (True, False): 22 async_lock = AsynchronousLock(path=path, 23 scheduler=scheduler, _force_async=force_async, 24 _force_thread=True, 25 _force_dummy=force_dummy) 26 async_lock.start() 27 self.assertEqual(async_lock.wait(), os.EX_OK) 28 self.assertEqual(async_lock.returncode, os.EX_OK) 29 async_lock.unlock() 30 31 async_lock = AsynchronousLock(path=path, 32 scheduler=scheduler, _force_async=force_async, 33 _force_process=True) 34 async_lock.start() 35 self.assertEqual(async_lock.wait(), os.EX_OK) 36 self.assertEqual(async_lock.returncode, os.EX_OK) 37 async_lock.unlock() 38 39 finally: 40 shutil.rmtree(tempdir)
41
42 - def testAsynchronousLock(self):
44 54
56 scheduler = global_event_loop() 57 tempdir = tempfile.mkdtemp() 58 try: 59 path = os.path.join(tempdir, 'lock_me') 60 lock1 = AsynchronousLock(path=path, scheduler=scheduler) 61 lock1.start() 62 self.assertEqual(lock1.wait(), os.EX_OK) 63 self.assertEqual(lock1.returncode, os.EX_OK) 64 65 # lock2 requires _force_async=True since the portage.locks 66 # module is not designed to work as intended here if the 67 # same process tries to lock the same file more than 68 # one time concurrently. 69 lock2 = AsynchronousLock(path=path, scheduler=scheduler, 70 _force_async=True, _force_process=True) 71 lock2.start() 72 # lock2 should be waiting for lock1 to release 73 self.assertEqual(lock2.poll(), None) 74 self.assertEqual(lock2.returncode, None) 75 76 lock1.unlock() 77 self.assertEqual(lock2.wait(), os.EX_OK) 78 self.assertEqual(lock2.returncode, os.EX_OK) 79 lock2.unlock() 80 finally: 81 shutil.rmtree(tempdir)
82
83 - def testAsynchronousLockWait(self):
85 95
97 scheduler = global_event_loop() 98 tempdir = tempfile.mkdtemp() 99 try: 100 path = os.path.join(tempdir, 'lock_me') 101 lock1 = AsynchronousLock(path=path, scheduler=scheduler) 102 lock1.start() 103 self.assertEqual(lock1.wait(), os.EX_OK) 104 self.assertEqual(lock1.returncode, os.EX_OK) 105 lock2 = AsynchronousLock(path=path, scheduler=scheduler, 106 _force_async=True, _force_process=True) 107 lock2.start() 108 # lock2 should be waiting for lock1 to release 109 self.assertEqual(lock2.poll(), None) 110 self.assertEqual(lock2.returncode, None) 111 112 # Cancel lock2 and then check wait() and returncode results. 113 lock2.cancel() 114 self.assertEqual(lock2.wait() == os.EX_OK, False) 115 self.assertEqual(lock2.returncode == os.EX_OK, False) 116 self.assertEqual(lock2.returncode is None, False) 117 lock1.unlock() 118 finally: 119 shutil.rmtree(tempdir)
120 123 133
135 scheduler = global_event_loop() 136 tempdir = tempfile.mkdtemp() 137 try: 138 path = os.path.join(tempdir, 'lock_me') 139 lock1 = AsynchronousLock(path=path, scheduler=scheduler) 140 lock1.start() 141 self.assertEqual(lock1.wait(), os.EX_OK) 142 self.assertEqual(lock1.returncode, os.EX_OK) 143 lock2 = AsynchronousLock(path=path, scheduler=scheduler, 144 _force_async=True, _force_process=True) 145 lock2.start() 146 # lock2 should be waiting for lock1 to release 147 self.assertEqual(lock2.poll(), None) 148 self.assertEqual(lock2.returncode, None) 149 150 # Kill lock2's process and then check wait() and 151 # returncode results. This is intended to simulate 152 # a SIGINT sent via the controlling tty. 153 self.assertEqual(lock2._imp is not None, True) 154 self.assertEqual(lock2._imp._proc is not None, True) 155 self.assertEqual(lock2._imp._proc.pid is not None, True) 156 lock2._imp._kill_test = True 157 os.kill(lock2._imp._proc.pid, signal.SIGTERM) 158 self.assertEqual(lock2.wait() == os.EX_OK, False) 159 self.assertEqual(lock2.returncode == os.EX_OK, False) 160 self.assertEqual(lock2.returncode is None, False) 161 lock1.unlock() 162 finally: 163 shutil.rmtree(tempdir)
164 167
177