import hashlib

import io

import os

import struct

import sys

import tarfile

import tempfile

import unittest

import zstandard as zstd

from .common import (

    make_cffi,

    NonClosingBytesIO,

    OpCountingBytesIO,

)

if sys.version_info[0] >= 3:

    next = lambda it: it.__next__()

else:

    next = lambda it: it.next()

def multithreaded_chunk_size(level, source_size=0):

    return 1 << (params.window_log + 2)

@make_cffi

    def test_level_bounds(self):

        with self.assertRaises(ValueError):

            zstd.ZstdCompressor(level=23)

    def test_memory_size(self):

        cctx = zstd.ZstdCompressor(level=1)

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

@make_cffi

    def test_compress_empty(self):

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        params = zstd.get_frame_parameters(result)

        self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(params.window_size, 524288)

        self.assertEqual(params.dict_id, 0)

        self.assertFalse(params.has_checksum, 0)

        cctx = zstd.ZstdCompressor()

        params = zstd.get_frame_parameters(result)

        self.assertEqual(params.content_size, 0)

    def test_input_types(self):

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        mutable_array = bytearray(3)

        sources = [

            mutable_array,

        ]

        for source in sources:

            self.assertEqual(cctx.compress(source), expected)

    def test_compress_large(self):

        chunks = []

        for i in range(255):

        cctx = zstd.ZstdCompressor(level=3, write_content_size=False)

        self.assertEqual(len(result), 999)

        # This matches the test for read_to_iter() below.

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

    def test_negative_level(self):

        cctx = zstd.ZstdCompressor(level=-4)

    def test_no_magic(self):

        cctx = zstd.ZstdCompressor(compression_params=params)

        params = zstd.ZstdCompressionParameters.from_level(

        cctx = zstd.ZstdCompressor(compression_params=params)

        self.assertEqual(magic[4:], no_magic)

    def test_write_checksum(self):

        cctx = zstd.ZstdCompressor(level=1)

        cctx = zstd.ZstdCompressor(level=1, write_checksum=True)

        self.assertEqual(len(with_checksum), len(no_checksum) + 4)

        no_params = zstd.get_frame_parameters(no_checksum)

        with_params = zstd.get_frame_parameters(with_checksum)

        self.assertFalse(no_params.has_checksum)

        self.assertTrue(with_params.has_checksum)

    def test_write_content_size(self):

        cctx = zstd.ZstdCompressor(level=1)

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        self.assertEqual(len(with_size), len(no_size) + 1)

        no_params = zstd.get_frame_parameters(no_size)

        with_params = zstd.get_frame_parameters(with_size)

        self.assertEqual(no_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(with_params.content_size, 1536)

    def test_no_dict_id(self):

        samples = []

        for i in range(128):

        d = zstd.train_dictionary(1024, samples)

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

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

        self.assertEqual(len(with_dict_id), len(no_dict_id) + 4)

        no_params = zstd.get_frame_parameters(no_dict_id)

        with_params = zstd.get_frame_parameters(with_dict_id)

        self.assertEqual(no_params.dict_id, 0)

        self.assertEqual(with_params.dict_id, 1880053135)

    def test_compress_dict_multiple(self):

        samples = []

        for i in range(128):

        d = zstd.train_dictionary(8192, samples)

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

        for i in range(32):

    def test_dict_precompute(self):

        samples = []

        for i in range(128):

        d = zstd.train_dictionary(8192, samples)

        d.precompute_compress(level=1)

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

        for i in range(32):

    def test_multithreaded(self):

        chunk_size = multithreaded_chunk_size(1)

        cctx = zstd.ZstdCompressor(level=1, threads=2)

        compressed = cctx.compress(source)

        params = zstd.get_frame_parameters(compressed)

        self.assertEqual(params.content_size, chunk_size * 2)

        self.assertEqual(params.dict_id, 0)

        self.assertFalse(params.has_checksum)

        dctx = zstd.ZstdDecompressor()

        self.assertEqual(dctx.decompress(compressed), source)

    def test_multithreaded_dict(self):

        samples = []

        for i in range(128):

        d = zstd.train_dictionary(1024, samples)

        cctx = zstd.ZstdCompressor(dict_data=d, threads=2)

        self.assertEqual(params.dict_id, d.dict_id())

    def test_multithreaded_compression_params(self):

        params = zstd.ZstdCompressionParameters.from_level(0, threads=2)

        cctx = zstd.ZstdCompressor(compression_params=params)

@make_cffi

    def test_compressobj_empty(self):

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        cobj = cctx.compressobj()

    def test_input_types(self):

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        mutable_array = bytearray(3)

        sources = [

            mutable_array,

        ]

        for source in sources:

            cobj = cctx.compressobj()

            self.assertEqual(cobj.flush(), expected)

    def test_compressobj_large(self):

        chunks = []

        for i in range(255):

        cctx = zstd.ZstdCompressor(level=3)

        cobj = cctx.compressobj()

        self.assertEqual(len(result), 999)

        params = zstd.get_frame_parameters(result)

        self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(params.window_size, 2097152)

        self.assertEqual(params.dict_id, 0)

        self.assertFalse(params.has_checksum)

    def test_write_checksum(self):

        cctx = zstd.ZstdCompressor(level=1)

        cobj = cctx.compressobj()

        cctx = zstd.ZstdCompressor(level=1, write_checksum=True)

        cobj = cctx.compressobj()

        no_params = zstd.get_frame_parameters(no_checksum)

        with_params = zstd.get_frame_parameters(with_checksum)

        self.assertEqual(no_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(with_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(no_params.dict_id, 0)

        self.assertEqual(with_params.dict_id, 0)

        self.assertFalse(no_params.has_checksum)

        self.assertTrue(with_params.has_checksum)

        self.assertEqual(len(with_checksum), len(no_checksum) + 4)

    def test_write_content_size(self):

        cctx = zstd.ZstdCompressor(level=1)

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        no_params = zstd.get_frame_parameters(no_size)

        with_params = zstd.get_frame_parameters(with_size)

        self.assertEqual(no_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(with_params.content_size, 1536)

        self.assertEqual(no_params.dict_id, 0)

        self.assertEqual(with_params.dict_id, 0)

        self.assertFalse(no_params.has_checksum)

        self.assertFalse(with_params.has_checksum)

        self.assertEqual(len(with_size), len(no_size) + 1)

    def test_compress_after_finished(self):

        cctx = zstd.ZstdCompressor()

        cobj = cctx.compressobj()

        cobj.flush()

            cobj.flush()

    def test_flush_block_repeated(self):

        cctx = zstd.ZstdCompressor(level=1)

        cobj = cctx.compressobj()

        # 3 byte header plus content.

    def test_flush_empty_block(self):

        cctx = zstd.ZstdCompressor(write_checksum=True)

        cobj = cctx.compressobj()

        cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK)

        # No-op if no block is active (this is internal to zstd).

        trailing = cobj.flush()

        # 3 bytes block header + 4 bytes frame checksum

        self.assertEqual(len(trailing), 7)

        header = trailing[0:3]

    def test_multithreaded(self):

        source = io.BytesIO()

        source.seek(0)

        cctx = zstd.ZstdCompressor(level=1, threads=2)

        cobj = cctx.compressobj()

        chunks = []

        while True:

            d = source.read(8192)

            if not d:

                break

            chunks.append(cobj.compress(d))

        chunks.append(cobj.flush())

    def test_frame_progression(self):

        cctx = zstd.ZstdCompressor()

        self.assertEqual(cctx.frame_progression(), (0, 0, 0))

        cobj = cctx.compressobj()

        self.assertEqual(cctx.frame_progression(), (6, 0, 0))

        cobj.flush()

        self.assertEqual(cctx.frame_progression(), (6, 6, 15))

    def test_bad_size(self):

        cctx = zstd.ZstdCompressor()

        cobj = cctx.compressobj(size=2)

        # Try another operation on this instance.

        # Try another operation on the compressor.

        cctx.compressobj(size=4)

@make_cffi

    def test_no_read(self):

        source = object()

        dest = io.BytesIO()

        cctx = zstd.ZstdCompressor()

        with self.assertRaises(ValueError):

            cctx.copy_stream(source, dest)

    def test_no_write(self):

        source = io.BytesIO()

        dest = object()

        cctx = zstd.ZstdCompressor()

        with self.assertRaises(ValueError):

            cctx.copy_stream(source, dest)

    def test_empty(self):

        source = io.BytesIO()

        dest = io.BytesIO()

        cctx = zstd.ZstdCompressor(level=1, write_content_size=False)

        r, w = cctx.copy_stream(source, dest)

        self.assertEqual(int(r), 0)

        self.assertEqual(w, 9)

    def test_large_data(self):

        source = io.BytesIO()

        for i in range(255):

        source.seek(0)

        dest = io.BytesIO()

        cctx = zstd.ZstdCompressor()

        r, w = cctx.copy_stream(source, dest)

        self.assertEqual(r, 255 * 16384)

        self.assertEqual(w, 999)

        params = zstd.get_frame_parameters(dest.getvalue())

        self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(params.window_size, 2097152)

        self.assertEqual(params.dict_id, 0)

        self.assertFalse(params.has_checksum)

    def test_write_checksum(self):

        no_checksum = io.BytesIO()

        cctx = zstd.ZstdCompressor(level=1)

        cctx.copy_stream(source, no_checksum)

        source.seek(0)

        with_checksum = io.BytesIO()

        cctx = zstd.ZstdCompressor(level=1, write_checksum=True)

        cctx.copy_stream(source, with_checksum)

        no_params = zstd.get_frame_parameters(no_checksum.getvalue())

        with_params = zstd.get_frame_parameters(with_checksum.getvalue())

        self.assertEqual(no_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(with_params.content_size, zstd.CONTENTSIZE_UNKNOWN)

        self.assertEqual(no_params.dict_id, 0)

        self.assertEqual(with_params.dict_id, 0)

        self.assertFalse(no_params.has_checksum)

        self.assertTrue(with_params.has_checksum)