util: remove unused ctxmanager
This was meant as a substitute for Python's "with" with multiple
context managers before we moved to Python 2.7. We're now on 2.7, so
we should have no reason to keep ctxmanager. "hg grep --all
ctxmanager" says that it was never used anyway.
Differential Revision: https://phab.mercurial-scm.org/D73
--- a/contrib/python3-whitelist Thu Jul 13 18:31:35 2017 -0700
+++ b/contrib/python3-whitelist Thu Jul 13 09:51:50 2017 -0700
@@ -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
--- a/mercurial/util.py Thu Jul 13 18:31:35 2017 -0700
+++ b/mercurial/util.py Thu Jul 13 09:51:50 2017 -0700
@@ -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'
--- a/tests/test-check-module-imports.t Thu Jul 13 18:31:35 2017 -0700
+++ b/tests/test-check-module-imports.t Thu Jul 13 09:51:50 2017 -0700
@@ -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 \
--- a/tests/test-ctxmanager.py Thu Jul 13 18:31:35 2017 -0700
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -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 '<ctx %r>' % 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__)