aboutsummaryrefslogtreecommitdiff
path: root/file_lock_machine_test.py
diff options
context:
space:
mode:
authorAndroid Build Coastguard Worker <android-build-coastguard-worker@google.com>2022-11-13 02:20:05 +0000
committerAndroid Build Coastguard Worker <android-build-coastguard-worker@google.com>2022-11-13 02:20:05 +0000
commitcec5bc676dfc0ade1b32f3873e93a3bd2c30e390 (patch)
tree77dc031614745bb406dbd90cea9a082a1b5cdd54 /file_lock_machine_test.py
parent773bb66b7b5fab0b5f14d66455274040063e14bf (diff)
parent517424dcc11380511bc34f4a081f119104ff9e80 (diff)
downloadtoolchain-utils-cec5bc676dfc0ade1b32f3873e93a3bd2c30e390.tar.gz
Change-Id: I0dfd9fe9a88cd3daf076b0fb77b69e3138c08e6c
Diffstat (limited to 'file_lock_machine_test.py')
-rwxr-xr-xfile_lock_machine_test.py207
1 files changed, 104 insertions, 103 deletions
diff --git a/file_lock_machine_test.py b/file_lock_machine_test.py
index bc20a88b..467c183d 100755
--- a/file_lock_machine_test.py
+++ b/file_lock_machine_test.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
-# Copyright 2019 The Chromium OS Authors. All rights reserved.
+# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
@@ -10,9 +10,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 +21,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()