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

rebase: move storestatus onto rebaseruntime...

rebase: move storestatus onto rebaseruntime The rebaseruntime class already has the restorestatus function, so let's make it own the store status function too. This get's rid of a lot of unnecessary argument passing and will make a future patch cleaner that refactors storestatus to support transactions.

Gregory Szorc - - Load All Authors

File last commit:

r30895:c32454d6 default


                r31223:685b8d07

default

Download file

             test_roundtrip.py
        
                    68 lines
            
             | 2.2 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

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

              r30895
            
      from .common import (

          make_cffi,

      )

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

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

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

              r30895
            
      @make_cffi

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

              r30435
            
      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

Gregory Szorc zstd: vendor python-zstandard 0.7.0...	r30895	from .common import (
		make_cffi,
		)
Gregory Szorc zstd: vendor python-zstandard 0.5.0...	r30435
		compression_levels = strategies.integers(min_value=1, max_value=22)


Gregory Szorc zstd: vendor python-zstandard 0.7.0...	r30895	@make_cffi
Gregory Szorc zstd: vendor python-zstandard 0.5.0...	r30435	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)