src/third_party/chromite/lib/paygen/flock_unittest.py

   1 #!/usr/bin/python
   2 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """Test flock library.
   7
   8 DEPRECATED: Should be migrated to chromite.lib.locking_unittest.
   9 """
  10
  11 from __future__ import print_function
  12
  13 import mox
  14 import multiprocessing
  15 import os
  16 import sys
  17 import time
  18
  19 import fixup_path
  20 fixup_path.FixupPath()
  21
  22 from chromite.lib import cros_test_lib
  23 from chromite.lib import osutils
  24 from chromite.lib.paygen import flock
  25
  26
  27 LOCK_ACQUIRED = 5
  28 LOCK_NOT_ACQUIRED = 6
  29
  30
  31 class FLockTest(mox.MoxTestBase):
  32   """Test FLock lock class."""
  33
  34   def __init__(self, testCaseNames):
  35     self.tempdir = None
  36     mox.MoxTestBase.__init__(self, testCaseNames)
  37
  38   def setUp(self):
  39     """Prepare for each test."""
  40     self.mox = mox.Mox()
  41
  42     # To make certain we don't self update while running tests.
  43     os.environ['CROSTOOLS_NO_SOURCE_UPDATE'] = '1'
  44
  45   def tearDown(self):
  46     """Cleanup after each test."""
  47     self.mox.UnsetStubs()
  48
  49   @osutils.TempDirDecorator
  50   def _HelperSingleLockTest(self, blocking, shared):
  51     """Helper method that runs a basic test with/without blocking/sharing."""
  52     lock = flock.Lock('SingleLock',
  53                       lock_dir=self.tempdir,
  54                       blocking=blocking,
  55                       shared=shared)
  56
  57     expected_lock_file = os.path.join(self.tempdir, 'SingleLock')
  58
  59     self.assertFalse(os.path.exists(expected_lock_file))
  60     self.assertFalse(lock.IsLocked())
  61     lock.Acquire()
  62     self.assertTrue(os.path.exists(expected_lock_file))
  63     self.assertTrue(lock.IsLocked())
  64
  65     # Acquiring the lock again should be safe.
  66     lock.Acquire()
  67     self.assertTrue(lock.IsLocked())
  68
  69     # Ensure the lock file contains our pid, and nothing else.
  70     fd = open(expected_lock_file, 'r')
  71     self.assertEquals(['%d\n' % os.getpid()], fd.readlines())
  72     fd.close()
  73
  74     lock.Release()
  75     self.assertFalse(lock.IsLocked())
  76
  77   @osutils.TempDirDecorator
  78   def _HelperDoubleLockTest(self, blocking1, shared1, blocking2, shared2):
  79     """Helper method that runs a two-lock test with/without blocking/sharing."""
  80     lock1 = flock.Lock('DoubleLock',
  81                        lock_dir=self.tempdir,
  82                        blocking=blocking1,
  83                        shared=shared1)
  84     lock2 = flock.Lock('DoubleLock',
  85                        lock_dir=self.tempdir,
  86                        blocking=blocking2,
  87                        shared=shared2)
  88
  89     lock1.Acquire()
  90     self.assertTrue(lock1.IsLocked())
  91     self.assertFalse(lock2.IsLocked())
  92
  93     # The second lock should fail to acquire.
  94     self.assertRaises(flock.LockNotAcquired, lock2.Acquire)
  95     self.assertTrue(lock1.IsLocked())
  96     self.assertFalse(lock2.IsLocked())
  97
  98     lock1.Release()
  99     self.assertFalse(lock1.IsLocked())
 100     self.assertFalse(lock2.IsLocked())
 101
 102     # Releasing second lock should be harmless.
 103     lock2.Release()
 104     self.assertFalse(lock1.IsLocked())
 105     self.assertFalse(lock2.IsLocked())
 106
 107   def _HelperInsideProcess(self, name, lock_dir, blocking, shared):
 108     """Helper method that runs a basic test with/without blocking."""
 109
 110     try:
 111       with flock.Lock(name,
 112                       lock_dir=lock_dir,
 113                       blocking=blocking,
 114                       shared=shared):
 115         pass
 116       sys.exit(LOCK_ACQUIRED)
 117     except flock.LockNotAcquired:
 118       sys.exit(LOCK_NOT_ACQUIRED)
 119
 120   def _HelperStartProcess(self, name, blocking=False, shared=False):
 121     """Create a process and invoke _HelperInsideProcess in it."""
 122     p = multiprocessing.Process(target=self._HelperInsideProcess,
 123                                 args=(name, self.tempdir, blocking, shared))
 124     p.start()
 125
 126     # It's highly probably that p will have tried to grab the lock before the
 127     # timer expired, but not certain.
 128     time.sleep(0.1)
 129
 130     return p
 131
 132   def _HelperWithProcess(self, name, expected, blocking=False, shared=False):
 133     """Create a process and invoke _HelperInsideProcess in it."""
 134     p = multiprocessing.Process(target=self._HelperInsideProcess,
 135                                 args=(name, self.tempdir, blocking, shared))
 136     p.start()
 137     p.join()
 138     self.assertEquals(p.exitcode, expected)
 139
 140   def testLockName(self):
 141     """Make sure that we get the expected lock file name."""
 142     lock = flock.Lock(lock_name='/tmp/foo')
 143     self.assertEqual(lock.lock_file, '/tmp/foo')
 144
 145     lock = flock.Lock(lock_name='foo')
 146     self.assertEqual(lock.lock_file, '/tmp/run_once/foo')
 147
 148     lock = flock.Lock(lock_name='foo', lock_dir='/bar')
 149     self.assertEqual(lock.lock_file, '/bar/foo')
 150
 151   def testSingleLock(self):
 152     """Just test getting releasing a lock with options."""
 153     self._HelperSingleLockTest(blocking=False, shared=False)
 154     self._HelperSingleLockTest(blocking=True, shared=False)
 155     self._HelperSingleLockTest(blocking=False, shared=True)
 156     self._HelperSingleLockTest(blocking=True, shared=True)
 157
 158   def testDoubleLock(self):
 159     """Test two lock objects for the same lock file."""
 160     self._HelperDoubleLockTest(blocking1=False, shared1=False,
 161                                blocking2=False, shared2=False)
 162
 163   def testContextMgr(self):
 164     """Make sure we behave properly with 'with'."""
 165
 166     name = 'WithLock'
 167
 168     # Create an instance, and use it in a with
 169     prelock = flock.Lock(name, lock_dir=self.tempdir)
 170     self._HelperWithProcess(name, expected=LOCK_ACQUIRED)
 171
 172     with prelock as lock:
 173       # Assert the instance didn't change.
 174       self.assertIs(prelock, lock)
 175       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 176
 177     self._HelperWithProcess(name, expected=LOCK_ACQUIRED)
 178
 179     # Construct the instance in the with expression.
 180     with flock.Lock(name, lock_dir=self.tempdir) as lock:
 181       self.assertIsInstance(lock, flock.Lock)
 182       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 183
 184     self._HelperWithProcess(name, expected=LOCK_ACQUIRED)
 185
 186   def testAcquireBeforeWith(self):
 187     """Sometimes you want to Acquire a lock and then return it into 'with'."""
 188
 189     name = "WithLock"
 190     lock = flock.Lock(name, lock_dir=self.tempdir)
 191     lock.Acquire()
 192
 193     self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 194
 195     with lock:
 196       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 197
 198     self._HelperWithProcess(name, expected=LOCK_ACQUIRED)
 199
 200   @osutils.TempDirDecorator
 201   def testSingleProcessLock(self):
 202     """Test grabbing the same lock in processes with no conflicts."""
 203     self._HelperWithProcess('ProcessLock', expected=LOCK_ACQUIRED)
 204     self._HelperWithProcess('ProcessLock', expected=LOCK_ACQUIRED)
 205     self._HelperWithProcess('ProcessLock', expected=LOCK_ACQUIRED,
 206                             blocking=True)
 207     self._HelperWithProcess('ProcessLock', expected=LOCK_ACQUIRED,
 208                             shared=True)
 209     self._HelperWithProcess('ProcessLock', expected=LOCK_ACQUIRED,
 210                             blocking=True, shared=True)
 211
 212   @osutils.TempDirDecorator
 213   def testNonBlockingConflicts(self):
 214     """Test that we get a lock conflict for non-blocking locks."""
 215     name = 'ProcessLock'
 216     with flock.Lock(name, lock_dir=self.tempdir):
 217       self._HelperWithProcess(name,
 218                               expected=LOCK_NOT_ACQUIRED)
 219
 220       self._HelperWithProcess(name,
 221                               expected=LOCK_NOT_ACQUIRED,
 222                               shared=True)
 223
 224     # Can grab it after it's released
 225     self._HelperWithProcess(name, expected=LOCK_ACQUIRED)
 226
 227   @osutils.TempDirDecorator
 228   def testSharedLocks(self):
 229     """Test lock conflict for blocking locks."""
 230     name = 'ProcessLock'
 231
 232     # Intial lock is NOT shared
 233     with flock.Lock(name, lock_dir=self.tempdir, shared=False):
 234       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED, shared=True)
 235
 236     # Intial lock IS shared
 237     with flock.Lock(name, lock_dir=self.tempdir, shared=True):
 238       self._HelperWithProcess(name, expected=LOCK_ACQUIRED, shared=True)
 239       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED, shared=False)
 240
 241   @osutils.TempDirDecorator
 242   def testBlockingConflicts(self):
 243     """Test lock conflict for blocking locks."""
 244     name = 'ProcessLock'
 245
 246     # Intial lock is blocking, exclusive
 247     with flock.Lock(name, lock_dir=self.tempdir, blocking=True):
 248       self._HelperWithProcess(name,
 249                               expected=LOCK_NOT_ACQUIRED,
 250                               blocking=False)
 251
 252       p = self._HelperStartProcess(name, blocking=True, shared=False)
 253
 254     # when the with clause exits, p should unblock and get the lock, setting
 255     # its exit code to sucess now.
 256     p.join()
 257     self.assertEquals(p.exitcode, LOCK_ACQUIRED)
 258
 259     # Intial lock is NON blocking
 260     with flock.Lock(name, lock_dir=self.tempdir):
 261       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 262
 263       p = self._HelperStartProcess(name, blocking=True, shared=False)
 264
 265     # when the with clause exits, p should unblock and get the lock, setting
 266     # it's exit code to sucess now.
 267     p.join()
 268     self.assertEquals(p.exitcode, LOCK_ACQUIRED)
 269
 270     # Intial lock is shared, blocking lock is exclusive
 271     with flock.Lock(name, lock_dir=self.tempdir, shared=True):
 272       self._HelperWithProcess(name, expected=LOCK_NOT_ACQUIRED)
 273       self._HelperWithProcess(name, expected=LOCK_ACQUIRED, shared=True)
 274
 275       p = self._HelperStartProcess(name, blocking=True, shared=False)
 276       q = self._HelperStartProcess(name, blocking=True, shared=False)
 277
 278     # when the with clause exits, p should unblock and get the lock, setting
 279     # it's exit code to sucess now.
 280     p.join()
 281     self.assertEquals(p.exitcode, LOCK_ACQUIRED)
 282     q.join()
 283     self.assertEquals(p.exitcode, LOCK_ACQUIRED)
 284
 285
 286 if __name__ == '__main__':
 287   cros_test_lib.main()