aboutsummaryrefslogtreecommitdiff
path: root/file_lock_machine_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'file_lock_machine_test.py')
-rwxr-xr-xfile_lock_machine_test.py128
1 files changed, 0 insertions, 128 deletions
diff --git a/file_lock_machine_test.py b/file_lock_machine_test.py
deleted file mode 100755
index 467c183d..00000000
--- a/file_lock_machine_test.py
+++ /dev/null
@@ -1,128 +0,0 @@
-#!/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()