test-lock.py: fix testing for forks

The earlier test worked only because the held count went up to 2, so the first
release brought it down to 1. Making a copy of the lock fixes that issue.

tests/test-lock.py

@@ -1,147 +1,155 b''
 from __future__ import absolute_import
 
+import copy
 import os
 import silenttestrunner
 import tempfile
+import types
 import unittest
 
 from mercurial import (
     lock,
     scmutil,
 )
 
 testlockname = 'testlock'
 
+# work around http://bugs.python.org/issue1515
+if types.MethodType not in copy._deepcopy_dispatch:
+    def _deepcopy_method(x, memo):
+        return type(x)(x.im_func, copy.deepcopy(x.im_self, memo), x.im_class)
+    copy._deepcopy_dispatch[types.MethodType] = _deepcopy_method
+
 class lockwrapper(lock.lock):
     def __init__(self, pidoffset, *args, **kwargs):
         # lock.lock.__init__() calls lock(), so the pidoffset assignment needs
         # to be earlier
         self._pidoffset = pidoffset
         super(lockwrapper, self).__init__(*args, **kwargs)
     def _getpid(self):
         return os.getpid() + self._pidoffset
 
 class teststate(object):
     def __init__(self, testcase, dir, pidoffset=0):
         self._testcase = testcase
         self._acquirecalled = False
         self._releasecalled = False
         self._postreleasecalled = False
         self.vfs = scmutil.vfs(dir, audit=False)
         self._pidoffset = pidoffset
 
     def makelock(self, *args, **kwargs):
         l = lockwrapper(self._pidoffset, self.vfs, testlockname,
                         releasefn=self.releasefn, acquirefn=self.acquirefn,
                         *args, **kwargs)
         l.postrelease.append(self.postreleasefn)
         return l
 
     def acquirefn(self):
         self._acquirecalled = True
 
     def releasefn(self):
         self._releasecalled = True
 
     def postreleasefn(self):
         self._postreleasecalled = True
 
     def assertacquirecalled(self, called):
         self._testcase.assertEqual(
             self._acquirecalled, called,
             'expected acquire to be %s but was actually %s' % (
                 self._tocalled(called),
                 self._tocalled(self._acquirecalled),
             ))
 
     def resetacquirefn(self):
         self._acquirecalled = False
 
     def assertreleasecalled(self, called):
         self._testcase.assertEqual(
             self._releasecalled, called,
             'expected release to be %s but was actually %s' % (
                 self._tocalled(called),
                 self._tocalled(self._releasecalled),
             ))
 
     def assertpostreleasecalled(self, called):
         self._testcase.assertEqual(
             self._postreleasecalled, called,
             'expected postrelease to be %s but was actually %s' % (
                 self._tocalled(called),
                 self._tocalled(self._postreleasecalled),
             ))
 
     def assertlockexists(self, exists):
         actual = self.vfs.lexists(testlockname)
         self._testcase.assertEqual(
             actual, exists,
             'expected lock to %s but actually did %s' % (
                 self._toexists(exists),
                 self._toexists(actual),
             ))
 
     def _tocalled(self, called):
         if called:
             return 'called'
         else:
             return 'not called'
 
     def _toexists(self, exists):
         if exists:
             return 'exist'
         else:
             return 'not exist'
 
 class testlock(unittest.TestCase):
     def testlock(self):
         state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
         lock = state.makelock()
         state.assertacquirecalled(True)
         lock.release()
         state.assertreleasecalled(True)
         state.assertpostreleasecalled(True)
         state.assertlockexists(False)
 
     def testrecursivelock(self):
         state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
         lock = state.makelock()
         state.assertacquirecalled(True)
 
         state.resetacquirefn()
         lock.lock()
         # recursive lock should not call acquirefn again
         state.assertacquirecalled(False)
 
         lock.release() # brings lock refcount down from 2 to 1
         state.assertreleasecalled(False)
         state.assertpostreleasecalled(False)
         state.assertlockexists(True)
 
         lock.release() # releases the lock
         state.assertreleasecalled(True)
         state.assertpostreleasecalled(True)
         state.assertlockexists(False)
 
     def testlockfork(self):
         state = teststate(self, tempfile.mkdtemp(dir=os.getcwd()))
         lock = state.makelock()
         state.assertacquirecalled(True)
-        lock.lock()
+
         # fake a fork
-        lock.pid += 1
-        lock.release()
+        forklock = copy.deepcopy(lock)
+        forklock._pidoffset = 1
+        forklock.release()
         state.assertreleasecalled(False)
         state.assertpostreleasecalled(False)
         state.assertlockexists(True)
 
         # release the actual lock
-        lock.pid -= 1
         lock.release()
         state.assertreleasecalled(True)
         state.assertpostreleasecalled(True)
         state.assertlockexists(False)
 
 if __name__ == '__main__':
     silenttestrunner.main(__name__)