Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.locking_unittest.py @ 162c1c1f

History | View | Annotate | Download (5.1 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2006, 2007 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 0.0510-1301, USA.
20

    
21

    
22
"""Script for unittesting the locking module"""
23

    
24

    
25
import os
26
import unittest
27
import time
28
import Queue
29

    
30
from ganeti import locking
31
from threading import Thread
32

    
33

    
34
class TestSharedLock(unittest.TestCase):
35
  """Shared lock tests"""
36

    
37
  def setUp(self):
38
    self.sl = locking.SharedLock()
39
    # helper threads use the 'done' queue to tell the master they finished.
40
    self.done = Queue.Queue(0)
41

    
42
  def testSequenceAndOwnership(self):
43
    self.assert_(not self.sl._is_owned())
44
    self.sl.acquire(shared=1)
45
    self.assert_(self.sl._is_owned())
46
    self.assert_(self.sl._is_owned(shared=1))
47
    self.assert_(not self.sl._is_owned(shared=0))
48
    self.sl.release()
49
    self.assert_(not self.sl._is_owned())
50
    self.sl.acquire()
51
    self.assert_(self.sl._is_owned())
52
    self.assert_(not self.sl._is_owned(shared=1))
53
    self.assert_(self.sl._is_owned(shared=0))
54
    self.sl.release()
55
    self.assert_(not self.sl._is_owned())
56
    self.sl.acquire(shared=1)
57
    self.assert_(self.sl._is_owned())
58
    self.assert_(self.sl._is_owned(shared=1))
59
    self.assert_(not self.sl._is_owned(shared=0))
60
    self.sl.release()
61
    self.assert_(not self.sl._is_owned())
62

    
63
  def testBooleanValue(self):
64
    # semaphores are supposed to return a true value on a successful acquire
65
    self.assert_(self.sl.acquire(shared=1))
66
    self.sl.release()
67
    self.assert_(self.sl.acquire())
68
    self.sl.release()
69

    
70
  def testDoubleLockingStoE(self):
71
    self.sl.acquire(shared=1)
72
    self.assertRaises(AssertionError, self.sl.acquire)
73

    
74
  def testDoubleLockingEtoS(self):
75
    self.sl.acquire()
76
    self.assertRaises(AssertionError, self.sl.acquire, shared=1)
77

    
78
  def testDoubleLockingStoS(self):
79
    self.sl.acquire(shared=1)
80
    self.assertRaises(AssertionError, self.sl.acquire, shared=1)
81

    
82
  def testDoubleLockingEtoE(self):
83
    self.sl.acquire()
84
    self.assertRaises(AssertionError, self.sl.acquire)
85

    
86
  # helper functions: called in a separate thread they acquire the lock, send
87
  # their identifier on the done queue, then release it.
88
  def _doItSharer(self):
89
    self.sl.acquire(shared=1)
90
    self.done.put('SHR')
91
    self.sl.release()
92

    
93
  def _doItExclusive(self):
94
    self.sl.acquire()
95
    self.done.put('EXC')
96
    self.sl.release()
97

    
98
  def testSharersCanCoexist(self):
99
    self.sl.acquire(shared=1)
100
    Thread(target=self._doItSharer).start()
101
    self.assert_(self.done.get(True, 1))
102
    self.sl.release()
103

    
104
  def testExclusiveBlocksExclusive(self):
105
    self.sl.acquire()
106
    Thread(target=self._doItExclusive).start()
107
    # give it a bit of time to check that it's not actually doing anything
108
    self.assertRaises(Queue.Empty, self.done.get, True, 0.2)
109
    self.sl.release()
110
    self.assert_(self.done.get(True, 1))
111

    
112
  def testExclusiveBlocksSharer(self):
113
    self.sl.acquire()
114
    Thread(target=self._doItSharer).start()
115
    time.sleep(0.05)
116
    self.assertRaises(Queue.Empty, self.done.get, True, 0.2)
117
    self.sl.release()
118
    self.assert_(self.done.get(True, 1))
119

    
120
  def testSharerBlocksExclusive(self):
121
    self.sl.acquire(shared=1)
122
    Thread(target=self._doItExclusive).start()
123
    time.sleep(0.05)
124
    self.assertRaises(Queue.Empty, self.done.get, True, 0.2)
125
    self.sl.release()
126
    self.assert_(self.done.get(True, 1))
127

    
128
  def testWaitingExclusiveBlocksSharer(self):
129
    self.sl.acquire(shared=1)
130
    # the lock is acquired in shared mode...
131
    Thread(target=self._doItExclusive).start()
132
    # ...but now an exclusive is waiting...
133
    time.sleep(0.05)
134
    Thread(target=self._doItSharer).start()
135
    # ...so the sharer should be blocked as well
136
    self.assertRaises(Queue.Empty, self.done.get, True, 0.2)
137
    self.sl.release()
138
    # The exclusive passed before
139
    self.assertEqual(self.done.get(True, 1), 'EXC')
140
    self.assertEqual(self.done.get(True, 1), 'SHR')
141

    
142
  def testWaitingSharerBlocksExclusive(self):
143
    self.sl.acquire()
144
    # the lock is acquired in exclusive mode...
145
    Thread(target=self._doItSharer).start()
146
    # ...but now a sharer is waiting...
147
    time.sleep(0.05)
148
    Thread(target=self._doItExclusive).start()
149
    # ...the exclusive is waiting too...
150
    self.assertRaises(Queue.Empty, self.done.get, True, 0.2)
151
    self.sl.release()
152
    # The sharer passed before
153
    self.assertEqual(self.done.get(True, 1), 'SHR')
154
    self.assertEqual(self.done.get(True, 1), 'EXC')
155

    
156

    
157
if __name__ == '__main__':
158
  unittest.main()
159
  #suite = unittest.TestLoader().loadTestsFromTestCase(TestSharedLock)
160
  #unittest.TextTestRunner(verbosity=2).run(suite)