1 """Test script for the gzip module. 2 """ 3 4 import array 5 import functools 6 import io 7 import os 8 import pathlib 9 import struct 10 import sys 11 import unittest 12 from subprocess import PIPE, Popen 13 from test import support 14 from test.support import _4G, bigmemtest 15 from test.support.script_helper import assert_python_ok 16 17 gzip = support.import_module('gzip') 18 19 data1 = b""" int length=DEFAULTALLOC, err = Z_OK; 20 PyObject *RetVal; 21 int flushmode = Z_FINISH; 22 unsigned long start_total_out; 23 24 """ 25 26 data2 = 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 32 TEMPDIR = os.path.abspath(support.TESTFN) + '-gzdir' 33 34 35 class 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 46 class 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 56 class 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_with_open(self): 362 # GzipFile supports the context management protocol 363 with gzip.GzipFile(self.filename, "wb") as f: 364 f.write(b"xxx") 365 f = gzip.GzipFile(self.filename, "rb") 366 f.close() 367 try: 368 with f: 369 pass 370 except ValueError: 371 pass 372 else: 373 self.fail("__enter__ on a closed file didn't raise an exception") 374 try: 375 with gzip.GzipFile(self.filename, "wb") as f: 376 1/0 377 except ZeroDivisionError: 378 pass 379 else: 380 self.fail("1/0 didn't raise an exception") 381 382 def test_zero_padded_file(self): 383 with gzip.GzipFile(self.filename, "wb") as f: 384 f.write(data1 * 50) 385 386 # Pad the file with zeroes 387 with open(self.filename, "ab") as f: 388 f.write(b"\x00" * 50) 389 390 with gzip.GzipFile(self.filename, "rb") as f: 391 d = f.read() 392 self.assertEqual(d, data1 * 50, "Incorrect data in file") 393 394 def test_non_seekable_file(self): 395 uncompressed = data1 * 50 396 buf = UnseekableIO() 397 with gzip.GzipFile(fileobj=buf, mode="wb") as f: 398 f.write(uncompressed) 399 compressed = buf.getvalue() 400 buf = UnseekableIO(compressed) 401 with gzip.GzipFile(fileobj=buf, mode="rb") as f: 402 self.assertEqual(f.read(), uncompressed) 403 404 def test_peek(self): 405 uncompressed = data1 * 200 406 with gzip.GzipFile(self.filename, "wb") as f: 407 f.write(uncompressed) 408 409 def sizes(): 410 while True: 411 for n in range(5, 50, 10): 412 yield n 413 414 with gzip.GzipFile(self.filename, "rb") as f: 415 f.max_read_chunk = 33 416 nread = 0 417 for n in sizes(): 418 s = f.peek(n) 419 if s == b'': 420 break 421 self.assertEqual(f.read(len(s)), s) 422 nread += len(s) 423 self.assertEqual(f.read(100), b'') 424 self.assertEqual(nread, len(uncompressed)) 425 426 def test_textio_readlines(self): 427 # Issue #10791: TextIOWrapper.readlines() fails when wrapping GzipFile. 428 lines = (data1 * 50).decode("ascii").splitlines(keepends=True) 429 self.test_write() 430 with gzip.GzipFile(self.filename, 'r') as f: 431 with io.TextIOWrapper(f, encoding="ascii") as t: 432 self.assertEqual(t.readlines(), lines) 433 434 def test_fileobj_from_fdopen(self): 435 # Issue #13781: Opening a GzipFile for writing fails when using a 436 # fileobj created with os.fdopen(). 437 fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT) 438 with os.fdopen(fd, "wb") as f: 439 with gzip.GzipFile(fileobj=f, mode="w") as g: 440 pass 441 442 def test_fileobj_mode(self): 443 gzip.GzipFile(self.filename, "wb").close() 444 with open(self.filename, "r+b") as f: 445 with gzip.GzipFile(fileobj=f, mode='r') as g: 446 self.assertEqual(g.mode, gzip.READ) 447 with gzip.GzipFile(fileobj=f, mode='w') as g: 448 self.assertEqual(g.mode, gzip.WRITE) 449 with gzip.GzipFile(fileobj=f, mode='a') as g: 450 self.assertEqual(g.mode, gzip.WRITE) 451 with gzip.GzipFile(fileobj=f, mode='x') as g: 452 self.assertEqual(g.mode, gzip.WRITE) 453 with self.assertRaises(ValueError): 454 gzip.GzipFile(fileobj=f, mode='z') 455 for mode in "rb", "r+b": 456 with open(self.filename, mode) as f: 457 with gzip.GzipFile(fileobj=f) as g: 458 self.assertEqual(g.mode, gzip.READ) 459 for mode in "wb", "ab", "xb": 460 if "x" in mode: 461 support.unlink(self.filename) 462 with open(self.filename, mode) as f: 463 with gzip.GzipFile(fileobj=f) as g: 464 self.assertEqual(g.mode, gzip.WRITE) 465 466 def test_bytes_filename(self): 467 str_filename = self.filename 468 try: 469 bytes_filename = str_filename.encode("ascii") 470 except UnicodeEncodeError: 471 self.skipTest("Temporary file name needs to be ASCII") 472 with gzip.GzipFile(bytes_filename, "wb") as f: 473 f.write(data1 * 50) 474 with gzip.GzipFile(bytes_filename, "rb") as f: 475 self.assertEqual(f.read(), data1 * 50) 476 # Sanity check that we are actually operating on the right file. 477 with gzip.GzipFile(str_filename, "rb") as f: 478 self.assertEqual(f.read(), data1 * 50) 479 480 def test_decompress_limited(self): 481 """Decompressed data buffering should be limited""" 482 bomb = gzip.compress(b'\0' * int(2e6), compresslevel=9) 483 self.assertLess(len(bomb), io.DEFAULT_BUFFER_SIZE) 484 485 bomb = io.BytesIO(bomb) 486 decomp = gzip.GzipFile(fileobj=bomb) 487 self.assertEqual(decomp.read(1), b'\0') 488 max_decomp = 1 + io.DEFAULT_BUFFER_SIZE 489 self.assertLessEqual(decomp._buffer.raw.tell(), max_decomp, 490 "Excessive amount of data was decompressed") 491 492 # Testing compress/decompress shortcut functions 493 494 def test_compress(self): 495 for data in [data1, data2]: 496 for args in [(), (1,), (6,), (9,)]: 497 datac = gzip.compress(data, *args) 498 self.assertEqual(type(datac), bytes) 499 with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f: 500 self.assertEqual(f.read(), data) 501 502 def test_decompress(self): 503 for data in (data1, data2): 504 buf = io.BytesIO() 505 with gzip.GzipFile(fileobj=buf, mode="wb") as f: 506 f.write(data) 507 self.assertEqual(gzip.decompress(buf.getvalue()), data) 508 # Roundtrip with compress 509 datac = gzip.compress(data) 510 self.assertEqual(gzip.decompress(datac), data) 511 512 def test_read_truncated(self): 513 data = data1*50 514 # Drop the CRC (4 bytes) and file size (4 bytes). 515 truncated = gzip.compress(data)[:-8] 516 with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f: 517 self.assertRaises(EOFError, f.read) 518 with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f: 519 self.assertEqual(f.read(len(data)), data) 520 self.assertRaises(EOFError, f.read, 1) 521 # Incomplete 10-byte header. 522 for i in range(2, 10): 523 with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f: 524 self.assertRaises(EOFError, f.read, 1) 525 526 def test_read_with_extra(self): 527 # Gzip data with an extra field 528 gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff' 529 b'\x05\x00Extra' 530 b'\x0bI-.\x01\x002\xd1Mx\x04\x00\x00\x00') 531 with gzip.GzipFile(fileobj=io.BytesIO(gzdata)) as f: 532 self.assertEqual(f.read(), b'Test') 533 534 def test_prepend_error(self): 535 # See issue #20875 536 with gzip.open(self.filename, "wb") as f: 537 f.write(data1) 538 with gzip.open(self.filename, "rb") as f: 539 f._buffer.raw._fp.prepend() 540 541 class TestOpen(BaseTest): 542 def test_binary_modes(self): 543 uncompressed = data1 * 50 544 545 with gzip.open(self.filename, "wb") as f: 546 f.write(uncompressed) 547 with open(self.filename, "rb") as f: 548 file_data = gzip.decompress(f.read()) 549 self.assertEqual(file_data, uncompressed) 550 551 with gzip.open(self.filename, "rb") as f: 552 self.assertEqual(f.read(), uncompressed) 553 554 with gzip.open(self.filename, "ab") as f: 555 f.write(uncompressed) 556 with open(self.filename, "rb") as f: 557 file_data = gzip.decompress(f.read()) 558 self.assertEqual(file_data, uncompressed * 2) 559 560 with self.assertRaises(FileExistsError): 561 gzip.open(self.filename, "xb") 562 support.unlink(self.filename) 563 with gzip.open(self.filename, "xb") as f: 564 f.write(uncompressed) 565 with open(self.filename, "rb") as f: 566 file_data = gzip.decompress(f.read()) 567 self.assertEqual(file_data, uncompressed) 568 569 def test_pathlike_file(self): 570 filename = pathlib.Path(self.filename) 571 with gzip.open(filename, "wb") as f: 572 f.write(data1 * 50) 573 with gzip.open(filename, "ab") as f: 574 f.write(data1) 575 with gzip.open(filename) as f: 576 self.assertEqual(f.read(), data1 * 51) 577 578 def test_implicit_binary_modes(self): 579 # Test implicit binary modes (no "b" or "t" in mode string). 580 uncompressed = data1 * 50 581 582 with gzip.open(self.filename, "w") as f: 583 f.write(uncompressed) 584 with open(self.filename, "rb") as f: 585 file_data = gzip.decompress(f.read()) 586 self.assertEqual(file_data, uncompressed) 587 588 with gzip.open(self.filename, "r") as f: 589 self.assertEqual(f.read(), uncompressed) 590 591 with gzip.open(self.filename, "a") as f: 592 f.write(uncompressed) 593 with open(self.filename, "rb") as f: 594 file_data = gzip.decompress(f.read()) 595 self.assertEqual(file_data, uncompressed * 2) 596 597 with self.assertRaises(FileExistsError): 598 gzip.open(self.filename, "x") 599 support.unlink(self.filename) 600 with gzip.open(self.filename, "x") as f: 601 f.write(uncompressed) 602 with open(self.filename, "rb") as f: 603 file_data = gzip.decompress(f.read()) 604 self.assertEqual(file_data, uncompressed) 605 606 def test_text_modes(self): 607 uncompressed = data1.decode("ascii") * 50 608 uncompressed_raw = uncompressed.replace("\n", os.linesep) 609 with gzip.open(self.filename, "wt") as f: 610 f.write(uncompressed) 611 with open(self.filename, "rb") as f: 612 file_data = gzip.decompress(f.read()).decode("ascii") 613 self.assertEqual(file_data, uncompressed_raw) 614 with gzip.open(self.filename, "rt") as f: 615 self.assertEqual(f.read(), uncompressed) 616 with gzip.open(self.filename, "at") as f: 617 f.write(uncompressed) 618 with open(self.filename, "rb") as f: 619 file_data = gzip.decompress(f.read()).decode("ascii") 620 self.assertEqual(file_data, uncompressed_raw * 2) 621 622 def test_fileobj(self): 623 uncompressed_bytes = data1 * 50 624 uncompressed_str = uncompressed_bytes.decode("ascii") 625 compressed = gzip.compress(uncompressed_bytes) 626 with gzip.open(io.BytesIO(compressed), "r") as f: 627 self.assertEqual(f.read(), uncompressed_bytes) 628 with gzip.open(io.BytesIO(compressed), "rb") as f: 629 self.assertEqual(f.read(), uncompressed_bytes) 630 with gzip.open(io.BytesIO(compressed), "rt") as f: 631 self.assertEqual(f.read(), uncompressed_str) 632 633 def test_bad_params(self): 634 # Test invalid parameter combinations. 635 with self.assertRaises(TypeError): 636 gzip.open(123.456) 637 with self.assertRaises(ValueError): 638 gzip.open(self.filename, "wbt") 639 with self.assertRaises(ValueError): 640 gzip.open(self.filename, "xbt") 641 with self.assertRaises(ValueError): 642 gzip.open(self.filename, "rb", encoding="utf-8") 643 with self.assertRaises(ValueError): 644 gzip.open(self.filename, "rb", errors="ignore") 645 with self.assertRaises(ValueError): 646 gzip.open(self.filename, "rb", newline="\n") 647 648 def test_encoding(self): 649 # Test non-default encoding. 650 uncompressed = data1.decode("ascii") * 50 651 uncompressed_raw = uncompressed.replace("\n", os.linesep) 652 with gzip.open(self.filename, "wt", encoding="utf-16") as f: 653 f.write(uncompressed) 654 with open(self.filename, "rb") as f: 655 file_data = gzip.decompress(f.read()).decode("utf-16") 656 self.assertEqual(file_data, uncompressed_raw) 657 with gzip.open(self.filename, "rt", encoding="utf-16") as f: 658 self.assertEqual(f.read(), uncompressed) 659 660 def test_encoding_error_handler(self): 661 # Test with non-default encoding error handler. 662 with gzip.open(self.filename, "wb") as f: 663 f.write(b"foo\xffbar") 664 with gzip.open(self.filename, "rt", encoding="ascii", errors="ignore") \ 665 as f: 666 self.assertEqual(f.read(), "foobar") 667 668 def test_newline(self): 669 # Test with explicit newline (universal newline mode disabled). 670 uncompressed = data1.decode("ascii") * 50 671 with gzip.open(self.filename, "wt", newline="\n") as f: 672 f.write(uncompressed) 673 with gzip.open(self.filename, "rt", newline="\r") as f: 674 self.assertEqual(f.readlines(), [uncompressed]) 675 676 677 def create_and_remove_directory(directory): 678 def decorator(function): 679 @functools.wraps(function) 680 def wrapper(*args, **kwargs): 681 os.makedirs(directory) 682 try: 683 return function(*args, **kwargs) 684 finally: 685 support.rmtree(directory) 686 return wrapper 687 return decorator 688 689 690 class TestCommandLine(unittest.TestCase): 691 data = b'This is a simple test with gzip' 692 693 def test_decompress_stdin_stdout(self): 694 with io.BytesIO() as bytes_io: 695 with gzip.GzipFile(fileobj=bytes_io, mode='wb') as gzip_file: 696 gzip_file.write(self.data) 697 698 args = sys.executable, '-m', 'gzip', '-d' 699 with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc: 700 out, err = proc.communicate(bytes_io.getvalue()) 701 702 self.assertEqual(err, b'') 703 self.assertEqual(out, self.data) 704 705 @create_and_remove_directory(TEMPDIR) 706 def test_decompress_infile_outfile(self): 707 gzipname = os.path.join(TEMPDIR, 'testgzip.gz') 708 self.assertFalse(os.path.exists(gzipname)) 709 710 with gzip.open(gzipname, mode='wb') as fp: 711 fp.write(self.data) 712 rc, out, err = assert_python_ok('-m', 'gzip', '-d', gzipname) 713 714 with open(os.path.join(TEMPDIR, "testgzip"), "rb") as gunziped: 715 self.assertEqual(gunziped.read(), self.data) 716 717 self.assertTrue(os.path.exists(gzipname)) 718 self.assertEqual(rc, 0) 719 self.assertEqual(out, b'') 720 self.assertEqual(err, b'') 721 722 def test_decompress_infile_outfile_error(self): 723 rc, out, err = assert_python_ok('-m', 'gzip', '-d', 'thisisatest.out') 724 self.assertIn(b"filename doesn't end in .gz:", out) 725 self.assertEqual(rc, 0) 726 self.assertEqual(err, b'') 727 728 @create_and_remove_directory(TEMPDIR) 729 def test_compress_stdin_outfile(self): 730 args = sys.executable, '-m', 'gzip' 731 with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc: 732 out, err = proc.communicate(self.data) 733 734 self.assertEqual(err, b'') 735 self.assertEqual(out[:2], b"\x1f\x8b") 736 737 @create_and_remove_directory(TEMPDIR) 738 def test_compress_infile_outfile(self): 739 local_testgzip = os.path.join(TEMPDIR, 'testgzip') 740 gzipname = local_testgzip + '.gz' 741 self.assertFalse(os.path.exists(gzipname)) 742 743 with open(local_testgzip, 'wb') as fp: 744 fp.write(self.data) 745 746 rc, out, err = assert_python_ok('-m', 'gzip', local_testgzip) 747 748 self.assertTrue(os.path.exists(gzipname)) 749 self.assertEqual(rc, 0) 750 self.assertEqual(out, b'') 751 self.assertEqual(err, b'') 752 753 754 def test_main(verbose=None): 755 support.run_unittest(TestGzip, TestOpen, TestCommandLine) 756 757 758 if __name__ == "__main__": 759 test_main(verbose=True) 760