Mercurial > hg
view tests/test-lock.py @ 26304:bca60842e22d
update: move default destination into a revset
This is another step toward having "default" destination more clear and unified.
Not all the logic is there because some bookmark related computation happened
elsewhere. It will be moved later.
The function is private because as for the other ones, cleanup is needed before
we can proceed.
author | Pierre-Yves David <pierre-yves.david@fb.com> |
---|---|
date | Fri, 18 Sep 2015 17:23:10 -0700 |
parents | c4b667a7a51d |
children | db4c192cb9b3 |
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._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, *args, **kwargs) l.postrelease.append(self.postreleasefn) return l def releasefn(self): self._releasecalled = True def postreleasefn(self): self._postreleasecalled = True 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() lock.release() state.assertreleasecalled(True) state.assertpostreleasecalled(True) state.assertlockexists(False) def testrecursivelock(self): state = teststate(self) lock = state.makelock() lock.lock() 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() 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__)