test script for the gzip module data2 b zlibmodule c gzipcompatible data compression see http www gzip orgzlib see http www winimage comzlibdll for windows try flush and fileno test multiple close calls the following testwritexy methods test that write accepts the corresponding byteslike object type as input and that the data written equals bytesxy in all cases test that nonbyteslike types raise typeerror issue 21560 attempts to write incompatible types should not affect the state of the fileobject try reading check that position was updated correctly see issue10791 read chunk size over uintmax should be supported despite zlib s limitation per lowlevel call test that io operations on closed gzipfile objects raise a valueerror just like the corresponding functions on file objects write to a file open it for reading then close it open the file for writing then close it append to the previous file bug 1074261 was triggered when reading a file that contained many many members create such a file and verify that reading it works try reading the file issue 7471 a gzipfile can be wrapped in a bufferedreader for performance try readline with varying line lengths try readlines try seek read test try seekwhence1 read test try seek write test see rfc 1952 http www faqs orgrfcsrfc1952 html since the fname flag is set the zeroterminated filename follows rfc 1952 specifies that this is the name of the input file if any however the gzip module defaults to storing the name of the output file in this field since no other flags were set the header ends here rather than process the compressed data let s seek to the trailer see rfc 1952 http www faqs orgrfcsrfc1952 html specifically discussion of xfl in section 2 3 1 gzipfile supports the context management protocol pad the file with zeroes issue 10791 textiowrapper readlines fails when wrapping gzipfile issue 13781 opening a gzipfile for writing fails when using a fileobj created with os fdopen sanity check that we are actually operating on the right file decompressed data buffering should be limited bomb gzip compressb 0 int2e6 compresslevel9 self assertlesslenbomb io defaultbuffersize bomb io bytesiobomb decomp gzip gzipfilefileobjbomb self assertequaldecomp read1 b 0 maxdecomp 1 io defaultbuffersize self assertlessequaldecomp buffer raw tell maxdecomp excessive amount of data was decompressed testing compressdecompress shortcut functions def testcompressself for data in data1 data2 for args in 1 6 9 datac gzip compressdata args self assertequaltypedatac bytes with gzip gzipfilefileobjio bytesiodatac moderb as f self assertequalf read data def testcompressmtimeself mtime 123456789 for data in data1 data2 for args in 1 6 9 with self subtestdatadata argsargs datac gzip compressdata args mtimemtime self assertequaltypedatac bytes with gzip gzipfilefileobjio bytesiodatac moderb as f f read1 to set mtime attribute self assertequalf mtime mtime def testcompresscorrectlevelself gzip compress calls with mtime 0 take a different code path for mtime in 0 42 with self subtestmtimemtime nocompress gzip compressdata1 compresslevel0 mtimemtime yescompress gzip compressdata1 compresslevel1 mtimemtime self assertindata1 nocompress self assertnotindata1 yescompress def testdecompressself for data in data1 data2 buf io bytesio with gzip gzipfilefileobjbuf modewb as f f writedata self assertequalgzip decompressbuf getvalue data roundtrip with compress datac gzip compressdata self assertequalgzip decompressdatac data def testdecompresstruncatedtrailerself compresseddata gzip compressdata1 self assertraiseseoferror gzip decompress compresseddata 4 def testdecompressmissingtrailerself compresseddata gzip compressdata1 self assertraiseseoferror gzip decompress compresseddata 8 def testreadtruncatedself data data150 drop the crc 4 bytes and file size 4 bytes truncated gzip compressdata 8 with gzip gzipfilefileobjio bytesiotruncated as f self assertraiseseoferror f read with gzip gzipfilefileobjio bytesiotruncated as f self assertequalf readlendata data self assertraiseseoferror f read 1 incomplete 10byte header for i in range2 10 with gzip gzipfilefileobjio bytesiotruncated i as f self assertraiseseoferror f read 1 def testreadwithextraself gzip data with an extra field gzdata b x1fx8bx08x04xb2x17cqx02xff b x05x00extra b x0bi x01x002xd1mxx04x00x00x00 with gzip gzipfilefileobjio bytesiogzdata as f self assertequalf read b test def testprependerrorself see issue 20875 with gzip openself filename wb as f f writedata1 with gzip openself filename rb as f f buffer raw fp prepend def testissue44439self q array array q 1 2 3 4 5 length lenq q itemsize with gzip gzipfilefileobjio bytesio mode w as f self assertequalf writeq length self assertequalf tell length def testflushflushescompressorself see issue gh105808 b io bytesio message bimportant message here with gzip gzipfilefileobjb mode w as f f writemessage f flush partialdata b getvalue fulldata b getvalue self assertequalgzip decompressfulldata message the partial data should contain the gzip header and the complete message but not the endofstream markers so we can t just decompress it directly with self assertraiseseoferror gzip decompresspartialdata d zlib decompressobjwbitszlib maxwbits f io bytesiopartialdata gzip readgzipheaderf readmessage d decompressf read self assertequalreadmessage message def testflushmodesself make sure the argument to flush is properly passed to the zlib compressobj see issue gh105808 class fakecompressor def initself self modes def compressself data return b def flushself mode1 self modes appendmode return b b io bytesio fc fakecompressor with gzip gzipfilefileobjb mode w as f f compress fc f flush f flush50 f flushzlibmode100 the implicit close will also flush the compressor expectedmodes zlib zsyncflush 50 100 1 self assertequalfc modes expectedmodes def testwriteseekwriteself make sure that offset is uptodate before seeking see issue gh108111 b io bytesio message bimportant message here with gzip gzipfilefileobjb mode w as f f writemessage f seeklenmessage f writemessage data b getvalue self assertequalgzip decompressdata message 2 class testopenbasetest def testbinarymodesself uncompressed data1 50 with gzip openself filename wb as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed with gzip openself filename rb as f self assertequalf read uncompressed with gzip openself filename ab as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed 2 with self assertraisesfileexistserror gzip openself filename xb oshelper unlinkself filename with gzip openself filename xb as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed def testpathlikefileself filename pathlib pathself filename with gzip openfilename wb as f f writedata1 50 with gzip openfilename ab as f f writedata1 with gzip openfilename as f self assertequalf read data1 51 def testimplicitbinarymodesself test implicit binary modes no b or t in mode string uncompressed data1 50 with gzip openself filename w as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed with gzip openself filename r as f self assertequalf read uncompressed with gzip openself filename a as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed 2 with self assertraisesfileexistserror gzip openself filename x oshelper unlinkself filename with gzip openself filename x as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read self assertequalfiledata uncompressed def testtextmodesself uncompressed data1 decodeascii 50 uncompressedraw uncompressed replacen os linesep with gzip openself filename wt encodingascii as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read decodeascii self assertequalfiledata uncompressedraw with gzip openself filename rt encodingascii as f self assertequalf read uncompressed with gzip openself filename at encodingascii as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read decodeascii self assertequalfiledata uncompressedraw 2 def testfileobjself uncompressedbytes data1 50 uncompressedstr uncompressedbytes decodeascii compressed gzip compressuncompressedbytes with gzip openio bytesiocompressed r as f self assertequalf read uncompressedbytes with gzip openio bytesiocompressed rb as f self assertequalf read uncompressedbytes with gzip openio bytesiocompressed rt encodingascii as f self assertequalf read uncompressedstr def testbadparamsself test invalid parameter combinations with self assertraisestypeerror gzip open123 456 with self assertraisesvalueerror gzip openself filename wbt with self assertraisesvalueerror gzip openself filename xbt with self assertraisesvalueerror gzip openself filename rb encodingutf8 with self assertraisesvalueerror gzip openself filename rb errorsignore with self assertraisesvalueerror gzip openself filename rb newlinen def testencodingself test nondefault encoding uncompressed data1 decodeascii 50 uncompressedraw uncompressed replacen os linesep with gzip openself filename wt encodingutf16 as f f writeuncompressed with openself filename rb as f filedata gzip decompressf read decodeutf16 self assertequalfiledata uncompressedraw with gzip openself filename rt encodingutf16 as f self assertequalf read uncompressed def testencodingerrorhandlerself test with nondefault encoding error handler with gzip openself filename wb as f f writebfooxffbar with gzip openself filename rt encodingascii errorsignore as f self assertequalf read foobar def testnewlineself test with explicit newline universal newline mode disabled uncompressed data1 decodeascii 50 with gzip openself filename wt encodingascii newlinen as f f writeuncompressed with gzip openself filename rt encodingascii newliner as f self assertequalf readlines uncompressed def createandremovedirectorydirectory def decoratorfunction functools wrapsfunction def wrapperargs kwargs os makedirsdirectory try return functionargs kwargs finally oshelper rmtreedirectory return wrapper return decorator class testcommandlineunittest testcase data b this is a simple test with gzip requiressubprocess def testdecompressstdinstdoutself with io bytesio as bytesio with gzip gzipfilefileobjbytesio mode wb as gzipfile gzipfile writeself data args sys executable m gzip d with popenargs stdinpipe stdoutpipe stderrpipe as proc out err proc communicatebytesio getvalue self assertequalerr b self assertequalout self data createandremovedirectorytempdir def testdecompressinfileoutfileself gzipname os path jointempdir testgzip gz self assertfalseos path existsgzipname with gzip opengzipname mode wb as fp fp writeself data rc out err assertpythonok m gzip d gzipname with openos path jointempdir testgzip rb as gunziped self assertequalgunziped read self data self asserttrueos path existsgzipname self assertequalrc 0 self assertequalout b self assertequalerr b def testdecompressinfileoutfileerrorself rc out err assertpythonfailure m gzip d thisisatest out self assertequalbfilename doesn t end in gz thisisatest out err strip self assertequalrc 1 self assertequalout b requiressubprocess createandremovedirectorytempdir def testcompressstdinoutfileself args sys executable m gzip with popenargs stdinpipe stdoutpipe stderrpipe as proc out err proc communicateself data self assertequalerr b self assertequalout 2 bx1fx8b createandremovedirectorytempdir def testcompressinfileoutfiledefaultself localtestgzip os path jointempdir testgzip gzipname localtestgzip gz self assertfalseos path existsgzipname with openlocaltestgzip wb as fp fp writeself data rc out err assertpythonok m gzip localtestgzip self asserttrueos path existsgzipname self assertequalout b self assertequalerr b createandremovedirectorytempdir def testcompressinfileoutfileself for compresslevel in fast best with self subtestcompresslevelcompresslevel localtestgzip os path jointempdir testgzip gzipname localtestgzip gz self assertfalseos path existsgzipname with openlocaltestgzip wb as fp fp writeself data rc out err assertpythonok m gzip compresslevel localtestgzip self asserttrueos path existsgzipname self assertequalout b self assertequalerr b os removegzipname self assertfalseos path existsgzipname def testcompressfastbestareexclusiveself rc out err assertpythonfailure m gzip fast best self assertinberror argument best not allowed with argument fast err self assertequalout b def testdecompresscannothaveflagscompressionself rc out err assertpythonfailure m gzip fast d self assertinb error argument ddecompress not allowed with argument fast err self assertequalout b if name main unittest main int length defaultalloc err z_ok pyobject retval int flushmode z_finish unsigned long start_total_out zlibmodule c gzip compatible data compression see http www gzip org zlib see http www winimage com zlibdll for windows try flush and fileno test multiple close calls the following test_write_xy methods test that write accepts the corresponding bytes like object type as input and that the data written equals bytes xy in all cases test that non bytes like types raise typeerror issue 21560 attempts to write incompatible types should not affect the state of the fileobject try reading check that position was updated correctly see issue10791 read chunk size over uint_max should be supported despite zlib s limitation per low level call test that i o operations on closed gzipfile objects raise a valueerror just like the corresponding functions on file objects write to a file open it for reading then close it open the file for writing then close it append to the previous file bug 1074261 was triggered when reading a file that contained many many members create such a file and verify that reading it works append try reading the file issue 7471 a gzipfile can be wrapped in a bufferedreader for performance try readline with varying line lengths try readlines try seek read test negative seek positive seek try seek whence 1 read test try seek write test see rfc 1952 http www faqs org rfcs rfc1952 html gzip id deflate only the fname flag is set little endian maximum compression os unknown os independent since the fname flag is set the zero terminated filename follows rfc 1952 specifies that this is the name of the input file if any however the gzip module defaults to storing the name of the output file in this field since no other flags were set the header ends here rather than process the compressed data let s seek to the trailer crc32 of uncompressed data data1 see rfc 1952 http www faqs org rfcs rfc1952 html specifically discussion of xfl in section 2 3 1 gzipfile supports the context management protocol pad the file with zeroes issue 10791 textiowrapper readlines fails when wrapping gzipfile issue 13781 opening a gzipfile for writing fails when using a fileobj created with os fdopen sanity check that we are actually operating on the right file decompressed data buffering should be limited testing compress decompress shortcut functions to set mtime attribute gzip compress calls with mtime 0 take a different code path roundtrip with compress drop the crc 4 bytes and file size 4 bytes incomplete 10 byte header gzip data with an extra field see issue 20875 see issue gh 105808 the partial data should contain the gzip header and the complete message but not the end of stream markers so we can t just decompress it directly make sure the argument to flush is properly passed to the zlib compressobj see issue gh 105808 the implicit close will also flush the compressor make sure that offset is up to date before seeking see issue gh 108111 test implicit binary modes no b or t in mode string test invalid parameter combinations test non default encoding test with non default encoding error handler test with explicit newline universal newline mode disabled
import array
import functools
import io
import os
import pathlib
import struct
import sys
import unittest
from subprocess import PIPE, Popen
from test.support import import_helper
from test.support import os_helper
from test.support import _4G, bigmemtest, requires_subprocess
from test.support.script_helper import assert_python_ok, assert_python_failure

gzip = import_helper.import_module('gzip')
zlib = import_helper.import_module('zlib')

data1 = b

data2 = b


TEMPDIR = os.path.abspath(os_helper.TESTFN) + '-gzdir'


class UnseekableIO(io.BytesIO):
    def seekable(self):
        return False

    def tell(self):
        raise io.UnsupportedOperation

    def seek(self, *args):
        raise io.UnsupportedOperation


class BaseTest(unittest.TestCase):
    filename = os_helper.TESTFN

    def setUp(self):
        os_helper.unlink(self.filename)

    def tearDown(self):
        os_helper.unlink(self.filename)


class TestGzip(BaseTest):
    def write_and_read_back(self, data, mode='b'):
        b_data = bytes(data)
        with gzip.GzipFile(self.filename, 'w'+mode) as f:
            l = f.write(data)
        self.assertEqual(l, len(b_data))
        with gzip.GzipFile(self.filename, 'r'+mode) as f:
            self.assertEqual(f.read(), b_data)

    def test_write(self):
        with gzip.GzipFile(self.filename, 'wb') as f:
            f.write(data1 * 50)

            
            f.flush()
            f.fileno()
            if hasattr(os, 'fsync'):
                os.fsync(f.fileno())
            f.close()

        
        f.close()

    def test_write_read_with_pathlike_file(self):
        filename = pathlib.Path(self.filename)
        with gzip.GzipFile(filename, 'w') as f:
            f.write(data1 * 50)
        self.assertIsInstance(f.name, str)
        with gzip.GzipFile(filename, 'a') as f:
            f.write(data1)
        with gzip.GzipFile(filename) as f:
            d = f.read()
        self.assertEqual(d, data1 * 51)
        self.assertIsInstance(f.name, str)

    
    
    
    def test_write_memoryview(self):
        self.write_and_read_back(memoryview(data1 * 50))
        m = memoryview(bytes(range(256)))
        data = m.cast('B', shape=[8,8,4])
        self.write_and_read_back(data)

    def test_write_bytearray(self):
        self.write_and_read_back(bytearray(data1 * 50))

    def test_write_array(self):
        self.write_and_read_back(array.array('I', data1 * 40))

    def test_write_incompatible_type(self):
        
        
        
        with gzip.GzipFile(self.filename, 'wb') as f:
            with self.assertRaises(TypeError):
                f.write('')
            with self.assertRaises(TypeError):
                f.write([])
            f.write(data1)
        with gzip.GzipFile(self.filename, 'rb') as f:
            self.assertEqual(f.read(), data1)

    def test_read(self):
        self.test_write()
        
        with gzip.GzipFile(self.filename, 'r') as f:
            d = f.read()
        self.assertEqual(d, data1*50)

    def test_read1(self):
        self.test_write()
        blocks = []
        nread = 0
        with gzip.GzipFile(self.filename, 'r') as f:
            while True:
                d = f.read1()
                if not d:
                    break
                blocks.append(d)
                nread += len(d)
                
                self.assertEqual(f.tell(), nread)
        self.assertEqual(b''.join(blocks), data1 * 50)

    @bigmemtest(size=_4G, memuse=1)
    def test_read_large(self, size):
        
        
        compressed = gzip.compress(data1, compresslevel=1)
        f = gzip.GzipFile(fileobj=io.BytesIO(compressed), mode='rb')
        self.assertEqual(f.read(size), data1)

    def test_io_on_closed_object(self):
        
        

        
        self.test_write()
        f = gzip.GzipFile(self.filename, 'r')
        fileobj = f.fileobj
        self.assertFalse(fileobj.closed)
        f.close()
        self.assertTrue(fileobj.closed)
        with self.assertRaises(ValueError):
            f.read(1)
        with self.assertRaises(ValueError):
            f.seek(0)
        with self.assertRaises(ValueError):
            f.tell()
        
        f = gzip.GzipFile(self.filename, 'w')
        fileobj = f.fileobj
        self.assertFalse(fileobj.closed)
        f.close()
        self.assertTrue(fileobj.closed)
        with self.assertRaises(ValueError):
            f.write(b'')
        with self.assertRaises(ValueError):
            f.flush()

    def test_append(self):
        self.test_write()
        
        with gzip.GzipFile(self.filename, 'ab') as f:
            f.write(data2 * 15)

        with gzip.GzipFile(self.filename, 'rb') as f:
            d = f.read()
        self.assertEqual(d, (data1*50) + (data2*15))

    def test_many_append(self):
        
        
        
        with gzip.GzipFile(self.filename, 'wb', 9) as f:
            f.write(b'a')
        for i in range(0, 200):
            with gzip.GzipFile(self.filename, "ab", 9) as f: 
                f.write(b'a')

        
        with gzip.GzipFile(self.filename, "rb") as zgfile:
            contents = b""
            while 1:
                ztxt = zgfile.read(8192)
                contents += ztxt
                if not ztxt: break
        self.assertEqual(contents, b'a'*201)

    def test_exclusive_write(self):
        with gzip.GzipFile(self.filename, 'xb') as f:
            f.write(data1 * 50)
        with gzip.GzipFile(self.filename, 'rb') as f:
            self.assertEqual(f.read(), data1 * 50)
        with self.assertRaises(FileExistsError):
            gzip.GzipFile(self.filename, 'xb')

    def test_buffered_reader(self):
        
        
        self.test_write()

        with gzip.GzipFile(self.filename, 'rb') as f:
            with io.BufferedReader(f) as r:
                lines = [line for line in r]

        self.assertEqual(lines, 50 * data1.splitlines(keepends=True))

    def test_readline(self):
        self.test_write()
        

        with gzip.GzipFile(self.filename, 'rb') as f:
            line_length = 0
            while 1:
                L = f.readline(line_length)
                if not L and line_length != 0: break
                self.assertTrue(len(L) <= line_length)
                line_length = (line_length + 1) % 50

    def test_readlines(self):
        self.test_write()
        

        with gzip.GzipFile(self.filename, 'rb') as f:
            L = f.readlines()

        with gzip.GzipFile(self.filename, 'rb') as f:
            while 1:
                L = f.readlines(150)
                if L == []: break

    def test_seek_read(self):
        self.test_write()
        

        with gzip.GzipFile(self.filename) as f:
            while 1:
                oldpos = f.tell()
                line1 = f.readline()
                if not line1: break
                newpos = f.tell()
                f.seek(oldpos)  
                if len(line1)>10:
                    amount = 10
                else:
                    amount = len(line1)
                line2 = f.read(amount)
                self.assertEqual(line1[:amount], line2)
                f.seek(newpos)  

    def test_seek_whence(self):
        self.test_write()
        

        with gzip.GzipFile(self.filename) as f:
            f.read(10)
            f.seek(10, whence=1)
            y = f.read(10)
        self.assertEqual(y, data1[20:30])

    def test_seek_write(self):
        
        with gzip.GzipFile(self.filename, 'w') as f:
            for pos in range(0, 256, 16):
                f.seek(pos)
                f.write(b'GZ\n')

    def test_mode(self):
        self.test_write()
        with gzip.GzipFile(self.filename, 'r') as f:
            self.assertEqual(f.myfileobj.mode, 'rb')
        os_helper.unlink(self.filename)
        with gzip.GzipFile(self.filename, 'x') as f:
            self.assertEqual(f.myfileobj.mode, 'xb')

    def test_1647484(self):
        for mode in ('wb', 'rb'):
            with gzip.GzipFile(self.filename, mode) as f:
                self.assertTrue(hasattr(f, "name"))
                self.assertEqual(f.name, self.filename)

    def test_paddedfile_getattr(self):
        self.test_write()
        with gzip.GzipFile(self.filename, 'rb') as f:
            self.assertTrue(hasattr(f.fileobj, "name"))
            self.assertEqual(f.fileobj.name, self.filename)

    def test_mtime(self):
        mtime = 123456789
        with gzip.GzipFile(self.filename, 'w', mtime = mtime) as fWrite:
            fWrite.write(data1)
        with gzip.GzipFile(self.filename) as fRead:
            self.assertTrue(hasattr(fRead, 'mtime'))
            self.assertIsNone(fRead.mtime)
            dataRead = fRead.read()
            self.assertEqual(dataRead, data1)
            self.assertEqual(fRead.mtime, mtime)

    def test_metadata(self):
        mtime = 123456789

        with gzip.GzipFile(self.filename, 'w', mtime = mtime) as fWrite:
            fWrite.write(data1)

        with open(self.filename, 'rb') as fRead:
            

            idBytes = fRead.read(2)
            self.assertEqual(idBytes, b'\x1f\x8b') 

            cmByte = fRead.read(1)
            self.assertEqual(cmByte, b'\x08') 

            try:
                expectedname = self.filename.encode('Latin-1') + b'\x00'
                expectedflags = b'\x08' 
            except UnicodeEncodeError:
                expectedname = b''
                expectedflags = b'\x00'

            flagsByte = fRead.read(1)
            self.assertEqual(flagsByte, expectedflags)

            mtimeBytes = fRead.read(4)
            self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) 

            xflByte = fRead.read(1)
            self.assertEqual(xflByte, b'\x02') 

            osByte = fRead.read(1)
            self.assertEqual(osByte, b'\xff') 

            
            
            
            
            nameBytes = fRead.read(len(expectedname))
            self.assertEqual(nameBytes, expectedname)

            
            
            fRead.seek(os.stat(self.filename).st_size - 8)

            crc32Bytes = fRead.read(4) 
            self.assertEqual(crc32Bytes, b'\xaf\xd7d\x83')

            isizeBytes = fRead.read(4)
            self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))

    def test_metadata_ascii_name(self):
        self.filename = os_helper.TESTFN_ASCII
        self.test_metadata()

    def test_compresslevel_metadata(self):
        
        
        cases = [
            ('fast', 1, b'\x04'),
            ('best', 9, b'\x02'),
            ('tradeoff', 6, b'\x00'),
        ]
        xflOffset = 8

        for (name, level, expectedXflByte) in cases:
            with self.subTest(name):
                fWrite = gzip.GzipFile(self.filename, 'w', compresslevel=level)
                with fWrite:
                    fWrite.write(data1)
                with open(self.filename, 'rb') as fRead:
                    fRead.seek(xflOffset)
                    xflByte = fRead.read(1)
                    self.assertEqual(xflByte, expectedXflByte)

    def test_with_open(self):
        
        with gzip.GzipFile(self.filename, "wb") as f:
            f.write(b"xxx")
        f = gzip.GzipFile(self.filename, "rb")
        f.close()
        try:
            with f:
                pass
        except ValueError:
            pass
        else:
            self.fail("__enter__ on a closed file didn't raise an exception")
        try:
            with gzip.GzipFile(self.filename, "wb") as f:
                1/0
        except ZeroDivisionError:
            pass
        else:
            self.fail("1/0 didn't raise an exception")

    def test_zero_padded_file(self):
        with gzip.GzipFile(self.filename, "wb") as f:
            f.write(data1 * 50)

        
        with open(self.filename, "ab") as f:
            f.write(b"\x00" * 50)

        with gzip.GzipFile(self.filename, "rb") as f:
            d = f.read()
            self.assertEqual(d, data1 * 50, "Incorrect data in file")

    def test_gzip_BadGzipFile_exception(self):
        self.assertTrue(issubclass(gzip.BadGzipFile, OSError))

    def test_bad_gzip_file(self):
        with open(self.filename, 'wb') as file:
            file.write(data1 * 50)
        with gzip.GzipFile(self.filename, 'r') as file:
            self.assertRaises(gzip.BadGzipFile, file.readlines)

    def test_non_seekable_file(self):
        uncompressed = data1 * 50
        buf = UnseekableIO()
        with gzip.GzipFile(fileobj=buf, mode="wb") as f:
            f.write(uncompressed)
        compressed = buf.getvalue()
        buf = UnseekableIO(compressed)
        with gzip.GzipFile(fileobj=buf, mode="rb") as f:
            self.assertEqual(f.read(), uncompressed)

    def test_peek(self):
        uncompressed = data1 * 200
        with gzip.GzipFile(self.filename, "wb") as f:
            f.write(uncompressed)

        def sizes():
            while True:
                for n in range(5, 50, 10):
                    yield n

        with gzip.GzipFile(self.filename, "rb") as f:
            f.max_read_chunk = 33
            nread = 0
            for n in sizes():
                s = f.peek(n)
                if s == b'':
                    break
                self.assertEqual(f.read(len(s)), s)
                nread += len(s)
            self.assertEqual(f.read(100), b'')
            self.assertEqual(nread, len(uncompressed))

    def test_textio_readlines(self):
        
        lines = (data1 * 50).decode("ascii").splitlines(keepends=True)
        self.test_write()
        with gzip.GzipFile(self.filename, 'r') as f:
            with io.TextIOWrapper(f, encoding="ascii") as t:
                self.assertEqual(t.readlines(), lines)

    def test_fileobj_from_fdopen(self):
        
        
        fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT)
        with os.fdopen(fd, "wb") as f:
            with gzip.GzipFile(fileobj=f, mode="w") as g:
                pass

    def test_fileobj_mode(self):
        gzip.GzipFile(self.filename, "wb").close()
        with open(self.filename, "r+b") as f:
            with gzip.GzipFile(fileobj=f, mode='r') as g:
                self.assertEqual(g.mode, gzip.READ)
            with gzip.GzipFile(fileobj=f, mode='w') as g:
                self.assertEqual(g.mode, gzip.WRITE)
            with gzip.GzipFile(fileobj=f, mode='a') as g:
                self.assertEqual(g.mode, gzip.WRITE)
            with gzip.GzipFile(fileobj=f, mode='x') as g:
                self.assertEqual(g.mode, gzip.WRITE)
            with self.assertRaises(ValueError):
                gzip.GzipFile(fileobj=f, mode='z')
        for mode in "rb", "r+b":
            with open(self.filename, mode) as f:
                with gzip.GzipFile(fileobj=f) as g:
                    self.assertEqual(g.mode, gzip.READ)
        for mode in "wb", "ab", "xb":
            if "x" in mode:
                os_helper.unlink(self.filename)
            with open(self.filename, mode) as f:
                with self.assertWarns(FutureWarning):
                    g = gzip.GzipFile(fileobj=f)
                with g:
                    self.assertEqual(g.mode, gzip.WRITE)

    def test_bytes_filename(self):
        str_filename = self.filename
        try:
            bytes_filename = str_filename.encode("ascii")
        except UnicodeEncodeError:
            self.skipTest("Temporary file name needs to be ASCII")
        with gzip.GzipFile(bytes_filename, "wb") as f:
            f.write(data1 * 50)
        with gzip.GzipFile(bytes_filename, "rb") as f:
            self.assertEqual(f.read(), data1 * 50)
        
        with gzip.GzipFile(str_filename, "rb") as f:
            self.assertEqual(f.read(), data1 * 50)

    def test_decompress_limited(self):
        
        bomb = gzip.compress(b'\0' * int(2e6), compresslevel=9)
        self.assertLess(len(bomb), io.DEFAULT_BUFFER_SIZE)

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

    

    def test_compress(self):
        for data in [data1, data2]:
            for args in [(), (1,), (6,), (9,)]:
                datac = gzip.compress(data, *args)
                self.assertEqual(type(datac), bytes)
                with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f:
                    self.assertEqual(f.read(), data)

    def test_compress_mtime(self):
        mtime = 123456789
        for data in [data1, data2]:
            for args in [(), (1,), (6,), (9,)]:
                with self.subTest(data=data, args=args):
                    datac = gzip.compress(data, *args, mtime=mtime)
                    self.assertEqual(type(datac), bytes)
                    with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f:
                        f.read(1) 
                        self.assertEqual(f.mtime, mtime)

    def test_compress_correct_level(self):
        
        for mtime in (0, 42):
            with self.subTest(mtime=mtime):
                nocompress = gzip.compress(data1, compresslevel=0, mtime=mtime)
                yescompress = gzip.compress(data1, compresslevel=1, mtime=mtime)
                self.assertIn(data1, nocompress)
                self.assertNotIn(data1, yescompress)

    def test_decompress(self):
        for data in (data1, data2):
            buf = io.BytesIO()
            with gzip.GzipFile(fileobj=buf, mode="wb") as f:
                f.write(data)
            self.assertEqual(gzip.decompress(buf.getvalue()), data)
            
            datac = gzip.compress(data)
            self.assertEqual(gzip.decompress(datac), data)

    def test_decompress_truncated_trailer(self):
        compressed_data = gzip.compress(data1)
        self.assertRaises(EOFError, gzip.decompress, compressed_data[:-4])

    def test_decompress_missing_trailer(self):
        compressed_data = gzip.compress(data1)
        self.assertRaises(EOFError, gzip.decompress, compressed_data[:-8])

    def test_read_truncated(self):
        data = data1*50
        
        truncated = gzip.compress(data)[:-8]
        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
            self.assertRaises(EOFError, f.read)
        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
            self.assertEqual(f.read(len(data)), data)
            self.assertRaises(EOFError, f.read, 1)
        
        for i in range(2, 10):
            with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f:
                self.assertRaises(EOFError, f.read, 1)

    def test_read_with_extra(self):
        
        gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff'
                  b'\x05\x00Extra'
                  b'\x0bI-.\x01\x002\xd1Mx\x04\x00\x00\x00')
        with gzip.GzipFile(fileobj=io.BytesIO(gzdata)) as f:
            self.assertEqual(f.read(), b'Test')

    def test_prepend_error(self):
        
        with gzip.open(self.filename, "wb") as f:
            f.write(data1)
        with gzip.open(self.filename, "rb") as f:
            f._buffer.raw._fp.prepend()

    def test_issue44439(self):
        q = array.array('Q', [1, 2, 3, 4, 5])
        LENGTH = len(q) * q.itemsize

        with gzip.GzipFile(fileobj=io.BytesIO(), mode='w') as f:
            self.assertEqual(f.write(q), LENGTH)
            self.assertEqual(f.tell(), LENGTH)

    def test_flush_flushes_compressor(self):
        
        b = io.BytesIO()
        message = b"important message here."
        with gzip.GzipFile(fileobj=b, mode='w') as f:
            f.write(message)
            f.flush()
            partial_data = b.getvalue()
        full_data = b.getvalue()
        self.assertEqual(gzip.decompress(full_data), message)
        
        
        
        with self.assertRaises(EOFError):
            gzip.decompress(partial_data)
        d = zlib.decompressobj(wbits=-zlib.MAX_WBITS)
        f = io.BytesIO(partial_data)
        gzip._read_gzip_header(f)
        read_message = d.decompress(f.read())
        self.assertEqual(read_message, message)

    def test_flush_modes(self):
        
        
        class FakeCompressor:
            def __init__(self):
                self.modes = []
            def compress(self, data):
                return b''
            def flush(self, mode=-1):
                self.modes.append(mode)
                return b''
        b = io.BytesIO()
        fc = FakeCompressor()
        with gzip.GzipFile(fileobj=b, mode='w') as f:
            f.compress = fc
            f.flush()
            f.flush(50)
            f.flush(zlib_mode=100)
        
        expected_modes = [
            zlib.Z_SYNC_FLUSH,
            50,
            100,
            -1,
        ]
        self.assertEqual(fc.modes, expected_modes)

    def test_write_seek_write(self):
        
        
        b = io.BytesIO()
        message = b"important message here."
        with gzip.GzipFile(fileobj=b, mode='w') as f:
            f.write(message)
            f.seek(len(message))
            f.write(message)
        data = b.getvalue()
        self.assertEqual(gzip.decompress(data), message * 2)


class TestOpen(BaseTest):
    def test_binary_modes(self):
        uncompressed = data1 * 50

        with gzip.open(self.filename, "wb") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

        with gzip.open(self.filename, "rb") as f:
            self.assertEqual(f.read(), uncompressed)

        with gzip.open(self.filename, "ab") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed * 2)

        with self.assertRaises(FileExistsError):
            gzip.open(self.filename, "xb")
        os_helper.unlink(self.filename)
        with gzip.open(self.filename, "xb") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

    def test_pathlike_file(self):
        filename = pathlib.Path(self.filename)
        with gzip.open(filename, "wb") as f:
            f.write(data1 * 50)
        with gzip.open(filename, "ab") as f:
            f.write(data1)
        with gzip.open(filename) as f:
            self.assertEqual(f.read(), data1 * 51)

    def test_implicit_binary_modes(self):
        
        uncompressed = data1 * 50

        with gzip.open(self.filename, "w") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

        with gzip.open(self.filename, "r") as f:
            self.assertEqual(f.read(), uncompressed)

        with gzip.open(self.filename, "a") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed * 2)

        with self.assertRaises(FileExistsError):
            gzip.open(self.filename, "x")
        os_helper.unlink(self.filename)
        with gzip.open(self.filename, "x") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

    def test_text_modes(self):
        uncompressed = data1.decode("ascii") * 50
        uncompressed_raw = uncompressed.replace("\n", os.linesep)
        with gzip.open(self.filename, "wt", encoding="ascii") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read()).decode("ascii")
            self.assertEqual(file_data, uncompressed_raw)
        with gzip.open(self.filename, "rt", encoding="ascii") as f:
            self.assertEqual(f.read(), uncompressed)
        with gzip.open(self.filename, "at", encoding="ascii") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read()).decode("ascii")
            self.assertEqual(file_data, uncompressed_raw * 2)

    def test_fileobj(self):
        uncompressed_bytes = data1 * 50
        uncompressed_str = uncompressed_bytes.decode("ascii")
        compressed = gzip.compress(uncompressed_bytes)
        with gzip.open(io.BytesIO(compressed), "r") as f:
            self.assertEqual(f.read(), uncompressed_bytes)
        with gzip.open(io.BytesIO(compressed), "rb") as f:
            self.assertEqual(f.read(), uncompressed_bytes)
        with gzip.open(io.BytesIO(compressed), "rt", encoding="ascii") as f:
            self.assertEqual(f.read(), uncompressed_str)

    def test_bad_params(self):
        
        with self.assertRaises(TypeError):
            gzip.open(123.456)
        with self.assertRaises(ValueError):
            gzip.open(self.filename, "wbt")
        with self.assertRaises(ValueError):
            gzip.open(self.filename, "xbt")
        with self.assertRaises(ValueError):
            gzip.open(self.filename, "rb", encoding="utf-8")
        with self.assertRaises(ValueError):
            gzip.open(self.filename, "rb", errors="ignore")
        with self.assertRaises(ValueError):
            gzip.open(self.filename, "rb", newline="\n")

    def test_encoding(self):
        
        uncompressed = data1.decode("ascii") * 50
        uncompressed_raw = uncompressed.replace("\n", os.linesep)
        with gzip.open(self.filename, "wt", encoding="utf-16") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read()).decode("utf-16")
            self.assertEqual(file_data, uncompressed_raw)
        with gzip.open(self.filename, "rt", encoding="utf-16") as f:
            self.assertEqual(f.read(), uncompressed)

    def test_encoding_error_handler(self):
        
        with gzip.open(self.filename, "wb") as f:
            f.write(b"foo\xffbar")
        with gzip.open(self.filename, "rt", encoding="ascii", errors="ignore") \
                as f:
            self.assertEqual(f.read(), "foobar")

    def test_newline(self):
        
        uncompressed = data1.decode("ascii") * 50
        with gzip.open(self.filename, "wt", encoding="ascii", newline="\n") as f:
            f.write(uncompressed)
        with gzip.open(self.filename, "rt", encoding="ascii", newline="\r") as f:
            self.assertEqual(f.readlines(), [uncompressed])


def create_and_remove_directory(directory):
    def decorator(function):
        @functools.wraps(function)
        def wrapper(*args, **kwargs):
            os.makedirs(directory)
            try:
                return function(*args, **kwargs)
            finally:
                os_helper.rmtree(directory)
        return wrapper
    return decorator


class TestCommandLine(unittest.TestCase):
    data = b'This is a simple test with gzip'

    @requires_subprocess()
    def test_decompress_stdin_stdout(self):
        with io.BytesIO() as bytes_io:
            with gzip.GzipFile(fileobj=bytes_io, mode='wb') as gzip_file:
                gzip_file.write(self.data)

            args = sys.executable, '-m', 'gzip', '-d'
            with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc:
                out, err = proc.communicate(bytes_io.getvalue())

        self.assertEqual(err, b'')
        self.assertEqual(out, self.data)

    @create_and_remove_directory(TEMPDIR)
    def test_decompress_infile_outfile(self):
        gzipname = os.path.join(TEMPDIR, 'testgzip.gz')
        self.assertFalse(os.path.exists(gzipname))

        with gzip.open(gzipname, mode='wb') as fp:
            fp.write(self.data)
        rc, out, err = assert_python_ok('-m', 'gzip', '-d', gzipname)

        with open(os.path.join(TEMPDIR, "testgzip"), "rb") as gunziped:
            self.assertEqual(gunziped.read(), self.data)

        self.assertTrue(os.path.exists(gzipname))
        self.assertEqual(rc, 0)
        self.assertEqual(out, b'')
        self.assertEqual(err, b'')

    def test_decompress_infile_outfile_error(self):
        rc, out, err = assert_python_failure('-m', 'gzip', '-d', 'thisisatest.out')
        self.assertEqual(b"filename doesn't end in .gz: 'thisisatest.out'", err.strip())
        self.assertEqual(rc, 1)
        self.assertEqual(out, b'')

    @requires_subprocess()
    @create_and_remove_directory(TEMPDIR)
    def test_compress_stdin_outfile(self):
        args = sys.executable, '-m', 'gzip'
        with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc:
            out, err = proc.communicate(self.data)

        self.assertEqual(err, b'')
        self.assertEqual(out[:2], b"\x1f\x8b")

    @create_and_remove_directory(TEMPDIR)
    def test_compress_infile_outfile_default(self):
        local_testgzip = os.path.join(TEMPDIR, 'testgzip')
        gzipname = local_testgzip + '.gz'
        self.assertFalse(os.path.exists(gzipname))

        with open(local_testgzip, 'wb') as fp:
            fp.write(self.data)

        rc, out, err = assert_python_ok('-m', 'gzip', local_testgzip)

        self.assertTrue(os.path.exists(gzipname))
        self.assertEqual(out, b'')
        self.assertEqual(err, b'')

    @create_and_remove_directory(TEMPDIR)
    def test_compress_infile_outfile(self):
        for compress_level in ('--fast', '--best'):
            with self.subTest(compress_level=compress_level):
                local_testgzip = os.path.join(TEMPDIR, 'testgzip')
                gzipname = local_testgzip + '.gz'
                self.assertFalse(os.path.exists(gzipname))

                with open(local_testgzip, 'wb') as fp:
                    fp.write(self.data)

                rc, out, err = assert_python_ok('-m', 'gzip', compress_level, local_testgzip)

                self.assertTrue(os.path.exists(gzipname))
                self.assertEqual(out, b'')
                self.assertEqual(err, b'')
                os.remove(gzipname)
                self.assertFalse(os.path.exists(gzipname))

    def test_compress_fast_best_are_exclusive(self):
        rc, out, err = assert_python_failure('-m', 'gzip', '--fast', '--best')
        self.assertIn(b"error: argument --best: not allowed with argument --fast", err)
        self.assertEqual(out, b'')

    def test_decompress_cannot_have_flags_compression(self):
        rc, out, err = assert_python_failure('-m', 'gzip', '--fast', '-d')
        self.assertIn(b'error: argument -d/--decompress: not allowed with argument --fast', err)
        self.assertEqual(out, b'')


if __name__ == "__main__":
    unittest.main()