1 import io

     2 

     3 try:

     4     import unittest2 as unittest

     5 except ImportError:

     6     import unittest

     7 

     8 import zstd

     9 

    10 try:

    11     import zstd_cffi

    12 except ImportError:

    13     raise unittest.SkipTest('cffi version of zstd not available')

    14 

    15 

    16 class TestCFFIWriteToToCDecompressor(unittest.TestCase):

    17     def test_simple(self):

    18         orig = io.BytesIO()

    19         orig.write(b'foo')

    20         orig.write(b'bar')

    21         orig.write(b'foobar' * 16384)

    22 

    23         dest = io.BytesIO()

    24         cctx = zstd_cffi.ZstdCompressor()

    25         with cctx.write_to(dest) as compressor:

    26             compressor.write(orig.getvalue())

    27 

    28         uncompressed = io.BytesIO()

    29         dctx = zstd.ZstdDecompressor()

    30         with dctx.write_to(uncompressed) as decompressor:

    31             decompressor.write(dest.getvalue())

    32 

    33         self.assertEqual(uncompressed.getvalue(), orig.getvalue())

    34 

    35