Mercurial > hg
view tests/test-lock.py @ 26351:8c7d8d5e1e0f
mercurial: add debugextensions command (issue4676)
Add debugextensions command to help users debug their extension
problems. If there are no extensions command prints nothing,
otherwise it prints names of extension modules. If quiet or
verbose option is not specified it prints(after extensions name)
last version of mercurial in which given module was tested for
non internal modules or not tested with user mercurial version.
If verbose is specified it prints following information for every
extension: extension name, import source, testedwith and buglink
information.
Extensions are printed sorted by extension name.
author | liscju <piotr.listkiewicz@gmail.com> |
---|---|
date | Thu, 10 Sep 2015 16:53:07 +0200 |
parents | db4c192cb9b3 |
children | 94dc10834b79 |
line wrap: on
line source
from __future__ import absolute_import import os import silenttestrunner import tempfile import unittest from mercurial import ( lock, scmutil, ) testlockname = 'testlock' class teststate(object): def __init__(self, testcase): self._testcase = testcase self._acquirecalled = False self._releasecalled = False self._postreleasecalled = False d = tempfile.mkdtemp(dir=os.getcwd()) self.vfs = scmutil.vfs(d, audit=False) def makelock(self, *args, **kwargs): l = lock.lock(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 'exists' else: return 'not exists' class testlock(unittest.TestCase): def testlock(self): state = teststate(self) lock = state.makelock() state.assertacquirecalled(True) lock.release() state.assertreleasecalled(True) state.assertpostreleasecalled(True) state.assertlockexists(False) def testrecursivelock(self): state = teststate(self) 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) lock = state.makelock() state.assertacquirecalled(True) lock.lock() # fake a fork lock.pid += 1 lock.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__)