• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2010 Google Inc. All Rights Reserved.
2"""lock_machine.py related unit-tests.
3
4MachineManagerTest tests MachineManager.
5"""
6
7from __future__ import print_function
8
9__author__ = 'asharif@google.com (Ahmad Sharif)'
10
11from multiprocessing import Process
12import time
13import unittest
14
15import file_lock_machine
16
17
18def LockAndSleep(machine):
19  file_lock_machine.Machine(machine, auto=True).Lock(exclusive=True)
20  time.sleep(1)
21
22
23class 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
120if __name__ == '__main__':
121  unittest.main()
122