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