Home | History | Annotate | Download | only in toolchain-utils 
      1 # Copyright 2010 Google Inc. All Rights Reserved.
      2 """lock_machine.py related unit-tests.
      3 
      4 MachineManagerTest tests MachineManager.
      5 """
      6 
      7 from __future__ import print_function
      8 
      9 __author__ = 'asharif (at] google.com (Ahmad Sharif)'
     10 
     11 from multiprocessing import Process
     12 import time
     13 import unittest
     14 
     15 import file_lock_machine
     16 
     17 
     18 def LockAndSleep(machine):
     19   file_lock_machine.Machine(machine, auto=True).Lock(exclusive=True)
     20   time.sleep(1)
     21 
     22 
     23 class MachineTest(unittest.TestCase):
     24   """Class for testing machine locking."""
     25 
     26   def setUp(self):
     27     pass
     28 
     29   def testRepeatedUnlock(self):
     30     mach = file_lock_machine.Machine('qqqraymes.mtv')
     31     for _ in range(10):
     32       self.assertFalse(mach.Unlock())
     33     mach = file_lock_machine.Machine('qqqraymes.mtv', auto=True)
     34     for _ in range(10):
     35       self.assertFalse(mach.Unlock())
     36 
     37   def testLockUnlock(self):
     38     mach = file_lock_machine.Machine('otter.mtv', '/tmp')
     39     for _ in range(10):
     40       self.assertTrue(mach.Lock(exclusive=True))
     41       self.assertTrue(mach.Unlock(exclusive=True))
     42 
     43     mach = file_lock_machine.Machine('otter.mtv', '/tmp', True)
     44     for _ in range(10):
     45       self.assertTrue(mach.Lock(exclusive=True))
     46       self.assertTrue(mach.Unlock(exclusive=True))
     47 
     48   def testSharedLock(self):
     49     mach = file_lock_machine.Machine('chrotomation.mtv')
     50     for _ in range(10):
     51       self.assertTrue(mach.Lock(exclusive=False))
     52     for _ in range(10):
     53       self.assertTrue(mach.Unlock(exclusive=False))
     54     self.assertTrue(mach.Lock(exclusive=True))
     55     self.assertTrue(mach.Unlock(exclusive=True))
     56 
     57     mach = file_lock_machine.Machine('chrotomation.mtv', auto=True)
     58     for _ in range(10):
     59       self.assertTrue(mach.Lock(exclusive=False))
     60     for _ in range(10):
     61       self.assertTrue(mach.Unlock(exclusive=False))
     62     self.assertTrue(mach.Lock(exclusive=True))
     63     self.assertTrue(mach.Unlock(exclusive=True))
     64 
     65   def testExclusiveLock(self):
     66     mach = file_lock_machine.Machine('atree.mtv')
     67     self.assertTrue(mach.Lock(exclusive=True))
     68     for _ in range(10):
     69       self.assertFalse(mach.Lock(exclusive=True))
     70       self.assertFalse(mach.Lock(exclusive=False))
     71     self.assertTrue(mach.Unlock(exclusive=True))
     72 
     73     mach = file_lock_machine.Machine('atree.mtv', auto=True)
     74     self.assertTrue(mach.Lock(exclusive=True))
     75     for _ in range(10):
     76       self.assertFalse(mach.Lock(exclusive=True))
     77       self.assertFalse(mach.Lock(exclusive=False))
     78     self.assertTrue(mach.Unlock(exclusive=True))
     79 
     80   def testExclusiveState(self):
     81     mach = file_lock_machine.Machine('testExclusiveState')
     82     self.assertTrue(mach.Lock(exclusive=True))
     83     for _ in range(10):
     84       self.assertFalse(mach.Lock(exclusive=False))
     85     self.assertTrue(mach.Unlock(exclusive=True))
     86 
     87     mach = file_lock_machine.Machine('testExclusiveState', auto=True)
     88     self.assertTrue(mach.Lock(exclusive=True))
     89     for _ in range(10):
     90       self.assertFalse(mach.Lock(exclusive=False))
     91     self.assertTrue(mach.Unlock(exclusive=True))
     92 
     93   def testAutoLockGone(self):
     94     mach = file_lock_machine.Machine('lockgone', auto=True)
     95     p = Process(target=LockAndSleep, args=('lockgone',))
     96     p.start()
     97     time.sleep(1.1)
     98     p.join()
     99     self.assertTrue(mach.Lock(exclusive=True))
    100 
    101   def testAutoLockFromOther(self):
    102     mach = file_lock_machine.Machine('other_lock', auto=True)
    103     p = Process(target=LockAndSleep, args=('other_lock',))
    104     p.start()
    105     time.sleep(0.5)
    106     self.assertFalse(mach.Lock(exclusive=True))
    107     p.join()
    108     time.sleep(0.6)
    109     self.assertTrue(mach.Lock(exclusive=True))
    110 
    111   def testUnlockByOthers(self):
    112     mach = file_lock_machine.Machine('other_unlock', auto=True)
    113     p = Process(target=LockAndSleep, args=('other_unlock',))
    114     p.start()
    115     time.sleep(0.5)
    116     self.assertTrue(mach.Unlock(exclusive=True))
    117     self.assertTrue(mach.Lock(exclusive=True))
    118 
    119 
    120 if __name__ == '__main__':
    121   unittest.main()
    122