1 files changed, 120 insertions, 0 deletions
diff --git a/lock_machine_test.py b/lock_machine_test.py
new file mode 100644
index 00000000..d61878b8
--- /dev/null
+++ b/lock_machine_test.py
@@ -0,0 +1,120 @@
+#!/usr/bin/python
+#
+# Copyright 2010 Google Inc. All Rights Reserved.
+
+"""lock_machine.py related unit-tests.
+
+MachineManagerTest tests MachineManager.
+"""
+
+__author__ = "asharif@google.com (Ahmad Sharif)"
+
+from multiprocessing import Process
+import time
+import unittest
+
+import lock_machine
+
+
+def LockAndSleep(machine):
+  lock_machine.Machine(machine, auto=True).Lock(exclusive=True)
+  time.sleep(1)
+
+
+class MachineTest(unittest.TestCase):
+  def setUp(self):
+    pass
+
+  def testRepeatedUnlock(self):
+    mach = lock_machine.Machine("qqqraymes.mtv")
+    for i in range(10):
+      self.assertFalse(mach.Unlock())
+    mach = lock_machine.Machine("qqqraymes.mtv", auto=True)
+    for i in range(10):
+      self.assertFalse(mach.Unlock())
+
+  def testLockUnlock(self):
+    mach = lock_machine.Machine("otter.mtv", "/tmp")
+    for i in range(10):
+      self.assertTrue(mach.Lock(exclusive=True))
+      self.assertTrue(mach.Unlock(exclusive=True))
+
+    mach = lock_machine.Machine("otter.mtv", "/tmp", True)
+    for i in range(10):
+      self.assertTrue(mach.Lock(exclusive=True))
+      self.assertTrue(mach.Unlock(exclusive=True))
+
+  def testSharedLock(self):
+    mach = lock_machine.Machine("chrotomation.mtv")
+    for i in range(10):
+      self.assertTrue(mach.Lock(exclusive=False))
+    for i in range(10):
+      self.assertTrue(mach.Unlock(exclusive=False))
+    self.assertTrue(mach.Lock(exclusive=True))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+    mach = lock_machine.Machine("chrotomation.mtv", auto=True)
+    for i in range(10):
+      self.assertTrue(mach.Lock(exclusive=False))
+    for i in range(10):
+      self.assertTrue(mach.Unlock(exclusive=False))
+    self.assertTrue(mach.Lock(exclusive=True))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+  def testExclusiveLock(self):
+    mach = lock_machine.Machine("atree.mtv")
+    self.assertTrue(mach.Lock(exclusive=True))
+    for i in range(10):
+      self.assertFalse(mach.Lock(exclusive=True))
+      self.assertFalse(mach.Lock(exclusive=False))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+    mach = lock_machine.Machine("atree.mtv", auto=True)
+    self.assertTrue(mach.Lock(exclusive=True))
+    for i in range(10):
+      self.assertFalse(mach.Lock(exclusive=True))
+      self.assertFalse(mach.Lock(exclusive=False))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+  def testExclusiveState(self):
+    mach = lock_machine.Machine("testExclusiveState")
+    self.assertTrue(mach.Lock(exclusive=True))
+    for i in range(10):
+      self.assertFalse(mach.Lock(exclusive=False))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+    mach = lock_machine.Machine("testExclusiveState", auto=True)
+    self.assertTrue(mach.Lock(exclusive=True))
+    for i in range(10):
+      self.assertFalse(mach.Lock(exclusive=False))
+    self.assertTrue(mach.Unlock(exclusive=True))
+
+  def testAutoLockGone(self):
+    mach = lock_machine.Machine("lockgone", auto=True)
+    p = Process(target=LockAndSleep, args=("lockgone",))
+    p.start()
+    time.sleep(1.1)
+    p.join()
+    self.assertTrue(mach.Lock(exclusive=True))
+
+  def testAutoLockFromOther(self):
+    mach = lock_machine.Machine("other_lock", auto=True)
+    p = Process(target=LockAndSleep, args=("other_lock",))
+    p.start()
+    time.sleep(0.5)
+    self.assertFalse(mach.Lock(exclusive=True))
+    p.join()
+    time.sleep(0.6)
+    self.assertTrue(mach.Lock(exclusive=True))
+
+  def testUnlockByOthers(self):
+    mach = lock_machine.Machine("other_unlock", auto=True)
+    p = Process(target=LockAndSleep, args=("other_unlock",))
+    p.start()
+    time.sleep(0.5)
+    self.assertTrue(mach.Unlock(exclusive=True))
+    self.assertTrue(mach.Lock(exclusive=True))
+
+
+if __name__ == "__main__":
+  unittest.main()