1"""Test script for the gzip module.
2"""
3
4import array
5import functools
6import io
7import os
8import pathlib
9import struct
10import sys
11import unittest
12from subprocess import PIPE, Popen
13from test import support
14from test.support import _4G, bigmemtest
15from test.support.script_helper import assert_python_ok, assert_python_failure
16
17gzip = support.import_module('gzip')
18
19data1 = b"""  int length=DEFAULTALLOC, err = Z_OK;
20  PyObject *RetVal;
21  int flushmode = Z_FINISH;
22  unsigned long start_total_out;
23
24"""
25
26data2 = b"""/* zlibmodule.c -- gzip-compatible data compression */
27/* See http://www.gzip.org/zlib/
28/* See http://www.winimage.com/zLibDll for Windows */
29"""
30
31
32TEMPDIR = os.path.abspath(support.TESTFN) + '-gzdir'
33
34
35class UnseekableIO(io.BytesIO):
36    def seekable(self):
37        return False
38
39    def tell(self):
40        raise io.UnsupportedOperation
41
42    def seek(self, *args):
43        raise io.UnsupportedOperation
44
45
46class BaseTest(unittest.TestCase):
47    filename = support.TESTFN
48
49    def setUp(self):
50        support.unlink(self.filename)
51
52    def tearDown(self):
53        support.unlink(self.filename)
54
55
56class TestGzip(BaseTest):
57    def write_and_read_back(self, data, mode='b'):
58        b_data = bytes(data)
59        with gzip.GzipFile(self.filename, 'w'+mode) as f:
60            l = f.write(data)
61        self.assertEqual(l, len(b_data))
62        with gzip.GzipFile(self.filename, 'r'+mode) as f:
63            self.assertEqual(f.read(), b_data)
64
65    def test_write(self):
66        with gzip.GzipFile(self.filename, 'wb') as f:
67            f.write(data1 * 50)
68
69            # Try flush and fileno.
70            f.flush()
71            f.fileno()
72            if hasattr(os, 'fsync'):
73                os.fsync(f.fileno())
74            f.close()
75
76        # Test multiple close() calls.
77        f.close()
78
79    def test_write_read_with_pathlike_file(self):
80        filename = pathlib.Path(self.filename)
81        with gzip.GzipFile(filename, 'w') as f:
82            f.write(data1 * 50)
83        self.assertIsInstance(f.name, str)
84        with gzip.GzipFile(filename, 'a') as f:
85            f.write(data1)
86        with gzip.GzipFile(filename) as f:
87            d = f.read()
88        self.assertEqual(d, data1 * 51)
89        self.assertIsInstance(f.name, str)
90
91    # The following test_write_xy methods test that write accepts
92    # the corresponding bytes-like object type as input
93    # and that the data written equals bytes(xy) in all cases.
94    def test_write_memoryview(self):
95        self.write_and_read_back(memoryview(data1 * 50))
96        m = memoryview(bytes(range(256)))
97        data = m.cast('B', shape=[8,8,4])
98        self.write_and_read_back(data)
99
100    def test_write_bytearray(self):
101        self.write_and_read_back(bytearray(data1 * 50))
102
103    def test_write_array(self):
104        self.write_and_read_back(array.array('I', data1 * 40))
105
106    def test_write_incompatible_type(self):
107        # Test that non-bytes-like types raise TypeError.
108        # Issue #21560: attempts to write incompatible types
109        # should not affect the state of the fileobject
110        with gzip.GzipFile(self.filename, 'wb') as f:
111            with self.assertRaises(TypeError):
112                f.write('')
113            with self.assertRaises(TypeError):
114                f.write([])
115            f.write(data1)
116        with gzip.GzipFile(self.filename, 'rb') as f:
117            self.assertEqual(f.read(), data1)
118
119    def test_read(self):
120        self.test_write()
121        # Try reading.
122        with gzip.GzipFile(self.filename, 'r') as f:
123            d = f.read()
124        self.assertEqual(d, data1*50)
125
126    def test_read1(self):
127        self.test_write()
128        blocks = []
129        nread = 0
130        with gzip.GzipFile(self.filename, 'r') as f:
131            while True:
132                d = f.read1()
133                if not d:
134                    break
135                blocks.append(d)
136                nread += len(d)
137                # Check that position was updated correctly (see issue10791).
138                self.assertEqual(f.tell(), nread)
139        self.assertEqual(b''.join(blocks), data1 * 50)
140
141    @bigmemtest(size=_4G, memuse=1)
142    def test_read_large(self, size):
143        # Read chunk size over UINT_MAX should be supported, despite zlib's
144        # limitation per low-level call
145        compressed = gzip.compress(data1, compresslevel=1)
146        f = gzip.GzipFile(fileobj=io.BytesIO(compressed), mode='rb')
147        self.assertEqual(f.read(size), data1)
148
149    def test_io_on_closed_object(self):
150        # Test that I/O operations on closed GzipFile objects raise a
151        # ValueError, just like the corresponding functions on file objects.
152
153        # Write to a file, open it for reading, then close it.
154        self.test_write()
155        f = gzip.GzipFile(self.filename, 'r')
156        fileobj = f.fileobj
157        self.assertFalse(fileobj.closed)
158        f.close()
159        self.assertTrue(fileobj.closed)
160        with self.assertRaises(ValueError):
161            f.read(1)
162        with self.assertRaises(ValueError):
163            f.seek(0)
164        with self.assertRaises(ValueError):
165            f.tell()
166        # Open the file for writing, then close it.
167        f = gzip.GzipFile(self.filename, 'w')
168        fileobj = f.fileobj
169        self.assertFalse(fileobj.closed)
170        f.close()
171        self.assertTrue(fileobj.closed)
172        with self.assertRaises(ValueError):
173            f.write(b'')
174        with self.assertRaises(ValueError):
175            f.flush()
176
177    def test_append(self):
178        self.test_write()
179        # Append to the previous file
180        with gzip.GzipFile(self.filename, 'ab') as f:
181            f.write(data2 * 15)
182
183        with gzip.GzipFile(self.filename, 'rb') as f:
184            d = f.read()
185        self.assertEqual(d, (data1*50) + (data2*15))
186
187    def test_many_append(self):
188        # Bug #1074261 was triggered when reading a file that contained
189        # many, many members.  Create such a file and verify that reading it
190        # works.
191        with gzip.GzipFile(self.filename, 'wb', 9) as f:
192            f.write(b'a')
193        for i in range(0, 200):
194            with gzip.GzipFile(self.filename, "ab", 9) as f: # append
195                f.write(b'a')
196
197        # Try reading the file
198        with gzip.GzipFile(self.filename, "rb") as zgfile:
199            contents = b""
200            while 1:
201                ztxt = zgfile.read(8192)
202                contents += ztxt
203                if not ztxt: break
204        self.assertEqual(contents, b'a'*201)
205
206    def test_exclusive_write(self):
207        with gzip.GzipFile(self.filename, 'xb') as f:
208            f.write(data1 * 50)
209        with gzip.GzipFile(self.filename, 'rb') as f:
210            self.assertEqual(f.read(), data1 * 50)
211        with self.assertRaises(FileExistsError):
212            gzip.GzipFile(self.filename, 'xb')
213
214    def test_buffered_reader(self):
215        # Issue #7471: a GzipFile can be wrapped in a BufferedReader for
216        # performance.
217        self.test_write()
218
219        with gzip.GzipFile(self.filename, 'rb') as f:
220            with io.BufferedReader(f) as r:
221                lines = [line for line in r]
222
223        self.assertEqual(lines, 50 * data1.splitlines(keepends=True))
224
225    def test_readline(self):
226        self.test_write()
227        # Try .readline() with varying line lengths
228
229        with gzip.GzipFile(self.filename, 'rb') as f:
230            line_length = 0
231            while 1:
232                L = f.readline(line_length)
233                if not L and line_length != 0: break
234                self.assertTrue(len(L) <= line_length)
235                line_length = (line_length + 1) % 50
236
237    def test_readlines(self):
238        self.test_write()
239        # Try .readlines()
240
241        with gzip.GzipFile(self.filename, 'rb') as f:
242            L = f.readlines()
243
244        with gzip.GzipFile(self.filename, 'rb') as f:
245            while 1:
246                L = f.readlines(150)
247                if L == []: break
248
249    def test_seek_read(self):
250        self.test_write()
251        # Try seek, read test
252
253        with gzip.GzipFile(self.filename) as f:
254            while 1:
255                oldpos = f.tell()
256                line1 = f.readline()
257                if not line1: break
258                newpos = f.tell()
259                f.seek(oldpos)  # negative seek
260                if len(line1)>10:
261                    amount = 10
262                else:
263                    amount = len(line1)
264                line2 = f.read(amount)
265                self.assertEqual(line1[:amount], line2)
266                f.seek(newpos)  # positive seek
267
268    def test_seek_whence(self):
269        self.test_write()
270        # Try seek(whence=1), read test
271
272        with gzip.GzipFile(self.filename) as f:
273            f.read(10)
274            f.seek(10, whence=1)
275            y = f.read(10)
276        self.assertEqual(y, data1[20:30])
277
278    def test_seek_write(self):
279        # Try seek, write test
280        with gzip.GzipFile(self.filename, 'w') as f:
281            for pos in range(0, 256, 16):
282                f.seek(pos)
283                f.write(b'GZ\n')
284
285    def test_mode(self):
286        self.test_write()
287        with gzip.GzipFile(self.filename, 'r') as f:
288            self.assertEqual(f.myfileobj.mode, 'rb')
289        support.unlink(self.filename)
290        with gzip.GzipFile(self.filename, 'x') as f:
291            self.assertEqual(f.myfileobj.mode, 'xb')
292
293    def test_1647484(self):
294        for mode in ('wb', 'rb'):
295            with gzip.GzipFile(self.filename, mode) as f:
296                self.assertTrue(hasattr(f, "name"))
297                self.assertEqual(f.name, self.filename)
298
299    def test_paddedfile_getattr(self):
300        self.test_write()
301        with gzip.GzipFile(self.filename, 'rb') as f:
302            self.assertTrue(hasattr(f.fileobj, "name"))
303            self.assertEqual(f.fileobj.name, self.filename)
304
305    def test_mtime(self):
306        mtime = 123456789
307        with gzip.GzipFile(self.filename, 'w', mtime = mtime) as fWrite:
308            fWrite.write(data1)
309        with gzip.GzipFile(self.filename) as fRead:
310            self.assertTrue(hasattr(fRead, 'mtime'))
311            self.assertIsNone(fRead.mtime)
312            dataRead = fRead.read()
313            self.assertEqual(dataRead, data1)
314            self.assertEqual(fRead.mtime, mtime)
315
316    def test_metadata(self):
317        mtime = 123456789
318
319        with gzip.GzipFile(self.filename, 'w', mtime = mtime) as fWrite:
320            fWrite.write(data1)
321
322        with open(self.filename, 'rb') as fRead:
323            # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
324
325            idBytes = fRead.read(2)
326            self.assertEqual(idBytes, b'\x1f\x8b') # gzip ID
327
328            cmByte = fRead.read(1)
329            self.assertEqual(cmByte, b'\x08') # deflate
330
331            flagsByte = fRead.read(1)
332            self.assertEqual(flagsByte, b'\x08') # only the FNAME flag is set
333
334            mtimeBytes = fRead.read(4)
335            self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) # little-endian
336
337            xflByte = fRead.read(1)
338            self.assertEqual(xflByte, b'\x02') # maximum compression
339
340            osByte = fRead.read(1)
341            self.assertEqual(osByte, b'\xff') # OS "unknown" (OS-independent)
342
343            # Since the FNAME flag is set, the zero-terminated filename follows.
344            # RFC 1952 specifies that this is the name of the input file, if any.
345            # However, the gzip module defaults to storing the name of the output
346            # file in this field.
347            expected = self.filename.encode('Latin-1') + b'\x00'
348            nameBytes = fRead.read(len(expected))
349            self.assertEqual(nameBytes, expected)
350
351            # Since no other flags were set, the header ends here.
352            # Rather than process the compressed data, let's seek to the trailer.
353            fRead.seek(os.stat(self.filename).st_size - 8)
354
355            crc32Bytes = fRead.read(4) # CRC32 of uncompressed data [data1]
356            self.assertEqual(crc32Bytes, b'\xaf\xd7d\x83')
357
358            isizeBytes = fRead.read(4)
359            self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))
360
361    def test_compresslevel_metadata(self):
362        # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
363        # specifically, discussion of XFL in section 2.3.1
364        cases = [
365            ('fast', 1, b'\x04'),
366            ('best', 9, b'\x02'),
367            ('tradeoff', 6, b'\x00'),
368        ]
369        xflOffset = 8
370
371        for (name, level, expectedXflByte) in cases:
372            with self.subTest(name):
373                fWrite = gzip.GzipFile(self.filename, 'w', compresslevel=level)
374                with fWrite:
375                    fWrite.write(data1)
376                with open(self.filename, 'rb') as fRead:
377                    fRead.seek(xflOffset)
378                    xflByte = fRead.read(1)
379                    self.assertEqual(xflByte, expectedXflByte)
380
381    def test_with_open(self):
382        # GzipFile supports the context management protocol
383        with gzip.GzipFile(self.filename, "wb") as f:
384            f.write(b"xxx")
385        f = gzip.GzipFile(self.filename, "rb")
386        f.close()
387        try:
388            with f:
389                pass
390        except ValueError:
391            pass
392        else:
393            self.fail("__enter__ on a closed file didn't raise an exception")
394        try:
395            with gzip.GzipFile(self.filename, "wb") as f:
396                1/0
397        except ZeroDivisionError:
398            pass
399        else:
400            self.fail("1/0 didn't raise an exception")
401
402    def test_zero_padded_file(self):
403        with gzip.GzipFile(self.filename, "wb") as f:
404            f.write(data1 * 50)
405
406        # Pad the file with zeroes
407        with open(self.filename, "ab") as f:
408            f.write(b"\x00" * 50)
409
410        with gzip.GzipFile(self.filename, "rb") as f:
411            d = f.read()
412            self.assertEqual(d, data1 * 50, "Incorrect data in file")
413
414    def test_gzip_BadGzipFile_exception(self):
415        self.assertTrue(issubclass(gzip.BadGzipFile, OSError))
416
417    def test_bad_gzip_file(self):
418        with open(self.filename, 'wb') as file:
419            file.write(data1 * 50)
420        with gzip.GzipFile(self.filename, 'r') as file:
421            self.assertRaises(gzip.BadGzipFile, file.readlines)
422
423    def test_non_seekable_file(self):
424        uncompressed = data1 * 50
425        buf = UnseekableIO()
426        with gzip.GzipFile(fileobj=buf, mode="wb") as f:
427            f.write(uncompressed)
428        compressed = buf.getvalue()
429        buf = UnseekableIO(compressed)
430        with gzip.GzipFile(fileobj=buf, mode="rb") as f:
431            self.assertEqual(f.read(), uncompressed)
432
433    def test_peek(self):
434        uncompressed = data1 * 200
435        with gzip.GzipFile(self.filename, "wb") as f:
436            f.write(uncompressed)
437
438        def sizes():
439            while True:
440                for n in range(5, 50, 10):
441                    yield n
442
443        with gzip.GzipFile(self.filename, "rb") as f:
444            f.max_read_chunk = 33
445            nread = 0
446            for n in sizes():
447                s = f.peek(n)
448                if s == b'':
449                    break
450                self.assertEqual(f.read(len(s)), s)
451                nread += len(s)
452            self.assertEqual(f.read(100), b'')
453            self.assertEqual(nread, len(uncompressed))
454
455    def test_textio_readlines(self):
456        # Issue #10791: TextIOWrapper.readlines() fails when wrapping GzipFile.
457        lines = (data1 * 50).decode("ascii").splitlines(keepends=True)
458        self.test_write()
459        with gzip.GzipFile(self.filename, 'r') as f:
460            with io.TextIOWrapper(f, encoding="ascii") as t:
461                self.assertEqual(t.readlines(), lines)
462
463    def test_fileobj_from_fdopen(self):
464        # Issue #13781: Opening a GzipFile for writing fails when using a
465        # fileobj created with os.fdopen().
466        fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT)
467        with os.fdopen(fd, "wb") as f:
468            with gzip.GzipFile(fileobj=f, mode="w") as g:
469                pass
470
471    def test_fileobj_mode(self):
472        gzip.GzipFile(self.filename, "wb").close()
473        with open(self.filename, "r+b") as f:
474            with gzip.GzipFile(fileobj=f, mode='r') as g:
475                self.assertEqual(g.mode, gzip.READ)
476            with gzip.GzipFile(fileobj=f, mode='w') as g:
477                self.assertEqual(g.mode, gzip.WRITE)
478            with gzip.GzipFile(fileobj=f, mode='a') as g:
479                self.assertEqual(g.mode, gzip.WRITE)
480            with gzip.GzipFile(fileobj=f, mode='x') as g:
481                self.assertEqual(g.mode, gzip.WRITE)
482            with self.assertRaises(ValueError):
483                gzip.GzipFile(fileobj=f, mode='z')
484        for mode in "rb", "r+b":
485            with open(self.filename, mode) as f:
486                with gzip.GzipFile(fileobj=f) as g:
487                    self.assertEqual(g.mode, gzip.READ)
488        for mode in "wb", "ab", "xb":
489            if "x" in mode:
490                support.unlink(self.filename)
491            with open(self.filename, mode) as f:
492                with gzip.GzipFile(fileobj=f) as g:
493                    self.assertEqual(g.mode, gzip.WRITE)
494
495    def test_bytes_filename(self):
496        str_filename = self.filename
497        try:
498            bytes_filename = str_filename.encode("ascii")
499        except UnicodeEncodeError:
500            self.skipTest("Temporary file name needs to be ASCII")
501        with gzip.GzipFile(bytes_filename, "wb") as f:
502            f.write(data1 * 50)
503        with gzip.GzipFile(bytes_filename, "rb") as f:
504            self.assertEqual(f.read(), data1 * 50)
505        # Sanity check that we are actually operating on the right file.
506        with gzip.GzipFile(str_filename, "rb") as f:
507            self.assertEqual(f.read(), data1 * 50)
508
509    def test_decompress_limited(self):
510        """Decompressed data buffering should be limited"""
511        bomb = gzip.compress(b'\0' * int(2e6), compresslevel=9)
512        self.assertLess(len(bomb), io.DEFAULT_BUFFER_SIZE)
513
514        bomb = io.BytesIO(bomb)
515        decomp = gzip.GzipFile(fileobj=bomb)
516        self.assertEqual(decomp.read(1), b'\0')
517        max_decomp = 1 + io.DEFAULT_BUFFER_SIZE
518        self.assertLessEqual(decomp._buffer.raw.tell(), max_decomp,
519            "Excessive amount of data was decompressed")
520
521    # Testing compress/decompress shortcut functions
522
523    def test_compress(self):
524        for data in [data1, data2]:
525            for args in [(), (1,), (6,), (9,)]:
526                datac = gzip.compress(data, *args)
527                self.assertEqual(type(datac), bytes)
528                with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f:
529                    self.assertEqual(f.read(), data)
530
531    def test_compress_mtime(self):
532        mtime = 123456789
533        for data in [data1, data2]:
534            for args in [(), (1,), (6,), (9,)]:
535                with self.subTest(data=data, args=args):
536                    datac = gzip.compress(data, *args, mtime=mtime)
537                    self.assertEqual(type(datac), bytes)
538                    with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f:
539                        f.read(1) # to set mtime attribute
540                        self.assertEqual(f.mtime, mtime)
541
542    def test_decompress(self):
543        for data in (data1, data2):
544            buf = io.BytesIO()
545            with gzip.GzipFile(fileobj=buf, mode="wb") as f:
546                f.write(data)
547            self.assertEqual(gzip.decompress(buf.getvalue()), data)
548            # Roundtrip with compress
549            datac = gzip.compress(data)
550            self.assertEqual(gzip.decompress(datac), data)
551
552    def test_read_truncated(self):
553        data = data1*50
554        # Drop the CRC (4 bytes) and file size (4 bytes).
555        truncated = gzip.compress(data)[:-8]
556        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
557            self.assertRaises(EOFError, f.read)
558        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
559            self.assertEqual(f.read(len(data)), data)
560            self.assertRaises(EOFError, f.read, 1)
561        # Incomplete 10-byte header.
562        for i in range(2, 10):
563            with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f:
564                self.assertRaises(EOFError, f.read, 1)
565
566    def test_read_with_extra(self):
567        # Gzip data with an extra field
568        gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff'
569                  b'\x05\x00Extra'
570                  b'\x0bI-.\x01\x002\xd1Mx\x04\x00\x00\x00')
571        with gzip.GzipFile(fileobj=io.BytesIO(gzdata)) as f:
572            self.assertEqual(f.read(), b'Test')
573
574    def test_prepend_error(self):
575        # See issue #20875
576        with gzip.open(self.filename, "wb") as f:
577            f.write(data1)
578        with gzip.open(self.filename, "rb") as f:
579            f._buffer.raw._fp.prepend()
580
581class TestOpen(BaseTest):
582    def test_binary_modes(self):
583        uncompressed = data1 * 50
584
585        with gzip.open(self.filename, "wb") as f:
586            f.write(uncompressed)
587        with open(self.filename, "rb") as f:
588            file_data = gzip.decompress(f.read())
589            self.assertEqual(file_data, uncompressed)
590
591        with gzip.open(self.filename, "rb") as f:
592            self.assertEqual(f.read(), uncompressed)
593
594        with gzip.open(self.filename, "ab") as f:
595            f.write(uncompressed)
596        with open(self.filename, "rb") as f:
597            file_data = gzip.decompress(f.read())
598            self.assertEqual(file_data, uncompressed * 2)
599
600        with self.assertRaises(FileExistsError):
601            gzip.open(self.filename, "xb")
602        support.unlink(self.filename)
603        with gzip.open(self.filename, "xb") as f:
604            f.write(uncompressed)
605        with open(self.filename, "rb") as f:
606            file_data = gzip.decompress(f.read())
607            self.assertEqual(file_data, uncompressed)
608
609    def test_pathlike_file(self):
610        filename = pathlib.Path(self.filename)
611        with gzip.open(filename, "wb") as f:
612            f.write(data1 * 50)
613        with gzip.open(filename, "ab") as f:
614            f.write(data1)
615        with gzip.open(filename) as f:
616            self.assertEqual(f.read(), data1 * 51)
617
618    def test_implicit_binary_modes(self):
619        # Test implicit binary modes (no "b" or "t" in mode string).
620        uncompressed = data1 * 50
621
622        with gzip.open(self.filename, "w") as f:
623            f.write(uncompressed)
624        with open(self.filename, "rb") as f:
625            file_data = gzip.decompress(f.read())
626            self.assertEqual(file_data, uncompressed)
627
628        with gzip.open(self.filename, "r") as f:
629            self.assertEqual(f.read(), uncompressed)
630
631        with gzip.open(self.filename, "a") as f:
632            f.write(uncompressed)
633        with open(self.filename, "rb") as f:
634            file_data = gzip.decompress(f.read())
635            self.assertEqual(file_data, uncompressed * 2)
636
637        with self.assertRaises(FileExistsError):
638            gzip.open(self.filename, "x")
639        support.unlink(self.filename)
640        with gzip.open(self.filename, "x") as f:
641            f.write(uncompressed)
642        with open(self.filename, "rb") as f:
643            file_data = gzip.decompress(f.read())
644            self.assertEqual(file_data, uncompressed)
645
646    def test_text_modes(self):
647        uncompressed = data1.decode("ascii") * 50
648        uncompressed_raw = uncompressed.replace("\n", os.linesep)
649        with gzip.open(self.filename, "wt") as f:
650            f.write(uncompressed)
651        with open(self.filename, "rb") as f:
652            file_data = gzip.decompress(f.read()).decode("ascii")
653            self.assertEqual(file_data, uncompressed_raw)
654        with gzip.open(self.filename, "rt") as f:
655            self.assertEqual(f.read(), uncompressed)
656        with gzip.open(self.filename, "at") as f:
657            f.write(uncompressed)
658        with open(self.filename, "rb") as f:
659            file_data = gzip.decompress(f.read()).decode("ascii")
660            self.assertEqual(file_data, uncompressed_raw * 2)
661
662    def test_fileobj(self):
663        uncompressed_bytes = data1 * 50
664        uncompressed_str = uncompressed_bytes.decode("ascii")
665        compressed = gzip.compress(uncompressed_bytes)
666        with gzip.open(io.BytesIO(compressed), "r") as f:
667            self.assertEqual(f.read(), uncompressed_bytes)
668        with gzip.open(io.BytesIO(compressed), "rb") as f:
669            self.assertEqual(f.read(), uncompressed_bytes)
670        with gzip.open(io.BytesIO(compressed), "rt") as f:
671            self.assertEqual(f.read(), uncompressed_str)
672
673    def test_bad_params(self):
674        # Test invalid parameter combinations.
675        with self.assertRaises(TypeError):
676            gzip.open(123.456)
677        with self.assertRaises(ValueError):
678            gzip.open(self.filename, "wbt")
679        with self.assertRaises(ValueError):
680            gzip.open(self.filename, "xbt")
681        with self.assertRaises(ValueError):
682            gzip.open(self.filename, "rb", encoding="utf-8")
683        with self.assertRaises(ValueError):
684            gzip.open(self.filename, "rb", errors="ignore")
685        with self.assertRaises(ValueError):
686            gzip.open(self.filename, "rb", newline="\n")
687
688    def test_encoding(self):
689        # Test non-default encoding.
690        uncompressed = data1.decode("ascii") * 50
691        uncompressed_raw = uncompressed.replace("\n", os.linesep)
692        with gzip.open(self.filename, "wt", encoding="utf-16") as f:
693            f.write(uncompressed)
694        with open(self.filename, "rb") as f:
695            file_data = gzip.decompress(f.read()).decode("utf-16")
696            self.assertEqual(file_data, uncompressed_raw)
697        with gzip.open(self.filename, "rt", encoding="utf-16") as f:
698            self.assertEqual(f.read(), uncompressed)
699
700    def test_encoding_error_handler(self):
701        # Test with non-default encoding error handler.
702        with gzip.open(self.filename, "wb") as f:
703            f.write(b"foo\xffbar")
704        with gzip.open(self.filename, "rt", encoding="ascii", errors="ignore") \
705                as f:
706            self.assertEqual(f.read(), "foobar")
707
708    def test_newline(self):
709        # Test with explicit newline (universal newline mode disabled).
710        uncompressed = data1.decode("ascii") * 50
711        with gzip.open(self.filename, "wt", newline="\n") as f:
712            f.write(uncompressed)
713        with gzip.open(self.filename, "rt", newline="\r") as f:
714            self.assertEqual(f.readlines(), [uncompressed])
715
716
717def create_and_remove_directory(directory):
718    def decorator(function):
719        @functools.wraps(function)
720        def wrapper(*args, **kwargs):
721            os.makedirs(directory)
722            try:
723                return function(*args, **kwargs)
724            finally:
725                support.rmtree(directory)
726        return wrapper
727    return decorator
728
729
730class TestCommandLine(unittest.TestCase):
731    data = b'This is a simple test with gzip'
732
733    def test_decompress_stdin_stdout(self):
734        with io.BytesIO() as bytes_io:
735            with gzip.GzipFile(fileobj=bytes_io, mode='wb') as gzip_file:
736                gzip_file.write(self.data)
737
738            args = sys.executable, '-m', 'gzip', '-d'
739            with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc:
740                out, err = proc.communicate(bytes_io.getvalue())
741
742        self.assertEqual(err, b'')
743        self.assertEqual(out, self.data)
744
745    @create_and_remove_directory(TEMPDIR)
746    def test_decompress_infile_outfile(self):
747        gzipname = os.path.join(TEMPDIR, 'testgzip.gz')
748        self.assertFalse(os.path.exists(gzipname))
749
750        with gzip.open(gzipname, mode='wb') as fp:
751            fp.write(self.data)
752        rc, out, err = assert_python_ok('-m', 'gzip', '-d', gzipname)
753
754        with open(os.path.join(TEMPDIR, "testgzip"), "rb") as gunziped:
755            self.assertEqual(gunziped.read(), self.data)
756
757        self.assertTrue(os.path.exists(gzipname))
758        self.assertEqual(rc, 0)
759        self.assertEqual(out, b'')
760        self.assertEqual(err, b'')
761
762    def test_decompress_infile_outfile_error(self):
763        rc, out, err = assert_python_failure('-m', 'gzip', '-d', 'thisisatest.out')
764        self.assertEqual(b"filename doesn't end in .gz: 'thisisatest.out'", err.strip())
765        self.assertEqual(rc, 1)
766        self.assertEqual(out, b'')
767
768    @create_and_remove_directory(TEMPDIR)
769    def test_compress_stdin_outfile(self):
770        args = sys.executable, '-m', 'gzip'
771        with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc:
772            out, err = proc.communicate(self.data)
773
774        self.assertEqual(err, b'')
775        self.assertEqual(out[:2], b"\x1f\x8b")
776
777    @create_and_remove_directory(TEMPDIR)
778    def test_compress_infile_outfile_default(self):
779        local_testgzip = os.path.join(TEMPDIR, 'testgzip')
780        gzipname = local_testgzip + '.gz'
781        self.assertFalse(os.path.exists(gzipname))
782
783        with open(local_testgzip, 'wb') as fp:
784            fp.write(self.data)
785
786        rc, out, err = assert_python_ok('-m', 'gzip', local_testgzip)
787
788        self.assertTrue(os.path.exists(gzipname))
789        self.assertEqual(out, b'')
790        self.assertEqual(err, b'')
791
792    @create_and_remove_directory(TEMPDIR)
793    def test_compress_infile_outfile(self):
794        for compress_level in ('--fast', '--best'):
795            with self.subTest(compress_level=compress_level):
796                local_testgzip = os.path.join(TEMPDIR, 'testgzip')
797                gzipname = local_testgzip + '.gz'
798                self.assertFalse(os.path.exists(gzipname))
799
800                with open(local_testgzip, 'wb') as fp:
801                    fp.write(self.data)
802
803                rc, out, err = assert_python_ok('-m', 'gzip', compress_level, local_testgzip)
804
805                self.assertTrue(os.path.exists(gzipname))
806                self.assertEqual(out, b'')
807                self.assertEqual(err, b'')
808                os.remove(gzipname)
809                self.assertFalse(os.path.exists(gzipname))
810
811    def test_compress_fast_best_are_exclusive(self):
812        rc, out, err = assert_python_failure('-m', 'gzip', '--fast', '--best')
813        self.assertIn(b"error: argument --best: not allowed with argument --fast", err)
814        self.assertEqual(out, b'')
815
816    def test_decompress_cannot_have_flags_compression(self):
817        rc, out, err = assert_python_failure('-m', 'gzip', '--fast', '-d')
818        self.assertIn(b'error: argument -d/--decompress: not allowed with argument --fast', err)
819        self.assertEqual(out, b'')
820
821
822def test_main(verbose=None):
823    support.run_unittest(TestGzip, TestOpen, TestCommandLine)
824
825
826if __name__ == "__main__":
827    test_main(verbose=True)
828