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