aboutsummaryrefslogtreecommitdiff
path: root/file_lock_machine_test.py
diff options
context:
space:
mode:
authorGeorge Burgess IV <gbiv@google.com>2022-09-02 16:59:27 -0700
committerChromeos LUCI <chromeos-scoped@luci-project-accounts.iam.gserviceaccount.com>2022-09-07 21:15:07 +0000
commit74bd380a27f4f0e8e90ff2dc1cef0b502d74961b (patch)
treebe028f89ec1e2eca735bb4aa1610530147a53625 /file_lock_machine_test.py
parent8448c60a6a2337ec993923837e1d55b41f49dabc (diff)
downloadtoolchain-utils-74bd380a27f4f0e8e90ff2dc1cef0b502d74961b.tar.gz
Autoformat all Python code
This autoformats all Python code with our new Python formatter, `black`. BUG=b:244644217 TEST=None Change-Id: I15ee49233d98fb6295c0c53c129bbf8e78e0d9ff Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/toolchain-utils/+/3877337 Tested-by: George Burgess <gbiv@chromium.org> Reviewed-by: Jordan Abrahams-Whitehead <ajordanr@google.com> Commit-Queue: George Burgess <gbiv@chromium.org>
Diffstat (limited to 'file_lock_machine_test.py')
-rwxr-xr-xfile_lock_machine_test.py205
1 files changed, 104 insertions, 101 deletions
diff --git a/file_lock_machine_test.py b/file_lock_machine_test.py
index d14deaf4..d1189512 100755
--- a/file_lock_machine_test.py
+++ b/file_lock_machine_test.py
@@ -12,7 +12,8 @@ MachineManagerTest tests MachineManager.
from __future__ import print_function
-__author__ = 'asharif@google.com (Ahmad Sharif)'
+
+__author__ = "asharif@google.com (Ahmad Sharif)"
from multiprocessing import Process
import time
@@ -22,106 +23,108 @@ import file_lock_machine
def LockAndSleep(machine):
- file_lock_machine.Machine(machine, '/tmp', auto=True).Lock(exclusive=True)
- time.sleep(1)
+ file_lock_machine.Machine(machine, "/tmp", auto=True).Lock(exclusive=True)
+ time.sleep(1)
class MachineTest(unittest.TestCase):
- """Class for testing machine locking."""
-
- def setUp(self):
- pass
-
- def testRepeatedUnlock(self):
- mach = file_lock_machine.Machine('qqqraymes.mtv', '/tmp')
- for _ in range(10):
- self.assertTrue(mach.Unlock())
- mach = file_lock_machine.Machine('qqqraymes.mtv', '/tmp', auto=True)
- for _ in range(10):
- self.assertTrue(mach.Unlock())
-
- def testLockUnlock(self):
- mach = file_lock_machine.Machine('otter.mtv', '/tmp')
- for _ in range(10):
- self.assertTrue(mach.Lock(exclusive=True))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- mach = file_lock_machine.Machine('otter.mtv', '/tmp', True)
- for _ in range(10):
- self.assertTrue(mach.Lock(exclusive=True))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- def testSharedLock(self):
- mach = file_lock_machine.Machine('chrotomation.mtv', '/tmp')
- for _ in range(10):
- self.assertTrue(mach.Lock(exclusive=False))
- for _ in range(10):
- self.assertTrue(mach.Unlock(exclusive=False))
- self.assertTrue(mach.Lock(exclusive=True))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- mach = file_lock_machine.Machine('chrotomation.mtv', '/tmp', auto=True)
- for _ in range(10):
- self.assertTrue(mach.Lock(exclusive=False))
- for _ 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 = file_lock_machine.Machine('atree.mtv', '/tmp')
- self.assertTrue(mach.Lock(exclusive=True))
- for _ in range(10):
- self.assertFalse(mach.Lock(exclusive=True))
- self.assertFalse(mach.Lock(exclusive=False))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- mach = file_lock_machine.Machine('atree.mtv', '/tmp', auto=True)
- self.assertTrue(mach.Lock(exclusive=True))
- for _ 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 = file_lock_machine.Machine('testExclusiveState', '/tmp')
- self.assertTrue(mach.Lock(exclusive=True))
- for _ in range(10):
- self.assertFalse(mach.Lock(exclusive=False))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- mach = file_lock_machine.Machine('testExclusiveState', '/tmp', auto=True)
- self.assertTrue(mach.Lock(exclusive=True))
- for _ in range(10):
- self.assertFalse(mach.Lock(exclusive=False))
- self.assertTrue(mach.Unlock(exclusive=True))
-
- def testAutoLockGone(self):
- mach = file_lock_machine.Machine('lockgone', '/tmp', 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 = file_lock_machine.Machine('other_lock', '/tmp', 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 = file_lock_machine.Machine('other_unlock', '/tmp', 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()
+ """Class for testing machine locking."""
+
+ def setUp(self):
+ pass
+
+ def testRepeatedUnlock(self):
+ mach = file_lock_machine.Machine("qqqraymes.mtv", "/tmp")
+ for _ in range(10):
+ self.assertTrue(mach.Unlock())
+ mach = file_lock_machine.Machine("qqqraymes.mtv", "/tmp", auto=True)
+ for _ in range(10):
+ self.assertTrue(mach.Unlock())
+
+ def testLockUnlock(self):
+ mach = file_lock_machine.Machine("otter.mtv", "/tmp")
+ for _ in range(10):
+ self.assertTrue(mach.Lock(exclusive=True))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ mach = file_lock_machine.Machine("otter.mtv", "/tmp", True)
+ for _ in range(10):
+ self.assertTrue(mach.Lock(exclusive=True))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ def testSharedLock(self):
+ mach = file_lock_machine.Machine("chrotomation.mtv", "/tmp")
+ for _ in range(10):
+ self.assertTrue(mach.Lock(exclusive=False))
+ for _ in range(10):
+ self.assertTrue(mach.Unlock(exclusive=False))
+ self.assertTrue(mach.Lock(exclusive=True))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ mach = file_lock_machine.Machine("chrotomation.mtv", "/tmp", auto=True)
+ for _ in range(10):
+ self.assertTrue(mach.Lock(exclusive=False))
+ for _ 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 = file_lock_machine.Machine("atree.mtv", "/tmp")
+ self.assertTrue(mach.Lock(exclusive=True))
+ for _ in range(10):
+ self.assertFalse(mach.Lock(exclusive=True))
+ self.assertFalse(mach.Lock(exclusive=False))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ mach = file_lock_machine.Machine("atree.mtv", "/tmp", auto=True)
+ self.assertTrue(mach.Lock(exclusive=True))
+ for _ 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 = file_lock_machine.Machine("testExclusiveState", "/tmp")
+ self.assertTrue(mach.Lock(exclusive=True))
+ for _ in range(10):
+ self.assertFalse(mach.Lock(exclusive=False))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ mach = file_lock_machine.Machine(
+ "testExclusiveState", "/tmp", auto=True
+ )
+ self.assertTrue(mach.Lock(exclusive=True))
+ for _ in range(10):
+ self.assertFalse(mach.Lock(exclusive=False))
+ self.assertTrue(mach.Unlock(exclusive=True))
+
+ def testAutoLockGone(self):
+ mach = file_lock_machine.Machine("lockgone", "/tmp", 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 = file_lock_machine.Machine("other_lock", "/tmp", 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 = file_lock_machine.Machine("other_unlock", "/tmp", 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()