• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4# Copyright 2019 The Chromium OS Authors. All rights reserved.
5# Use of this source code is governed by a BSD-style license that can be
6# found in the LICENSE file.
7
8"""lock_machine.py related unit-tests.
9
10MachineManagerTest tests MachineManager.
11"""
12
13from __future__ import print_function
14
15__author__ = 'asharif@google.com (Ahmad Sharif)'
16
17from multiprocessing import Process
18import time
19import unittest
20
21import file_lock_machine
22
23
24def LockAndSleep(machine):
25  file_lock_machine.Machine(machine, '/tmp', auto=True).Lock(exclusive=True)
26  time.sleep(1)
27
28
29class MachineTest(unittest.TestCase):
30  """Class for testing machine locking."""
31
32  def setUp(self):
33    pass
34
35  def testRepeatedUnlock(self):
36    mach = file_lock_machine.Machine('qqqraymes.mtv', '/tmp')
37    for _ in range(10):
38      self.assertTrue(mach.Unlock())
39    mach = file_lock_machine.Machine('qqqraymes.mtv', '/tmp', auto=True)
40    for _ in range(10):
41      self.assertTrue(mach.Unlock())
42
43  def testLockUnlock(self):
44    mach = file_lock_machine.Machine('otter.mtv', '/tmp')
45    for _ in range(10):
46      self.assertTrue(mach.Lock(exclusive=True))
47      self.assertTrue(mach.Unlock(exclusive=True))
48
49    mach = file_lock_machine.Machine('otter.mtv', '/tmp', True)
50    for _ in range(10):
51      self.assertTrue(mach.Lock(exclusive=True))
52      self.assertTrue(mach.Unlock(exclusive=True))
53
54  def testSharedLock(self):
55    mach = file_lock_machine.Machine('chrotomation.mtv', '/tmp')
56    for _ in range(10):
57      self.assertTrue(mach.Lock(exclusive=False))
58    for _ in range(10):
59      self.assertTrue(mach.Unlock(exclusive=False))
60    self.assertTrue(mach.Lock(exclusive=True))
61    self.assertTrue(mach.Unlock(exclusive=True))
62
63    mach = file_lock_machine.Machine('chrotomation.mtv', '/tmp', auto=True)
64    for _ in range(10):
65      self.assertTrue(mach.Lock(exclusive=False))
66    for _ in range(10):
67      self.assertTrue(mach.Unlock(exclusive=False))
68    self.assertTrue(mach.Lock(exclusive=True))
69    self.assertTrue(mach.Unlock(exclusive=True))
70
71  def testExclusiveLock(self):
72    mach = file_lock_machine.Machine('atree.mtv', '/tmp')
73    self.assertTrue(mach.Lock(exclusive=True))
74    for _ in range(10):
75      self.assertFalse(mach.Lock(exclusive=True))
76      self.assertFalse(mach.Lock(exclusive=False))
77    self.assertTrue(mach.Unlock(exclusive=True))
78
79    mach = file_lock_machine.Machine('atree.mtv', '/tmp', auto=True)
80    self.assertTrue(mach.Lock(exclusive=True))
81    for _ in range(10):
82      self.assertFalse(mach.Lock(exclusive=True))
83      self.assertFalse(mach.Lock(exclusive=False))
84    self.assertTrue(mach.Unlock(exclusive=True))
85
86  def testExclusiveState(self):
87    mach = file_lock_machine.Machine('testExclusiveState', '/tmp')
88    self.assertTrue(mach.Lock(exclusive=True))
89    for _ in range(10):
90      self.assertFalse(mach.Lock(exclusive=False))
91    self.assertTrue(mach.Unlock(exclusive=True))
92
93    mach = file_lock_machine.Machine('testExclusiveState', '/tmp', auto=True)
94    self.assertTrue(mach.Lock(exclusive=True))
95    for _ in range(10):
96      self.assertFalse(mach.Lock(exclusive=False))
97    self.assertTrue(mach.Unlock(exclusive=True))
98
99  def testAutoLockGone(self):
100    mach = file_lock_machine.Machine('lockgone', '/tmp', auto=True)
101    p = Process(target=LockAndSleep, args=('lockgone',))
102    p.start()
103    time.sleep(1.1)
104    p.join()
105    self.assertTrue(mach.Lock(exclusive=True))
106
107  def testAutoLockFromOther(self):
108    mach = file_lock_machine.Machine('other_lock', '/tmp', auto=True)
109    p = Process(target=LockAndSleep, args=('other_lock',))
110    p.start()
111    time.sleep(0.5)
112    self.assertFalse(mach.Lock(exclusive=True))
113    p.join()
114    time.sleep(0.6)
115    self.assertTrue(mach.Lock(exclusive=True))
116
117  def testUnlockByOthers(self):
118    mach = file_lock_machine.Machine('other_unlock', '/tmp', auto=True)
119    p = Process(target=LockAndSleep, args=('other_unlock',))
120    p.start()
121    time.sleep(0.5)
122    self.assertTrue(mach.Unlock(exclusive=True))
123    self.assertTrue(mach.Lock(exclusive=True))
124
125
126if __name__ == '__main__':
127  unittest.main()
128