diff options
author | Android Build Coastguard Worker <android-build-coastguard-worker@google.com> | 2022-11-13 02:20:05 +0000 |
---|---|---|
committer | Android Build Coastguard Worker <android-build-coastguard-worker@google.com> | 2022-11-13 02:20:05 +0000 |
commit | cec5bc676dfc0ade1b32f3873e93a3bd2c30e390 (patch) | |
tree | 77dc031614745bb406dbd90cea9a082a1b5cdd54 /file_lock_machine_test.py | |
parent | 773bb66b7b5fab0b5f14d66455274040063e14bf (diff) | |
parent | 517424dcc11380511bc34f4a081f119104ff9e80 (diff) | |
download | toolchain-utils-cec5bc676dfc0ade1b32f3873e93a3bd2c30e390.tar.gz |
Snap for 9286410 from 517424dcc11380511bc34f4a081f119104ff9e80 to udc-d1-releaseandroid-14.0.0_r9android-14.0.0_r8android-14.0.0_r7android-14.0.0_r6android-14.0.0_r5android-14.0.0_r4android-14.0.0_r3android-14.0.0_r12android-14.0.0_r11android-14.0.0_r10android14-d1-s7-releaseandroid14-d1-s6-releaseandroid14-d1-s5-releaseandroid14-d1-s4-releaseandroid14-d1-s3-releaseandroid14-d1-s2-releaseandroid14-d1-s1-releaseandroid14-d1-release
Change-Id: I0dfd9fe9a88cd3daf076b0fb77b69e3138c08e6c
Diffstat (limited to 'file_lock_machine_test.py')
-rwxr-xr-x | file_lock_machine_test.py | 207 |
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() |