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

bundle2: keep hint close to the primary message when remote abort...

bundle2: keep hint close to the primary message when remote abort The remote hint message was ignored when reporting the remote error and passed to the local generic abort error. I think I might initially have tried to avoid reimplementing logic controlling the hint display depending of the verbosity level. However, first, there does not seems to have such verbosity related logic and second the resulting was wrong as the primary error and the hint were split apart. We now properly print the hint as remote output.

Gregory Szorc - - Load All Authors

File last commit:

r30895:c32454d6 default


                r30908:4c8dcb49

stable

Download file

             test_roundtrip.py
        
                    64 lines
            
             | 2.1 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / contrib / python-zstandard / tests / test_roundtrip.py
          
                    History
                
                 |
                  Source
                 | Raw
                 |Copy content
                 |Copy permalink

        Gregory Szorc
    
zstd: vendor python-zstandard 0.5.0...

              r30435
            
      import io

      try:

          import unittest2 as unittest

      except ImportError:

          import unittest

      try:

          import hypothesis

          import hypothesis.strategies as strategies

      except ImportError:

          raise unittest.SkipTest('hypothesis not available')

      import zstd

      compression_levels = strategies.integers(min_value=1, max_value=22)

      class TestRoundTrip(unittest.TestCase):

          @hypothesis.given(strategies.binary(), compression_levels)

          def test_compress_write_to(self, data, level):

              """Random data from compress() roundtrips via write_to."""

              cctx = zstd.ZstdCompressor(level=level)

              compressed = cctx.compress(data)

              buffer = io.BytesIO()

              dctx = zstd.ZstdDecompressor()

              with dctx.write_to(buffer) as decompressor:

                  decompressor.write(compressed)

              self.assertEqual(buffer.getvalue(), data)

          @hypothesis.given(strategies.binary(), compression_levels)

          def test_compressor_write_to_decompressor_write_to(self, data, level):

              """Random data from compressor write_to roundtrips via write_to."""

              compress_buffer = io.BytesIO()

              decompressed_buffer = io.BytesIO()

              cctx = zstd.ZstdCompressor(level=level)

              with cctx.write_to(compress_buffer) as compressor:

                  compressor.write(data)

              dctx = zstd.ZstdDecompressor()

              with dctx.write_to(decompressed_buffer) as decompressor:

                  decompressor.write(compress_buffer.getvalue())

              self.assertEqual(decompressed_buffer.getvalue(), data)

          @hypothesis.given(strategies.binary(average_size=1048576))

          @hypothesis.settings(perform_health_check=False)

          def test_compressor_write_to_decompressor_write_to_larger(self, data):

              compress_buffer = io.BytesIO()

              decompressed_buffer = io.BytesIO()

              cctx = zstd.ZstdCompressor(level=5)

              with cctx.write_to(compress_buffer) as compressor:

                  compressor.write(data)

              dctx = zstd.ZstdDecompressor()

              with dctx.write_to(decompressed_buffer) as decompressor:

                  decompressor.write(compress_buffer.getvalue())

              self.assertEqual(decompressed_buffer.getvalue(), 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

Gregory Szorc zstd: vendor python-zstandard 0.5.0...	r30435	import io

		try:
		import unittest2 as unittest
		except ImportError:
		import unittest

		try:
		import hypothesis
		import hypothesis.strategies as strategies
		except ImportError:
		raise unittest.SkipTest('hypothesis not available')

		import zstd


		compression_levels = strategies.integers(min_value=1, max_value=22)


		class TestRoundTrip(unittest.TestCase):
		@hypothesis.given(strategies.binary(), compression_levels)
		def test_compress_write_to(self, data, level):
		"""Random data from compress() roundtrips via write_to."""
		cctx = zstd.ZstdCompressor(level=level)
		compressed = cctx.compress(data)

		buffer = io.BytesIO()
		dctx = zstd.ZstdDecompressor()
		with dctx.write_to(buffer) as decompressor:
		decompressor.write(compressed)

		self.assertEqual(buffer.getvalue(), data)

		@hypothesis.given(strategies.binary(), compression_levels)
		def test_compressor_write_to_decompressor_write_to(self, data, level):
		"""Random data from compressor write_to roundtrips via write_to."""
		compress_buffer = io.BytesIO()
		decompressed_buffer = io.BytesIO()

		cctx = zstd.ZstdCompressor(level=level)
		with cctx.write_to(compress_buffer) as compressor:
		compressor.write(data)

		dctx = zstd.ZstdDecompressor()
		with dctx.write_to(decompressed_buffer) as decompressor:
		decompressor.write(compress_buffer.getvalue())

		self.assertEqual(decompressed_buffer.getvalue(), data)

		@hypothesis.given(strategies.binary(average_size=1048576))
		@hypothesis.settings(perform_health_check=False)
		def test_compressor_write_to_decompressor_write_to_larger(self, data):
		compress_buffer = io.BytesIO()
		decompressed_buffer = io.BytesIO()

		cctx = zstd.ZstdCompressor(level=5)
		with cctx.write_to(compress_buffer) as compressor:
		compressor.write(data)

		dctx = zstd.ZstdDecompressor()
		with dctx.write_to(decompressed_buffer) as decompressor:
		decompressor.write(compress_buffer.getvalue())

		self.assertEqual(decompressed_buffer.getvalue(), data)