hmz007
36ed224bac
Signed-off-by: hmz007 <hmz007@gmail.com> Change-Id: Ib87fdbe7a0be7dc961af3500fe9ea4589a127f9f
129 lines
4.3 KiB
Python
Executable File
129 lines
4.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# 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.
|
|
|
|
"""lock_machine.py related unit-tests.
|
|
|
|
MachineManagerTest tests MachineManager.
|
|
"""
|
|
|
|
|
|
__author__ = "asharif@google.com (Ahmad Sharif)"
|
|
|
|
from multiprocessing import Process
|
|
import time
|
|
import unittest
|
|
|
|
import file_lock_machine
|
|
|
|
|
|
def LockAndSleep(machine):
|
|
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()
|