import os

import unittest

import zstandard as zstd

    generate_samples,

class TestFrameHeaderSize(unittest.TestCase):

    def test_empty(self):

        with self.assertRaisesRegexp(

            zstd.ZstdError, 'could not determine frame header size: Src size '

                            'is incorrect'):

            zstd.frame_header_size(b'')

    def test_too_small(self):

        with self.assertRaisesRegexp(

            zstd.ZstdError, 'could not determine frame header size: Src size '

                            'is incorrect'):

            zstd.frame_header_size(b'foob')

    def test_basic(self):

        # It doesn't matter that it isn't a valid frame.

        self.assertEqual(zstd.frame_header_size(b'long enough but no magic'), 6)

@make_cffi

class TestFrameContentSize(unittest.TestCase):

    def test_empty(self):

        with self.assertRaisesRegexp(zstd.ZstdError,

                                     'error when determining content size'):

            zstd.frame_content_size(b'')

    def test_too_small(self):

        with self.assertRaisesRegexp(zstd.ZstdError,

                                     'error when determining content size'):

            zstd.frame_content_size(b'foob')

    def test_bad_frame(self):

        with self.assertRaisesRegexp(zstd.ZstdError,

                                     'error when determining content size'):

            zstd.frame_content_size(b'invalid frame header')

    def test_unknown(self):

        cctx = zstd.ZstdCompressor(write_content_size=False)

        frame = cctx.compress(b'foobar')

        self.assertEqual(zstd.frame_content_size(frame), -1)

    def test_empty(self):

        cctx = zstd.ZstdCompressor()

        frame = cctx.compress(b'')

        self.assertEqual(zstd.frame_content_size(frame), 0)

    def test_basic(self):

        cctx = zstd.ZstdCompressor()

        frame = cctx.compress(b'foobar')

        self.assertEqual(zstd.frame_content_size(frame), 6)

@make_cffi

class TestDecompressor(unittest.TestCase):

    def test_memory_size(self):

        dctx = zstd.ZstdDecompressor()

        self.assertGreater(dctx.memory_size(), 100)

@make_cffi

        with self.assertRaisesRegexp(zstd.ZstdError, 'error determining content size from frame header'):

        with self.assertRaisesRegexp(zstd.ZstdError, 'error determining content size from frame header'):

    def test_input_types(self):

        cctx = zstd.ZstdCompressor(level=1)

        compressed = cctx.compress(b'foo')

        mutable_array = bytearray(len(compressed))

        mutable_array[:] = compressed

        sources = [

            memoryview(compressed),

            bytearray(compressed),

            mutable_array,

        ]

        dctx = zstd.ZstdDecompressor()

        for source in sources:

            self.assertEqual(dctx.decompress(source), b'foo')

        with self.assertRaisesRegexp(zstd.ZstdError, 'could not determine content size in frame header'):

        cctx = zstd.ZstdCompressor()

    def test_empty_roundtrip(self):

        cctx = zstd.ZstdCompressor()

        compressed = cctx.compress(b'')

        dctx = zstd.ZstdDecompressor()

        decompressed = dctx.decompress(compressed)

        self.assertEqual(decompressed, b'')

        with self.assertRaisesRegexp(zstd.ZstdError,

                'decompression error: did not decompress full frame'):

        cctx = zstd.ZstdCompressor(level=1, dict_data=d)

        cctx = zstd.ZstdCompressor(level=1, dict_data=d)

    def test_max_window_size(self):

        with open(__file__, 'rb') as fh:

            source = fh.read()

        # If we write a content size, the decompressor engages single pass

        # mode and the window size doesn't come into play.

        cctx = zstd.ZstdCompressor(write_content_size=False)

        frame = cctx.compress(source)

        dctx = zstd.ZstdDecompressor(max_window_size=1)

        with self.assertRaisesRegexp(

            zstd.ZstdError, 'decompression error: Frame requires too much memory'):

            dctx.decompress(frame, max_output_size=len(source))

class TestDecompressor_stream_reader(unittest.TestCase):

    def test_context_manager(self):

        dctx = zstd.ZstdDecompressor()

        reader = dctx.stream_reader(b'foo')

        with self.assertRaisesRegexp(zstd.ZstdError, 'read\(\) must be called from an active'):

            reader.read(1)

        with dctx.stream_reader(b'foo') as reader:

            with self.assertRaisesRegexp(ValueError, 'cannot __enter__ multiple times'):

                with reader as reader2:

                    pass

    def test_not_implemented(self):

        dctx = zstd.ZstdDecompressor()

        with dctx.stream_reader(b'foo') as reader:

            with self.assertRaises(NotImplementedError):

                reader.readline()

            with self.assertRaises(NotImplementedError):

                reader.readlines()

            with self.assertRaises(NotImplementedError):

                reader.readall()

            with self.assertRaises(NotImplementedError):

                iter(reader)

            with self.assertRaises(NotImplementedError):

                next(reader)

            with self.assertRaises(io.UnsupportedOperation):

                reader.write(b'foo')

            with self.assertRaises(io.UnsupportedOperation):

                reader.writelines([])

    def test_constant_methods(self):

        dctx = zstd.ZstdDecompressor()

        with dctx.stream_reader(b'foo') as reader:

            self.assertTrue(reader.readable())

            self.assertFalse(reader.writable())

            self.assertTrue(reader.seekable())

            self.assertFalse(reader.isatty())

            self.assertIsNone(reader.flush())

    def test_read_closed(self):

        dctx = zstd.ZstdDecompressor()

        with dctx.stream_reader(b'foo') as reader:

            reader.close()

            with self.assertRaisesRegexp(ValueError, 'stream is closed'):

                reader.read(1)

    def test_bad_read_size(self):

        dctx = zstd.ZstdDecompressor()

        with dctx.stream_reader(b'foo') as reader:

            with self.assertRaisesRegexp(ValueError, 'cannot read negative or size 0 amounts'):

                reader.read(-1)

            with self.assertRaisesRegexp(ValueError, 'cannot read negative or size 0 amounts'):

                reader.read(0)

    def test_read_buffer(self):

        cctx = zstd.ZstdCompressor()

        source = b''.join([b'foo' * 60, b'bar' * 60, b'baz' * 60])

        frame = cctx.compress(source)

        dctx = zstd.ZstdDecompressor()

        with dctx.stream_reader(frame) as reader:

            self.assertEqual(reader.tell(), 0)

            # We should get entire frame in one read.

            result = reader.read(8192)

            self.assertEqual(result, source)

            self.assertEqual(reader.tell(), len(source))

            # Read after EOF should return empty bytes.

            self.assertEqual(reader.read(), b'')

            self.assertEqual(reader.tell(), len(result))

        self.assertTrue(reader.closed())

    def test_read_buffer_small_chunks(self):

        cctx = zstd.ZstdCompressor()