import _compression
import array
from io import BytesIO, UnsupportedOperation, DEFAULT_BUFFER_SIZE
import os
import pathlib
import pickle
import random
import sys
from test import support
import unittest

from test.support import (
    _4G, bigmemtest, run_unittest
)
from test.support.import_helper import import_module
from test.support.os_helper import (
    TESTFN, unlink
)

lzma = import_module("lzma")
from lzma import LZMACompressor, LZMADecompressor, LZMAError, LZMAFile


class CompressorDecompressorTestCase(unittest.TestCase):

    # Test error cases.

    def test_simple_bad_args(self):
        self.assertRaises(TypeError, LZMACompressor, [])
        self.assertRaises(TypeError, LZMACompressor, format=3.45)
        self.assertRaises(TypeError, LZMACompressor, check="")
        self.assertRaises(TypeError, LZMACompressor, preset="asdf")
        self.assertRaises(TypeError, LZMACompressor, filters=3)
        # Can't specify FORMAT_AUTO when compressing.
        self.assertRaises(ValueError, LZMACompressor, format=lzma.FORMAT_AUTO)
        # Can't specify a preset and a custom filter chain at the same time.
        with self.assertRaises(ValueError):
            LZMACompressor(preset=7, filters=[{"id": lzma.FILTER_LZMA2}])

        self.assertRaises(TypeError, LZMADecompressor, ())
        self.assertRaises(TypeError, LZMADecompressor, memlimit=b"qw")
        with self.assertRaises(TypeError):
            LZMADecompressor(lzma.FORMAT_RAW, filters="zzz")
        # Cannot specify a memory limit with FILTER_RAW.
        with self.assertRaises(ValueError):
            LZMADecompressor(lzma.FORMAT_RAW, memlimit=0x1000000)
        # Can only specify a custom filter chain with FILTER_RAW.
        self.assertRaises(ValueError, LZMADecompressor, filters=FILTERS_RAW_1)
        with self.assertRaises(ValueError):
            LZMADecompressor(format=lzma.FORMAT_XZ, filters=FILTERS_RAW_1)
        with self.assertRaises(ValueError):
            LZMADecompressor(format=lzma.FORMAT_ALONE, filters=FILTERS_RAW_1)

        lzc = LZMACompressor()
        self.assertRaises(TypeError, lzc.compress)
        self.assertRaises(TypeError, lzc.compress, b"foo", b"bar")
        self.assertRaises(TypeError, lzc.flush, b"blah")
        empty = lzc.flush()
        self.assertRaises(ValueError, lzc.compress, b"quux")
        self.assertRaises(ValueError, lzc.flush)

        lzd = LZMADecompressor()
        self.assertRaises(TypeError, lzd.decompress)
        self.assertRaises(TypeError, lzd.decompress, b"foo", b"bar")
        lzd.decompress(empty)
        self.assertRaises(EOFError, lzd.decompress, b"quux")

    def test_bad_filter_spec(self):
        self.assertRaises(TypeError, LZMACompressor, filters=[b"wobsite"])
        self.assertRaises(ValueError, LZMACompressor, filters=[{"xyzzy": 3}])
        self.assertRaises(ValueError, LZMACompressor, filters=[{"id": 98765}])
        with self.assertRaises(ValueError):
            LZMACompressor(filters=[{"id": lzma.FILTER_LZMA2, "foo": 0}])
        with self.assertRaises(ValueError):
            LZMACompressor(filters=[{"id": lzma.FILTER_DELTA, "foo": 0}])
        with self.assertRaises(ValueError):
            LZMACompressor(filters=[{"id": lzma.FILTER_X86, "foo": 0}])

    def test_decompressor_after_eof(self):
        lzd = LZMADecompressor()
        lzd.decompress(COMPRESSED_XZ)
        self.assertRaises(EOFError, lzd.decompress, b"nyan")

    def test_decompressor_memlimit(self):
        lzd = LZMADecompressor(memlimit=1024)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)

        lzd = LZMADecompressor(lzma.FORMAT_XZ, memlimit=1024)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)

        lzd = LZMADecompressor(lzma.FORMAT_ALONE, memlimit=1024)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)

    # Test LZMADecompressor on known-good input data.

    def _test_decompressor(self, lzd, data, check, unused_data=b""):
        self.assertFalse(lzd.eof)
        out = lzd.decompress(data)
        self.assertEqual(out, INPUT)
        self.assertEqual(lzd.check, check)
        self.assertTrue(lzd.eof)
        self.assertEqual(lzd.unused_data, unused_data)

    def test_decompressor_auto(self):
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, COMPRESSED_XZ, lzma.CHECK_CRC64)

        lzd = LZMADecompressor()
        self._test_decompressor(lzd, COMPRESSED_ALONE, lzma.CHECK_NONE)

    def test_decompressor_xz(self):
        lzd = LZMADecompressor(lzma.FORMAT_XZ)
        self._test_decompressor(lzd, COMPRESSED_XZ, lzma.CHECK_CRC64)

    def test_decompressor_alone(self):
        lzd = LZMADecompressor(lzma.FORMAT_ALONE)
        self._test_decompressor(lzd, COMPRESSED_ALONE, lzma.CHECK_NONE)

    def test_decompressor_raw_1(self):
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
        self._test_decompressor(lzd, COMPRESSED_RAW_1, lzma.CHECK_NONE)

    def test_decompressor_raw_2(self):
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_2)
        self._test_decompressor(lzd, COMPRESSED_RAW_2, lzma.CHECK_NONE)

    def test_decompressor_raw_3(self):
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_3)
        self._test_decompressor(lzd, COMPRESSED_RAW_3, lzma.CHECK_NONE)

    def test_decompressor_raw_4(self):
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        self._test_decompressor(lzd, COMPRESSED_RAW_4, lzma.CHECK_NONE)

    def test_decompressor_chunks(self):
        lzd = LZMADecompressor()
        out = []
        for i in range(0, len(COMPRESSED_XZ), 10):
            self.assertFalse(lzd.eof)
            out.append(lzd.decompress(COMPRESSED_XZ[i:i+10]))
        out = b"".join(out)
        self.assertEqual(out, INPUT)
        self.assertEqual(lzd.check, lzma.CHECK_CRC64)
        self.assertTrue(lzd.eof)
        self.assertEqual(lzd.unused_data, b"")

    def test_decompressor_chunks_empty(self):
        lzd = LZMADecompressor()
        out = []
        for i in range(0, len(COMPRESSED_XZ), 10):
            self.assertFalse(lzd.eof)
            out.append(lzd.decompress(b''))
            out.append(lzd.decompress(b''))
            out.append(lzd.decompress(b''))
            out.append(lzd.decompress(COMPRESSED_XZ[i:i+10]))
        out = b"".join(out)
        self.assertEqual(out, INPUT)
        self.assertEqual(lzd.check, lzma.CHECK_CRC64)
        self.assertTrue(lzd.eof)
        self.assertEqual(lzd.unused_data, b"")

    def test_decompressor_chunks_maxsize(self):
        lzd = LZMADecompressor()
        max_length = 100
        out = []

        # Feed first half the input
        len_ = len(COMPRESSED_XZ) // 2
        out.append(lzd.decompress(COMPRESSED_XZ[:len_],
                                  max_length=max_length))
        self.assertFalse(lzd.needs_input)
        self.assertEqual(len(out[-1]), max_length)

        # Retrieve more data without providing more input
        out.append(lzd.decompress(b'', max_length=max_length))
        self.assertFalse(lzd.needs_input)
        self.assertEqual(len(out[-1]), max_length)

        # Retrieve more data while providing more input
        out.append(lzd.decompress(COMPRESSED_XZ[len_:],
                                  max_length=max_length))
        self.assertLessEqual(len(out[-1]), max_length)

        # Retrieve remaining uncompressed data
        while not lzd.eof:
            out.append(lzd.decompress(b'', max_length=max_length))
            self.assertLessEqual(len(out[-1]), max_length)

        out = b"".join(out)
        self.assertEqual(out, INPUT)
        self.assertEqual(lzd.check, lzma.CHECK_CRC64)
        self.assertEqual(lzd.unused_data, b"")

    def test_decompressor_inputbuf_1(self):
        # Test reusing input buffer after moving existing
        # contents to beginning
        lzd = LZMADecompressor()
        out = []

        # Create input buffer and fill it
        self.assertEqual(lzd.decompress(COMPRESSED_XZ[:100],
                                        max_length=0), b'')

        # Retrieve some results, freeing capacity at beginning
        # of input buffer
        out.append(lzd.decompress(b'', 2))

        # Add more data that fits into input buffer after
        # moving existing data to beginning
        out.append(lzd.decompress(COMPRESSED_XZ[100:105], 15))

        # Decompress rest of data
        out.append(lzd.decompress(COMPRESSED_XZ[105:]))
        self.assertEqual(b''.join(out), INPUT)

    def test_decompressor_inputbuf_2(self):
        # Test reusing input buffer by appending data at the
        # end right away
        lzd = LZMADecompressor()
        out = []

        # Create input buffer and empty it
        self.assertEqual(lzd.decompress(COMPRESSED_XZ[:200],
                                        max_length=0), b'')
        out.append(lzd.decompress(b''))

        # Fill buffer with new data
        out.append(lzd.decompress(COMPRESSED_XZ[200:280], 2))

        # Append some more data, not enough to require resize
        out.append(lzd.decompress(COMPRESSED_XZ[280:300], 2))

        # Decompress rest of data
        out.append(lzd.decompress(COMPRESSED_XZ[300:]))
        self.assertEqual(b''.join(out), INPUT)

    def test_decompressor_inputbuf_3(self):
        # Test reusing input buffer after extending it

        lzd = LZMADecompressor()
        out = []

        # Create almost full input buffer
        out.append(lzd.decompress(COMPRESSED_XZ[:200], 5))

        # Add even more data to it, requiring resize
        out.append(lzd.decompress(COMPRESSED_XZ[200:300], 5))

        # Decompress rest of data
        out.append(lzd.decompress(COMPRESSED_XZ[300:]))
        self.assertEqual(b''.join(out), INPUT)

    def test_decompressor_unused_data(self):
        lzd = LZMADecompressor()
        extra = b"fooblibar"
        self._test_decompressor(lzd, COMPRESSED_XZ + extra, lzma.CHECK_CRC64,
                                unused_data=extra)

    def test_decompressor_bad_input(self):
        lzd = LZMADecompressor()
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_RAW_1)

        lzd = LZMADecompressor(lzma.FORMAT_XZ)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)

        lzd = LZMADecompressor(lzma.FORMAT_ALONE)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)

        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)

    def test_decompressor_bug_28275(self):
        # Test coverage for Issue 28275
        lzd = LZMADecompressor()
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_RAW_1)
        # Previously, a second call could crash due to internal inconsistency
        self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_RAW_1)

    # Test that LZMACompressor->LZMADecompressor preserves the input data.

    def test_roundtrip_xz(self):
        lzc = LZMACompressor()
        cdata = lzc.compress(INPUT) + lzc.flush()
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, cdata, lzma.CHECK_CRC64)

    def test_roundtrip_alone(self):
        lzc = LZMACompressor(lzma.FORMAT_ALONE)
        cdata = lzc.compress(INPUT) + lzc.flush()
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, cdata, lzma.CHECK_NONE)

    def test_roundtrip_raw(self):
        lzc = LZMACompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        cdata = lzc.compress(INPUT) + lzc.flush()
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        self._test_decompressor(lzd, cdata, lzma.CHECK_NONE)

    def test_roundtrip_raw_empty(self):
        lzc = LZMACompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        cdata = lzc.compress(INPUT)
        cdata += lzc.compress(b'')
        cdata += lzc.compress(b'')
        cdata += lzc.compress(b'')
        cdata += lzc.flush()
        lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        self._test_decompressor(lzd, cdata, lzma.CHECK_NONE)

    def test_roundtrip_chunks(self):
        lzc = LZMACompressor()
        cdata = []
        for i in range(0, len(INPUT), 10):
            cdata.append(lzc.compress(INPUT[i:i+10]))
        cdata.append(lzc.flush())
        cdata = b"".join(cdata)
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, cdata, lzma.CHECK_CRC64)

    def test_roundtrip_empty_chunks(self):
        lzc = LZMACompressor()
        cdata = []
        for i in range(0, len(INPUT), 10):
            cdata.append(lzc.compress(INPUT[i:i+10]))
            cdata.append(lzc.compress(b''))
            cdata.append(lzc.compress(b''))
            cdata.append(lzc.compress(b''))
        cdata.append(lzc.flush())
        cdata = b"".join(cdata)
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, cdata, lzma.CHECK_CRC64)

    # LZMADecompressor intentionally does not handle concatenated streams.

    def test_decompressor_multistream(self):
        lzd = LZMADecompressor()
        self._test_decompressor(lzd, COMPRESSED_XZ + COMPRESSED_ALONE,
                                lzma.CHECK_CRC64, unused_data=COMPRESSED_ALONE)

    # Test with inputs larger than 4GiB.

    @support.skip_if_pgo_task
    @bigmemtest(size=_4G + 100, memuse=2)
    def test_compressor_bigmem(self, size):
        lzc = LZMACompressor()
        cdata = lzc.compress(b"x" * size) + lzc.flush()
        ddata = lzma.decompress(cdata)
        try:
            self.assertEqual(len(ddata), size)
            self.assertEqual(len(ddata.strip(b"x")), 0)
        finally:
            ddata = None

    @support.skip_if_pgo_task
    @bigmemtest(size=_4G + 100, memuse=3)
    def test_decompressor_bigmem(self, size):
        lzd = LZMADecompressor()
        blocksize = 10 * 1024 * 1024
        block = random.randbytes(blocksize)
        try:
            input = block * (size // blocksize + 1)
            cdata = lzma.compress(input)
            ddata = lzd.decompress(cdata)
            self.assertEqual(ddata, input)
        finally:
            input = cdata = ddata = None

    # Pickling raises an exception; there's no way to serialize an lzma_stream.

    def test_pickle(self):
        for proto in range(pickle.HIGHEST_PROTOCOL + 1):
            with self.assertRaises(TypeError):
                pickle.dumps(LZMACompressor(), proto)
            with self.assertRaises(TypeError):
                pickle.dumps(LZMADecompressor(), proto)

    @support.refcount_test
    def test_refleaks_in_decompressor___init__(self):
        gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
        lzd = LZMADecompressor()
        refs_before = gettotalrefcount()
        for i in range(100):
            lzd.__init__()
        self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)


class CompressDecompressFunctionTestCase(unittest.TestCase):

    # Test error cases:

    def test_bad_args(self):
        self.assertRaises(TypeError, lzma.compress)
        self.assertRaises(TypeError, lzma.compress, [])
        self.assertRaises(TypeError, lzma.compress, b"", format="xz")
        self.assertRaises(TypeError, lzma.compress, b"", check="none")
        self.assertRaises(TypeError, lzma.compress, b"", preset="blah")
        self.assertRaises(TypeError, lzma.compress, b"", filters=1024)
        # Can't specify a preset and a custom filter chain at the same time.
        with self.assertRaises(ValueError):
            lzma.compress(b"", preset=3, filters=[{"id": lzma.FILTER_LZMA2}])

        self.assertRaises(TypeError, lzma.decompress)
        self.assertRaises(TypeError, lzma.decompress, [])
        self.assertRaises(TypeError, lzma.decompress, b"", format="lzma")
        self.assertRaises(TypeError, lzma.decompress, b"", memlimit=7.3e9)
        with self.assertRaises(TypeError):
            lzma.decompress(b"", format=lzma.FORMAT_RAW, filters={})
        # Cannot specify a memory limit with FILTER_RAW.
        with self.assertRaises(ValueError):
            lzma.decompress(b"", format=lzma.FORMAT_RAW, memlimit=0x1000000)
        # Can only specify a custom filter chain with FILTER_RAW.
        with self.assertRaises(ValueError):
            lzma.decompress(b"", filters=FILTERS_RAW_1)
        with self.assertRaises(ValueError):
            lzma.decompress(b"", format=lzma.FORMAT_XZ, filters=FILTERS_RAW_1)
        with self.assertRaises(ValueError):
            lzma.decompress(
                    b"", format=lzma.FORMAT_ALONE, filters=FILTERS_RAW_1)

    def test_decompress_memlimit(self):
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_XZ, memlimit=1024)
        with self.assertRaises(LZMAError):
            lzma.decompress(
                    COMPRESSED_XZ, format=lzma.FORMAT_XZ, memlimit=1024)
        with self.assertRaises(LZMAError):
            lzma.decompress(
                    COMPRESSED_ALONE, format=lzma.FORMAT_ALONE, memlimit=1024)

    # Test LZMADecompressor on known-good input data.

    def test_decompress_good_input(self):
        ddata = lzma.decompress(COMPRESSED_XZ)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(COMPRESSED_ALONE)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(COMPRESSED_XZ, lzma.FORMAT_XZ)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(COMPRESSED_ALONE, lzma.FORMAT_ALONE)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(
                COMPRESSED_RAW_1, lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(
                COMPRESSED_RAW_2, lzma.FORMAT_RAW, filters=FILTERS_RAW_2)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(
                COMPRESSED_RAW_3, lzma.FORMAT_RAW, filters=FILTERS_RAW_3)
        self.assertEqual(ddata, INPUT)

        ddata = lzma.decompress(
                COMPRESSED_RAW_4, lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        self.assertEqual(ddata, INPUT)

    def test_decompress_incomplete_input(self):
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_XZ[:128])
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_ALONE[:128])
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_1[:128],
                          format=lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_2[:128],
                          format=lzma.FORMAT_RAW, filters=FILTERS_RAW_2)
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_3[:128],
                          format=lzma.FORMAT_RAW, filters=FILTERS_RAW_3)
        self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_4[:128],
                          format=lzma.FORMAT_RAW, filters=FILTERS_RAW_4)

    def test_decompress_bad_input(self):
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_BOGUS)
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_RAW_1)
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_ALONE, format=lzma.FORMAT_XZ)
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_XZ, format=lzma.FORMAT_ALONE)
        with self.assertRaises(LZMAError):
            lzma.decompress(COMPRESSED_XZ, format=lzma.FORMAT_RAW,
                            filters=FILTERS_RAW_1)

    # Test that compress()->decompress() preserves the input data.

    def test_roundtrip(self):
        cdata = lzma.compress(INPUT)
        ddata = lzma.decompress(cdata)
        self.assertEqual(ddata, INPUT)

        cdata = lzma.compress(INPUT, lzma.FORMAT_XZ)
        ddata = lzma.decompress(cdata)
        self.assertEqual(ddata, INPUT)

        cdata = lzma.compress(INPUT, lzma.FORMAT_ALONE)
        ddata = lzma.decompress(cdata)
        self.assertEqual(ddata, INPUT)

        cdata = lzma.compress(INPUT, lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        ddata = lzma.decompress(cdata, lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
        self.assertEqual(ddata, INPUT)

    # Unlike LZMADecompressor, decompress() *does* handle concatenated streams.

    def test_decompress_multistream(self):
        ddata = lzma.decompress(COMPRESSED_XZ + COMPRESSED_ALONE)
        self.assertEqual(ddata, INPUT * 2)

    # Test robust handling of non-LZMA data following the compressed stream(s).

    def test_decompress_trailing_junk(self):
        ddata = lzma.decompress(COMPRESSED_XZ + COMPRESSED_BOGUS)
        self.assertEqual(ddata, INPUT)

    def test_decompress_multistream_trailing_junk(self):
        ddata = lzma.decompress(COMPRESSED_XZ * 3 + COMPRESSED_BOGUS)
        self.assertEqual(ddata, INPUT * 3)


class TempFile:
    """Context manager - creates a file, and deletes it on __exit__."""

    def __init__(self, filename, data=b""):
        self.filename = filename
        self.data = data

    def __enter__(self):
        with open(self.filename, "wb") as f:
            f.write(self.data)

    def __exit__(self, *args):
        unlink(self.filename)


class FileTestCase(unittest.TestCase):

    def test_init(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            pass
        with LZMAFile(BytesIO(), "w") as f:
            pass
        with LZMAFile(BytesIO(), "x") as f:
            pass
        with LZMAFile(BytesIO(), "a") as f:
            pass

    def test_init_with_PathLike_filename(self):
        filename = pathlib.Path(TESTFN)
        with TempFile(filename, COMPRESSED_XZ):
            with LZMAFile(filename) as f:
                self.assertEqual(f.read(), INPUT)
            with LZMAFile(filename, "a") as f:
                f.write(INPUT)
            with LZMAFile(filename) as f:
                self.assertEqual(f.read(), INPUT * 2)

    def test_init_with_filename(self):
        with TempFile(TESTFN, COMPRESSED_XZ):
            with LZMAFile(TESTFN) as f:
                pass
            with LZMAFile(TESTFN, "w") as f:
                pass
            with LZMAFile(TESTFN, "a") as f:
                pass

    def test_init_mode(self):
        with TempFile(TESTFN):
            with LZMAFile(TESTFN, "r"):
                pass
            with LZMAFile(TESTFN, "rb"):
                pass
            with LZMAFile(TESTFN, "w"):
                pass
            with LZMAFile(TESTFN, "wb"):
                pass
            with LZMAFile(TESTFN, "a"):
                pass
            with LZMAFile(TESTFN, "ab"):
                pass

    def test_init_with_x_mode(self):
        self.addCleanup(unlink, TESTFN)
        for mode in ("x", "xb"):
            unlink(TESTFN)
            with LZMAFile(TESTFN, mode):
                pass
            with self.assertRaises(FileExistsError):
                with LZMAFile(TESTFN, mode):
                    pass

    def test_init_bad_mode(self):
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), (3, "x"))
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "xt")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "x+")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "rx")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "wx")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "rt")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "r+")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "wt")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "w+")
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), "rw")

    def test_init_bad_check(self):
        with self.assertRaises(TypeError):
            LZMAFile(BytesIO(), "w", check=b"asd")
        # CHECK_UNKNOWN and anything above CHECK_ID_MAX should be invalid.
        with self.assertRaises(LZMAError):
            LZMAFile(BytesIO(), "w", check=lzma.CHECK_UNKNOWN)
        with self.assertRaises(LZMAError):
            LZMAFile(BytesIO(), "w", check=lzma.CHECK_ID_MAX + 3)
        # Cannot specify a check with mode="r".
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_NONE)
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC32)
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC64)
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_SHA256)
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_UNKNOWN)

    def test_init_bad_preset(self):
        with self.assertRaises(TypeError):
            LZMAFile(BytesIO(), "w", preset=4.39)
        with self.assertRaises(LZMAError):
            LZMAFile(BytesIO(), "w", preset=10)
        with self.assertRaises(LZMAError):
            LZMAFile(BytesIO(), "w", preset=23)
        with self.assertRaises(OverflowError):
            LZMAFile(BytesIO(), "w", preset=-1)
        with self.assertRaises(OverflowError):
            LZMAFile(BytesIO(), "w", preset=-7)
        with self.assertRaises(TypeError):
            LZMAFile(BytesIO(), "w", preset="foo")
        # Cannot specify a preset with mode="r".
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(COMPRESSED_XZ), preset=3)

    def test_init_bad_filter_spec(self):
        with self.assertRaises(TypeError):
            LZMAFile(BytesIO(), "w", filters=[b"wobsite"])
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w", filters=[{"xyzzy": 3}])
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w", filters=[{"id": 98765}])
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w",
                     filters=[{"id": lzma.FILTER_LZMA2, "foo": 0}])
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w",
                     filters=[{"id": lzma.FILTER_DELTA, "foo": 0}])
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w",
                     filters=[{"id": lzma.FILTER_X86, "foo": 0}])

    def test_init_with_preset_and_filters(self):
        with self.assertRaises(ValueError):
            LZMAFile(BytesIO(), "w", format=lzma.FORMAT_RAW,
                     preset=6, filters=FILTERS_RAW_1)

    def test_close(self):
        with BytesIO(COMPRESSED_XZ) as src:
            f = LZMAFile(src)
            f.close()
            # LZMAFile.close() should not close the underlying file object.
            self.assertFalse(src.closed)
            # Try closing an already-closed LZMAFile.
            f.close()
            self.assertFalse(src.closed)

        # Test with a real file on disk, opened directly by LZMAFile.
        with TempFile(TESTFN, COMPRESSED_XZ):
            f = LZMAFile(TESTFN)
            fp = f._fp
            f.close()
            # Here, LZMAFile.close() *should* close the underlying file object.
            self.assertTrue(fp.closed)
            # Try closing an already-closed LZMAFile.
            f.close()

    def test_closed(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        try:
            self.assertFalse(f.closed)
            f.read()
            self.assertFalse(f.closed)
        finally:
            f.close()
        self.assertTrue(f.closed)

        f = LZMAFile(BytesIO(), "w")
        try:
            self.assertFalse(f.closed)
        finally:
            f.close()
        self.assertTrue(f.closed)

    def test_fileno(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        try:
            self.assertRaises(UnsupportedOperation, f.fileno)
        finally:
            f.close()
        self.assertRaises(ValueError, f.fileno)
        with TempFile(TESTFN, COMPRESSED_XZ):
            f = LZMAFile(TESTFN)
            try:
                self.assertEqual(f.fileno(), f._fp.fileno())
                self.assertIsInstance(f.fileno(), int)
            finally:
                f.close()
        self.assertRaises(ValueError, f.fileno)

    def test_seekable(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        try:
            self.assertTrue(f.seekable())
            f.read()
            self.assertTrue(f.seekable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.seekable)

        f = LZMAFile(BytesIO(), "w")
        try:
            self.assertFalse(f.seekable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.seekable)

        src = BytesIO(COMPRESSED_XZ)
        src.seekable = lambda: False
        f = LZMAFile(src)
        try:
            self.assertFalse(f.seekable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.seekable)

    def test_readable(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        try:
            self.assertTrue(f.readable())
            f.read()
            self.assertTrue(f.readable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.readable)

        f = LZMAFile(BytesIO(), "w")
        try:
            self.assertFalse(f.readable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.readable)

    def test_writable(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        try:
            self.assertFalse(f.writable())
            f.read()
            self.assertFalse(f.writable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.writable)

        f = LZMAFile(BytesIO(), "w")
        try:
            self.assertTrue(f.writable())
        finally:
            f.close()
        self.assertRaises(ValueError, f.writable)

    def test_read(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_ALONE)) as f:
            self.assertEqual(f.read(), INPUT)
        with LZMAFile(BytesIO(COMPRESSED_XZ), format=lzma.FORMAT_XZ) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_ALONE), format=lzma.FORMAT_ALONE) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_RAW_1),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_1) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_RAW_2),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_2) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_RAW_3),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_3) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")
        with LZMAFile(BytesIO(COMPRESSED_RAW_4),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_4) as f:
            self.assertEqual(f.read(), INPUT)
            self.assertEqual(f.read(), b"")

    def test_read_0(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertEqual(f.read(0), b"")
        with LZMAFile(BytesIO(COMPRESSED_ALONE)) as f:
            self.assertEqual(f.read(0), b"")
        with LZMAFile(BytesIO(COMPRESSED_XZ), format=lzma.FORMAT_XZ) as f:
            self.assertEqual(f.read(0), b"")
        with LZMAFile(BytesIO(COMPRESSED_ALONE), format=lzma.FORMAT_ALONE) as f:
            self.assertEqual(f.read(0), b"")

    def test_read_10(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            chunks = []
            while True:
                result = f.read(10)
                if not result:
                    break
                self.assertLessEqual(len(result), 10)
                chunks.append(result)
            self.assertEqual(b"".join(chunks), INPUT)

    def test_read_multistream(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ * 5)) as f:
            self.assertEqual(f.read(), INPUT * 5)
        with LZMAFile(BytesIO(COMPRESSED_XZ + COMPRESSED_ALONE)) as f:
            self.assertEqual(f.read(), INPUT * 2)
        with LZMAFile(BytesIO(COMPRESSED_RAW_3 * 4),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_3) as f:
            self.assertEqual(f.read(), INPUT * 4)

    def test_read_multistream_buffer_size_aligned(self):
        # Test the case where a stream boundary coincides with the end
        # of the raw read buffer.
        saved_buffer_size = _compression.BUFFER_SIZE
        _compression.BUFFER_SIZE = len(COMPRESSED_XZ)
        try:
            with LZMAFile(BytesIO(COMPRESSED_XZ *  5)) as f:
                self.assertEqual(f.read(), INPUT * 5)
        finally:
            _compression.BUFFER_SIZE = saved_buffer_size

    def test_read_trailing_junk(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ + COMPRESSED_BOGUS)) as f:
            self.assertEqual(f.read(), INPUT)

    def test_read_multistream_trailing_junk(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ * 5 + COMPRESSED_BOGUS)) as f:
            self.assertEqual(f.read(), INPUT * 5)

    def test_read_from_file(self):
        with TempFile(TESTFN, COMPRESSED_XZ):
            with LZMAFile(TESTFN) as f:
                self.assertEqual(f.read(), INPUT)
                self.assertEqual(f.read(), b"")

    def test_read_from_file_with_bytes_filename(self):
        try:
            bytes_filename = TESTFN.encode("ascii")
        except UnicodeEncodeError:
            self.skipTest("Temporary file name needs to be ASCII")
        with TempFile(TESTFN, COMPRESSED_XZ):
            with LZMAFile(bytes_filename) as f:
                self.assertEqual(f.read(), INPUT)
                self.assertEqual(f.read(), b"")

    def test_read_incomplete(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ[:128])) as f:
            self.assertRaises(EOFError, f.read)

    def test_read_truncated(self):
        # Drop stream footer: CRC (4 bytes), index size (4 bytes),
        # flags (2 bytes) and magic number (2 bytes).
        truncated = COMPRESSED_XZ[:-12]
        with LZMAFile(BytesIO(truncated)) as f:
            self.assertRaises(EOFError, f.read)
        with LZMAFile(BytesIO(truncated)) as f:
            self.assertEqual(f.read(len(INPUT)), INPUT)
            self.assertRaises(EOFError, f.read, 1)
        # Incomplete 12-byte header.
        for i in range(12):
            with LZMAFile(BytesIO(truncated[:i])) as f:
                self.assertRaises(EOFError, f.read, 1)

    def test_read_bad_args(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        f.close()
        self.assertRaises(ValueError, f.read)
        with LZMAFile(BytesIO(), "w") as f:
            self.assertRaises(ValueError, f.read)
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertRaises(TypeError, f.read, float())

    def test_read_bad_data(self):
        with LZMAFile(BytesIO(COMPRESSED_BOGUS)) as f:
            self.assertRaises(LZMAError, f.read)

    def test_read1(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            blocks = []
            while True:
                result = f.read1()
                if not result:
                    break
                blocks.append(result)
            self.assertEqual(b"".join(blocks), INPUT)
            self.assertEqual(f.read1(), b"")

    def test_read1_0(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertEqual(f.read1(0), b"")

    def test_read1_10(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            blocks = []
            while True:
                result = f.read1(10)
                if not result:
                    break
                blocks.append(result)
            self.assertEqual(b"".join(blocks), INPUT)
            self.assertEqual(f.read1(), b"")

    def test_read1_multistream(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ * 5)) as f:
            blocks = []
            while True:
                result = f.read1()
                if not result:
                    break
                blocks.append(result)
            self.assertEqual(b"".join(blocks), INPUT * 5)
            self.assertEqual(f.read1(), b"")

    def test_read1_bad_args(self):
        f = LZMAFile(BytesIO(COMPRESSED_XZ))
        f.close()
        self.assertRaises(ValueError, f.read1)
        with LZMAFile(BytesIO(), "w") as f:
            self.assertRaises(ValueError, f.read1)
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertRaises(TypeError, f.read1, None)

    def test_peek(self):
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            result = f.peek()
            self.assertGreater(len(result), 0)
            self.assertTrue(INPUT.startswith(result))
            self.assertEqual(f.read(), INPUT)
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            result = f.peek(10)
            self.assertGreater(len(result), 0)
            self.assertTrue(INPUT.startswith(result))
            self.assertEqual(f.read(), INPUT)

    def test_peek_bad_args(self):
        with LZMAFile(BytesIO(), "w") as f:
            self.assertRaises(ValueError, f.peek)

    def test_iterator(self):
        with BytesIO(INPUT) as f:
            lines = f.readlines()
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertListEqual(list(iter(f)), lines)
        with LZMAFile(BytesIO(COMPRESSED_ALONE)) as f:
            self.assertListEqual(list(iter(f)), lines)
        with LZMAFile(BytesIO(COMPRESSED_XZ), format=lzma.FORMAT_XZ) as f:
            self.assertListEqual(list(iter(f)), lines)
        with LZMAFile(BytesIO(COMPRESSED_ALONE), format=lzma.FORMAT_ALONE) as f:
            self.assertListEqual(list(iter(f)), lines)
        with LZMAFile(BytesIO(COMPRESSED_RAW_2),
                      format=lzma.FORMAT_RAW, filters=FILTERS_RAW_2) as f:
            self.assertListEqual(list(iter(f)), lines)

    def test_readline(self):
        with BytesIO(INPUT) as f:
            lines = f.readlines()
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            for line in lines:
                self.assertEqual(f.readline(), line)

    def test_readlines(self):
        with BytesIO(INPUT) as f:
            lines = f.readlines()
        with LZMAFile(BytesIO(COMPRESSED_XZ)) as f:
            self.assertListEqual(f.readlines(), lines)

    def test_decompress_limited(self):
        """Decompressed data buffering should be limited"""
        bomb = lzma.compress(b'\0' * int(2e6), preset=6)
        self.assertLess(len(bomb), _compression.BUFFER_SIZE)

        decomp = LZMAFile(BytesIO(bomb))
        self.assertEqual(decomp.read(1), b'\0')
        max_decomp = 1 + DEFAULT_BUFFER_SIZE
        self.assertLessEqual(decomp._buffer.raw.tell(), max_decomp,
            "Excessive amount of data was decompressed")

    def test_write(self):
        with BytesIO() as dst:
            with LZMAFile(dst, "w") as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT)
            self.assertEqual(dst.getvalue(), expected)
        with BytesIO() as dst:
            with LZMAFile(dst, "w", format=lzma.FORMAT_XZ) as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT, format=lzma.FORMAT_XZ)
            self.assertEqual(dst.getvalue(), expected)
        with BytesIO() as dst:
            with LZMAFile(dst, "w", format=lzma.FORMAT_ALONE) as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT, format=lzma.FORMAT_ALONE)
            self.assertEqual(dst.getvalue(), expected)
        with BytesIO() as dst:
            with LZMAFile(dst, "w", format=lzma.FORMAT_RAW,
                          filters=FILTERS_RAW_2) as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT, format=lzma.FORMAT_RAW,
                                     filters=FILTERS_RAW_2)
            self.assertEqual(dst.getvalue(), expected)

    def test_write_10(self):
        with BytesIO() as dst:
            with LZMAFile(dst, "w") as f:
                for start in range(0, len(INPUT), 10):
                    f.write(INPUT[start:start+10])
            expected = lzma.compress(INPUT)
            self.assertEqual(dst.getvalue(), expected)

    def test_write_append(self):
        part1 = INPUT[:1024]
        part2 = INPUT[1024:1536]
        part3 = INPUT[1536:]
        expected = b"".join(lzma.compress(x) for x in (part1, part2, part3))
        with BytesIO() as dst:
            with LZMAFile(dst, "w") as f:
                f.write(part1)
            with LZMAFile(dst, "a") as f:
                f.write(part2)
            with LZMAFile(dst, "a") as f:
                f.write(part3)
            self.assertEqual(dst.getvalue(), expected)

    def test_write_to_file(self):
        try:
            with LZMAFile(TESTFN, "w") as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT)
            with open(TESTFN, "rb") as f:
                self.assertEqual(f.read(), expected)
        finally:
            unlink(TESTFN)

    def test_write_to_file_with_bytes_filename(self):
        try:
            bytes_filename = TESTFN.encode("ascii")
        except UnicodeEncodeError:
            self.skipTest("Temporary file name needs to be ASCII")
        try:
            with LZMAFile(bytes_filename, "w") as f:
                f.write(INPUT)
            expected = lzma.compress(INPUT)
            with open(TESTFN, "rb") as f:
                self.assertEqual(f.read(), expected)
        finally:
            unlink(TESTFN)

    def test_write_append_to_file(self):
        part1 = INPUT[:1024]
        part2 = INPUT[1024:1536]
        part3 = INPUT[1536:]
        expected = b"".join(lzma.compress(x) for x in (part1, part2, part3))
        try:
            with LZMAFile(TESTFN, "w") as f:
                f.write(part1)
            with LZMAFile(TESTFN, "a") as f:
                f.write(part2)
            with LZMAFile(TESTFN, "a") as f:
                f.write(part3)