upstream/mercurial-mirror Files · contrib/python-zstandard/tests/common.py

rebase: allow aborting if last-message.txt is missing...

rebase: allow aborting if last-message.txt is missing Previously, if .hg/rebasestate existed but .hg/last-message.txt was missing, 'hg rebase --abort' would say there's no rebase in progress but 'hg checkout foo' would say 'abort: rebase in progress'. It turns out loading the collapse message will throw a "no rebase in progress" error if the file doesn't exist, even though .hg/rebasestate obviously indicates a rebase is in progress. The fix is to only throw an exception if we're trying to --continue, and to just eat the issues if we're doing --abort. This issue is exposed by us writing the rebase state earlier in the process. This will be used by later patches to ensure the user can appropriately 'hg rebase --abort' if there's a crash before the first the first commit has finished rebasing. Tests cover all of this. The only negative affect is we now require a hg rebase --abort in a very specific exception case, as shown in the test.

Gregory Szorc - - Load All Authors

File last commit:

r30895:c32454d6 default


                r31225:749b057b

default

Download file

             common.py
        
                    61 lines
            
             | 1.9 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / contrib / python-zstandard / tests / common.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      import inspect

      import io

      import types

      def make_cffi(cls):

          """Decorator to add CFFI versions of each test method."""

          try:

              import zstd_cffi

          except ImportError:

              return cls

          # If CFFI version is available, dynamically construct test methods

          # that use it.

          for attr in dir(cls):

              fn = getattr(cls, attr)

              if not inspect.ismethod(fn) and not inspect.isfunction(fn):

                  continue

              if not fn.__name__.startswith('test_'):

                  continue

              name = '%s_cffi' % fn.__name__

              # Replace the "zstd" symbol with the CFFI module instance. Then copy

              # the function object and install it in a new attribute.

              if isinstance(fn, types.FunctionType):

                  globs = dict(fn.__globals__)

                  globs['zstd'] = zstd_cffi

                  new_fn = types.FunctionType(fn.__code__, globs, name,

                                              fn.__defaults__, fn.__closure__)

                  new_method = new_fn

              else:

                  globs = dict(fn.__func__.func_globals)

                  globs['zstd'] = zstd_cffi

                  new_fn = types.FunctionType(fn.__func__.func_code, globs, name,

                                              fn.__func__.func_defaults,

                                              fn.__func__.func_closure)

                  new_method = types.UnboundMethodType(new_fn, fn.im_self,

                                                       fn.im_class)

              setattr(cls, name, new_method)

          return cls

      class OpCountingBytesIO(io.BytesIO):

          def __init__(self, *args, **kwargs):

              self._read_count = 0

              self._write_count = 0

              return super(OpCountingBytesIO, self).__init__(*args, **kwargs)

          def read(self, *args):

              self._read_count += 1

              return super(OpCountingBytesIO, self).read(*args)

          def write(self, data):

              self._write_count += 1

              return super(OpCountingBytesIO, self).write(data)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				import inspect
				import io
				import types


				def make_cffi(cls):
				"""Decorator to add CFFI versions of each test method."""

				try:
				import zstd_cffi
				except ImportError:
				return cls

				# If CFFI version is available, dynamically construct test methods
				# that use it.

				for attr in dir(cls):
				fn = getattr(cls, attr)
				if not inspect.ismethod(fn) and not inspect.isfunction(fn):
				continue

				if not fn.__name__.startswith('test_'):
				continue

				name = '%s_cffi' % fn.__name__

				# Replace the "zstd" symbol with the CFFI module instance. Then copy
				# the function object and install it in a new attribute.
				if isinstance(fn, types.FunctionType):
				globs = dict(fn.__globals__)
				globs['zstd'] = zstd_cffi
				new_fn = types.FunctionType(fn.__code__, globs, name,
				fn.__defaults__, fn.__closure__)
				new_method = new_fn
				else:
				globs = dict(fn.__func__.func_globals)
				globs['zstd'] = zstd_cffi
				new_fn = types.FunctionType(fn.__func__.func_code, globs, name,
				fn.__func__.func_defaults,
				fn.__func__.func_closure)
				new_method = types.UnboundMethodType(new_fn, fn.im_self,
				fn.im_class)

				setattr(cls, name, new_method)

				return cls


				class OpCountingBytesIO(io.BytesIO):
				def __init__(self, args, *kwargs):
				self._read_count = 0
				self._write_count = 0
				return super(OpCountingBytesIO, self).__init__(args, *kwargs)

				def read(self, *args):
				self._read_count += 1
				return super(OpCountingBytesIO, self).read(*args)

				def write(self, data):
				self._write_count += 1
				return super(OpCountingBytesIO, self).write(data)