##// END OF EJS Templates
test-lock.py: allow PID to be changed in test state...
Siddharth Agarwal -
r26385:fb1a424e default
parent child Browse files
Show More
@@ -1,145 +1,147 b''
1 1 from __future__ import absolute_import
2 2
3 3 import os
4 4 import silenttestrunner
5 5 import tempfile
6 6 import unittest
7 7
8 8 from mercurial import (
9 9 lock,
10 10 scmutil,
11 11 )
12 12
13 13 testlockname = 'testlock'
14 14
15 15 class lockwrapper(lock.lock):
16 16 def __init__(self, pidoffset, *args, **kwargs):
17 17 # lock.lock.__init__() calls lock(), so the pidoffset assignment needs
18 18 # to be earlier
19 19 self._pidoffset = pidoffset
20 20 super(lockwrapper, self).__init__(*args, **kwargs)
21 21 def _getpid(self):
22 22 return os.getpid() + self._pidoffset
23 23
24 24 class teststate(object):
25 def __init__(self, testcase, dir):
25 def __init__(self, testcase, dir, pidoffset=0):
26 26 self._testcase = testcase
27 27 self._acquirecalled = False
28 28 self._releasecalled = False
29 29 self._postreleasecalled = False
30 30 self.vfs = scmutil.vfs(dir, audit=False)
31 self._pidoffset = pidoffset
31 32
32 33 def makelock(self, *args, **kwargs):
33 l = lock.lock(self.vfs, testlockname, releasefn=self.releasefn,
34 acquirefn=self.acquirefn, *args, **kwargs)
34 l = lockwrapper(self._pidoffset, self.vfs, testlockname,
35 releasefn=self.releasefn, acquirefn=self.acquirefn,
36 *args, **kwargs)
35 37 l.postrelease.append(self.postreleasefn)
36 38 return l
37 39
38 40 def acquirefn(self):
39 41 self._acquirecalled = True
40 42
41 43 def releasefn(self):
42 44 self._releasecalled = True
43 45
44 46 def postreleasefn(self):
45 47 self._postreleasecalled = True
46 48
47 49 def assertacquirecalled(self, called):
48 50 self._testcase.assertEqual(
49 51 self._acquirecalled, called,
50 52 'expected acquire to be %s but was actually %s' % (
51 53 self._tocalled(called),
52 54 self._tocalled(self._acquirecalled),
53 55 ))
54 56
55 57 def resetacquirefn(self):
56 58 self._acquirecalled = False
57 59
58 60 def assertreleasecalled(self, called):
59 61 self._testcase.assertEqual(
60 62 self._releasecalled, called,
61 63 'expected release to be %s but was actually %s' % (
62 64 self._tocalled(called),
63 65 self._tocalled(self._releasecalled),
64 66 ))
65 67
66 68 def assertpostreleasecalled(self, called):
67 69 self._testcase.assertEqual(
68 70 self._postreleasecalled, called,
69 71 'expected postrelease to be %s but was actually %s' % (
70 72 self._tocalled(called),
71 73 self._tocalled(self._postreleasecalled),
72 74 ))
73 75
74 76 def assertlockexists(self, exists):
75 77 actual = self.vfs.lexists(testlockname)
76 78 self._testcase.assertEqual(
77 79 actual, exists,
78 80 'expected lock to %s but actually did %s' % (
79 81 self._toexists(exists),
80 82 self._toexists(actual),
81 83 ))
82 84
83 85 def _tocalled(self, called):
84 86 if called:
85 87 return 'called'
86 88 else:
87 89 return 'not called'
88 90
89 91 def _toexists(self, exists):
90 92 if exists:
91 93 return 'exist'
92 94 else:
93 95 return 'not exist'
94 96
95 97 class testlock(unittest.TestCase):
96 98 def testlock(self):
97 99 state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
98 100 lock = state.makelock()
99 101 state.assertacquirecalled(True)
100 102 lock.release()
101 103 state.assertreleasecalled(True)
102 104 state.assertpostreleasecalled(True)
103 105 state.assertlockexists(False)
104 106
105 107 def testrecursivelock(self):
106 108 state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
107 109 lock = state.makelock()
108 110 state.assertacquirecalled(True)
109 111
110 112 state.resetacquirefn()
111 113 lock.lock()
112 114 # recursive lock should not call acquirefn again
113 115 state.assertacquirecalled(False)
114 116
115 117 lock.release() # brings lock refcount down from 2 to 1
116 118 state.assertreleasecalled(False)
117 119 state.assertpostreleasecalled(False)
118 120 state.assertlockexists(True)
119 121
120 122 lock.release() # releases the lock
121 123 state.assertreleasecalled(True)
122 124 state.assertpostreleasecalled(True)
123 125 state.assertlockexists(False)
124 126
125 127 def testlockfork(self):
126 128 state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
127 129 lock = state.makelock()
128 130 state.assertacquirecalled(True)
129 131 lock.lock()
130 132 # fake a fork
131 133 lock.pid += 1
132 134 lock.release()
133 135 state.assertreleasecalled(False)
134 136 state.assertpostreleasecalled(False)
135 137 state.assertlockexists(True)
136 138
137 139 # release the actual lock
138 140 lock.pid -= 1
139 141 lock.release()
140 142 state.assertreleasecalled(True)
141 143 state.assertpostreleasecalled(True)
142 144 state.assertlockexists(False)
143 145
144 146 if __name__ == '__main__':
145 147 silenttestrunner.main(__name__)
General Comments 0
You need to be logged in to leave comments. Login now