diff --git a/contrib/python3-whitelist b/contrib/python3-whitelist --- a/contrib/python3-whitelist +++ b/contrib/python3-whitelist @@ -9,7 +9,6 @@ test-check-shbang.t test-contrib-check-code.t test-contrib-check-commit.t -test-ctxmanager.py test-diff-issue2761.t test-diff-newlines.t test-diff-reverse.t diff --git a/mercurial/util.py b/mercurial/util.py --- a/mercurial/util.py +++ b/mercurial/util.py @@ -3052,66 +3052,6 @@ yield path[:pos] pos = path.rfind('/', 0, pos) -class ctxmanager(object): - '''A context manager for use in 'with' blocks to allow multiple - contexts to be entered at once. This is both safer and more - flexible than contextlib.nested. - - Once Mercurial supports Python 2.7+, this will become mostly - unnecessary. - ''' - - def __init__(self, *args): - '''Accepts a list of no-argument functions that return context - managers. These will be invoked at __call__ time.''' - self._pending = args - self._atexit = [] - - def __enter__(self): - return self - - def enter(self): - '''Create and enter context managers in the order in which they were - passed to the constructor.''' - values = [] - for func in self._pending: - obj = func() - values.append(obj.__enter__()) - self._atexit.append(obj.__exit__) - del self._pending - return values - - def atexit(self, func, *args, **kwargs): - '''Add a function to call when this context manager exits. The - ordering of multiple atexit calls is unspecified, save that - they will happen before any __exit__ functions.''' - def wrapper(exc_type, exc_val, exc_tb): - func(*args, **kwargs) - self._atexit.append(wrapper) - return func - - def __exit__(self, exc_type, exc_val, exc_tb): - '''Context managers are exited in the reverse order from which - they were created.''' - received = exc_type is not None - suppressed = False - pending = None - self._atexit.reverse() - for exitfunc in self._atexit: - try: - if exitfunc(exc_type, exc_val, exc_tb): - suppressed = True - exc_type = None - exc_val = None - exc_tb = None - except BaseException: - pending = sys.exc_info() - exc_type, exc_val, exc_tb = pending = sys.exc_info() - del self._atexit - if pending: - raise exc_val - return received and suppressed - # compression code SERVERROLE = 'server' diff --git a/tests/test-check-module-imports.t b/tests/test-check-module-imports.t --- a/tests/test-check-module-imports.t +++ b/tests/test-check-module-imports.t @@ -24,7 +24,6 @@ > -X i18n/posplit \ > -X tests/test-hgweb-auth.py \ > -X tests/hypothesishelpers.py \ - > -X tests/test-ctxmanager.py \ > -X tests/test-lock.py \ > -X tests/test-verify-repo-operations.py \ > -X tests/test-hook.t \ diff --git a/tests/test-ctxmanager.py b/tests/test-ctxmanager.py deleted file mode 100644 --- a/tests/test-ctxmanager.py +++ /dev/null @@ -1,77 +0,0 @@ -from __future__ import absolute_import - -import silenttestrunner -import unittest - -from mercurial import util - -class contextmanager(object): - def __init__(self, name, trace): - self.name = name - self.entered = False - self.exited = False - self.trace = trace - - def __enter__(self): - self.entered = True - self.trace(('enter', self.name)) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.exited = exc_type, exc_val, exc_tb - self.trace(('exit', self.name)) - - def __repr__(self): - return '' % self.name - -class ctxerror(Exception): - pass - -class raise_on_enter(contextmanager): - def __enter__(self): - self.trace(('raise', self.name)) - raise ctxerror(self.name) - -class raise_on_exit(contextmanager): - def __exit__(self, exc_type, exc_val, exc_tb): - self.trace(('raise', self.name)) - raise ctxerror(self.name) - -def ctxmgr(name, trace): - return lambda: contextmanager(name, trace) - -class test_ctxmanager(unittest.TestCase): - def test_basics(self): - trace = [] - addtrace = trace.append - with util.ctxmanager(ctxmgr('a', addtrace), ctxmgr('b', addtrace)) as c: - a, b = c.enter() - c.atexit(addtrace, ('atexit', 'x')) - c.atexit(addtrace, ('atexit', 'y')) - self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'), - ('atexit', 'y'), ('atexit', 'x'), - ('exit', 'b'), ('exit', 'a')]) - - def test_raise_on_enter(self): - trace = [] - addtrace = trace.append - with self.assertRaises(ctxerror): - with util.ctxmanager(ctxmgr('a', addtrace), - lambda: raise_on_enter('b', addtrace)) as c: - c.enter() - addtrace('unreachable') - self.assertEqual(trace, [('enter', 'a'), ('raise', 'b'), ('exit', 'a')]) - - def test_raise_on_exit(self): - trace = [] - addtrace = trace.append - with self.assertRaises(ctxerror): - with util.ctxmanager(ctxmgr('a', addtrace), - lambda: raise_on_exit('b', addtrace)) as c: - c.enter() - addtrace('running') - self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'), 'running', - ('raise', 'b'), ('exit', 'a')]) - -if __name__ == '__main__': - silenttestrunner.main(__name__)