[Python-3000-checkins] r55500 - in python/branches/py3k-struni: Lib/py_compile.py Lib/test/test_zipfile.py Lib/zipfile.py Modules/_struct.c
guido.van.rossum
python-3000-checkins at python.org
Tue May 22 03:29:39 CEST 2007
Author: guido.van.rossum
Date: Tue May 22 03:29:33 2007
New Revision: 55500
Modified:
python/branches/py3k-struni/Lib/py_compile.py
python/branches/py3k-struni/Lib/test/test_zipfile.py
python/branches/py3k-struni/Lib/zipfile.py
python/branches/py3k-struni/Modules/_struct.c
Log:
Make test_zipfile pass.
The zipfile module now does all I/O in binary mode using bytes.
(Maybe we should support wrapping a TextIOWrapper around it
when text mode reading is requested?)
Even the password is a bytes array now.
Had to fix py_compile.py to use bytes while I was at it.
The _struct needed a patch to support bytes, str8 and str
for the 's' and 'p' formats.
Modified: python/branches/py3k-struni/Lib/py_compile.py
==============================================================================
--- python/branches/py3k-struni/Lib/py_compile.py (original)
+++ python/branches/py3k-struni/Lib/py_compile.py Tue May 22 03:29:33 2007
@@ -72,10 +72,10 @@
def wr_long(f, x):
"""Internal; write a 32-bit int to a file in little-endian order."""
- f.write(chr( x & 0xff))
- f.write(chr((x >> 8) & 0xff))
- f.write(chr((x >> 16) & 0xff))
- f.write(chr((x >> 24) & 0xff))
+ f.write(bytes([x & 0xff,
+ (x >> 8) & 0xff,
+ (x >> 16) & 0xff,
+ (x >> 24) & 0xff]))
def compile(file, cfile=None, dfile=None, doraise=False):
"""Byte-compile one Python source file to Python bytecode.
@@ -133,7 +133,7 @@
if cfile is None:
cfile = file + (__debug__ and 'c' or 'o')
fc = open(cfile, 'wb')
- fc.write('\0\0\0\0')
+ fc.write(b'\0\0\0\0')
wr_long(fc, timestamp)
marshal.dump(codeobject, fc)
fc.flush()
Modified: python/branches/py3k-struni/Lib/test/test_zipfile.py
==============================================================================
--- python/branches/py3k-struni/Lib/test/test_zipfile.py (original)
+++ python/branches/py3k-struni/Lib/test/test_zipfile.py Tue May 22 03:29:33 2007
@@ -3,10 +3,8 @@
import zlib
except ImportError:
zlib = None
+import zipfile, os, unittest, sys, shutil, struct, io
-import zipfile, os, unittest, sys, shutil, struct
-
-from StringIO import StringIO
from tempfile import TemporaryFile
from random import randint, random
@@ -18,9 +16,10 @@
class TestsWithSourceFile(unittest.TestCase):
def setUp(self):
- self.line_gen = ("Zipfile test line %d. random float: %f" % (i, random())
- for i in range(FIXEDTEST_SIZE))
- self.data = '\n'.join(self.line_gen) + '\n'
+ self.line_gen = (bytes("Zipfile test line %d. random float: %f" %
+ (i, random()))
+ for i in range(FIXEDTEST_SIZE))
+ self.data = b'\n'.join(self.line_gen) + b'\n'
# Make a source file with some lines
fp = open(TESTFN, "wb")
@@ -45,14 +44,8 @@
self.assertEqual(zipfp.read("strfile"), self.data)
# Print the ZIP directory
- fp = StringIO()
- stdout = sys.stdout
- try:
- sys.stdout = fp
-
- zipfp.printdir()
- finally:
- sys.stdout = stdout
+ fp = io.StringIO()
+ zipfp.printdir(file=fp)
directory = fp.getvalue()
lines = directory.splitlines()
@@ -95,7 +88,7 @@
zipfp.close()
def testStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipTest(f, zipfile.ZIP_STORED)
def zipOpenTest(self, f, compression):
@@ -119,12 +112,12 @@
break
zipdata2.append(read_data)
- self.assertEqual(''.join(zipdata1), self.data)
- self.assertEqual(''.join(zipdata2), self.data)
+ self.assertEqual(b''.join(zipdata1), self.data)
+ self.assertEqual(b''.join(zipdata2), self.data)
zipfp.close()
def testOpenStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipOpenTest(f, zipfile.ZIP_STORED)
def zipRandomOpenTest(self, f, compression):
@@ -140,11 +133,11 @@
break
zipdata1.append(read_data)
- self.assertEqual(''.join(zipdata1), self.data)
+ self.assertEqual(b''.join(zipdata1), self.data)
zipfp.close()
def testRandomOpenStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipRandomOpenTest(f, zipfile.ZIP_STORED)
def zipReadlineTest(self, f, compression):
@@ -181,40 +174,40 @@
zipfp.close()
def testReadlineStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipReadlineTest(f, zipfile.ZIP_STORED)
def testReadlinesStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipReadlinesTest(f, zipfile.ZIP_STORED)
def testIterlinesStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipIterlinesTest(f, zipfile.ZIP_STORED)
if zlib:
def testDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipTest(f, zipfile.ZIP_DEFLATED)
def testOpenDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipOpenTest(f, zipfile.ZIP_DEFLATED)
def testRandomOpenDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipRandomOpenTest(f, zipfile.ZIP_DEFLATED)
def testReadlineDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipReadlineTest(f, zipfile.ZIP_DEFLATED)
def testReadlinesDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipReadlinesTest(f, zipfile.ZIP_DEFLATED)
def testIterlinesDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipIterlinesTest(f, zipfile.ZIP_DEFLATED)
def testLowCompression(self):
@@ -227,8 +220,8 @@
# Get an open object for strfile
zipfp = zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_DEFLATED)
openobj = zipfp.open("strfile")
- self.assertEqual(openobj.read(1), '1')
- self.assertEqual(openobj.read(1), '2')
+ self.assertEqual(openobj.read(1), b'1')
+ self.assertEqual(openobj.read(1), b'2')
def testAbsoluteArcnames(self):
zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED)
@@ -251,8 +244,9 @@
self._limit = zipfile.ZIP64_LIMIT
zipfile.ZIP64_LIMIT = 5
- line_gen = ("Test of zipfile line %d." % i for i in range(0, FIXEDTEST_SIZE))
- self.data = '\n'.join(line_gen)
+ line_gen = (bytes("Test of zipfile line %d." % i)
+ for i in range(0, FIXEDTEST_SIZE))
+ self.data = b'\n'.join(line_gen)
# Make a source file with some lines
fp = open(TESTFN, "wb")
@@ -272,7 +266,7 @@
zipfp.close()
def testLargeFileException(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.largeFileExceptionTest(f, zipfile.ZIP_STORED)
self.largeFileExceptionTest2(f, zipfile.ZIP_STORED)
@@ -291,14 +285,8 @@
self.assertEqual(zipfp.read("strfile"), self.data)
# Print the ZIP directory
- fp = StringIO()
- stdout = sys.stdout
- try:
- sys.stdout = fp
-
- zipfp.printdir()
- finally:
- sys.stdout = stdout
+ fp = io.StringIO()
+ zipfp.printdir(fp)
directory = fp.getvalue()
lines = directory.splitlines()
@@ -343,13 +331,13 @@
zipfp.close()
def testStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipTest(f, zipfile.ZIP_STORED)
if zlib:
def testDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipTest(f, zipfile.ZIP_DEFLATED)
def testAbsoluteArcnames(self):
@@ -440,7 +428,7 @@
os.unlink(TESTFN)
filename = 'testfile.txt'
- content = 'hello, world. this is some content.'
+ content = b'hello, world. this is some content.'
try:
zf = zipfile.ZipFile(TESTFN, 'a')
@@ -483,7 +471,7 @@
# This test checks that the is_zipfile function correctly identifies
# a file that is a zip file
zipf = zipfile.ZipFile(TESTFN, mode="w")
- zipf.writestr("foo.txt", "O, for a Muse of Fire!")
+ zipf.writestr("foo.txt", b"O, for a Muse of Fire!")
zipf.close()
chk = zipfile.is_zipfile(TESTFN)
self.assert_(chk is True)
@@ -504,7 +492,7 @@
def testClosedZipRaisesRuntimeError(self):
# Verify that testzip() doesn't swallow inappropriate exceptions.
- data = StringIO()
+ data = io.BytesIO()
zipf = zipfile.ZipFile(data, mode="w")
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
zipf.close()
@@ -525,15 +513,15 @@
# ZIP file
data = (
- 'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00'
- '\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y'
- '\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl'
- 'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00'
- '\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81'
- '\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00'
- '\x00\x00L\x00\x00\x00\x00\x00' )
+ b'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00'
+ b'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y'
+ b'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl'
+ b'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00'
+ b'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81'
+ b'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00'
+ b'\x00\x00L\x00\x00\x00\x00\x00' )
- plain = 'zipfile.py encryption test'
+ plain = b'zipfile.py encryption test'
def setUp(self):
fp = open(TESTFN, "wb")
@@ -551,18 +539,19 @@
self.assertRaises(RuntimeError, self.zip.read, "test.txt")
def testBadPassword(self):
- self.zip.setpassword("perl")
+ self.zip.setpassword(b"perl")
self.assertRaises(RuntimeError, self.zip.read, "test.txt")
def testGoodPassword(self):
- self.zip.setpassword("python")
+ self.zip.setpassword(b"python")
self.assertEquals(self.zip.read("test.txt"), self.plain)
class TestsWithRandomBinaryFiles(unittest.TestCase):
def setUp(self):
datacount = randint(16, 64)*1024 + randint(1, 1024)
- self.data = ''.join((struct.pack('<f', random()*randint(-1000, 1000)) for i in range(datacount)))
+ self.data = b''.join(struct.pack('<f', random()*randint(-1000, 1000))
+ for i in range(datacount))
# Make a source file with some lines
fp = open(TESTFN, "wb")
@@ -592,7 +581,7 @@
zipfp.close()
def testStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipTest(f, zipfile.ZIP_STORED)
def zipOpenTest(self, f, compression):
@@ -616,17 +605,17 @@
break
zipdata2.append(read_data)
- testdata1 = ''.join(zipdata1)
+ testdata1 = b''.join(zipdata1)
self.assertEqual(len(testdata1), len(self.data))
self.assertEqual(testdata1, self.data)
- testdata2 = ''.join(zipdata2)
+ testdata2 = b''.join(zipdata2)
self.assertEqual(len(testdata1), len(self.data))
self.assertEqual(testdata1, self.data)
zipfp.close()
def testOpenStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipOpenTest(f, zipfile.ZIP_STORED)
def zipRandomOpenTest(self, f, compression):
@@ -642,13 +631,13 @@
break
zipdata1.append(read_data)
- testdata = ''.join(zipdata1)
+ testdata = b''.join(zipdata1)
self.assertEqual(len(testdata), len(self.data))
self.assertEqual(testdata, self.data)
zipfp.close()
def testRandomOpenStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zipRandomOpenTest(f, zipfile.ZIP_STORED)
class TestsWithMultipleOpens(unittest.TestCase):
@@ -682,8 +671,8 @@
data2 = zopen2.read(500)
data1 += zopen1.read(500)
data2 += zopen2.read(500)
- self.assertEqual(data1, '1'*FIXEDTEST_SIZE)
- self.assertEqual(data2, '2'*FIXEDTEST_SIZE)
+ self.assertEqual(data1, b'1'*FIXEDTEST_SIZE)
+ self.assertEqual(data2, b'2'*FIXEDTEST_SIZE)
zipf.close()
def testInterleaved(self):
@@ -696,8 +685,8 @@
data2 = zopen2.read(500)
data1 += zopen1.read(500)
data2 += zopen2.read(500)
- self.assertEqual(data1, '1'*FIXEDTEST_SIZE)
- self.assertEqual(data2, '2'*FIXEDTEST_SIZE)
+ self.assertEqual(data1, b'1'*FIXEDTEST_SIZE)
+ self.assertEqual(data2, b'2'*FIXEDTEST_SIZE)
zipf.close()
def tearDown(self):
@@ -706,13 +695,19 @@
class UniversalNewlineTests(unittest.TestCase):
def setUp(self):
- self.line_gen = ["Test of zipfile line %d." % i for i in range(FIXEDTEST_SIZE)]
+ self.line_gen = [bytes("Test of zipfile line %d." % i)
+ for i in range(FIXEDTEST_SIZE)]
self.seps = ('\r', '\r\n', '\n')
self.arcdata, self.arcfiles = {}, {}
for n, s in enumerate(self.seps):
- self.arcdata[s] = s.join(self.line_gen) + s
+ b = s.encode("ascii")
+ self.arcdata[s] = b.join(self.line_gen) + b
self.arcfiles[s] = '%s-%d' % (TESTFN, n)
- file(self.arcfiles[s], "wb").write(self.arcdata[s])
+ f = open(self.arcfiles[s], "wb")
+ try:
+ f.write(self.arcdata[s])
+ finally:
+ f.close()
def makeTestArchive(self, f, compression):
# Create the ZIP archive
@@ -741,7 +736,7 @@
zipopen = zipfp.open(fn, "rU")
for line in self.line_gen:
linedata = zipopen.readline()
- self.assertEqual(linedata, line + '\n')
+ self.assertEqual(linedata, line + b'\n')
zipfp.close()
@@ -753,7 +748,7 @@
for sep, fn in self.arcfiles.items():
ziplines = zipfp.open(fn, "rU").readlines()
for line, zipline in zip(self.line_gen, ziplines):
- self.assertEqual(zipline, line + '\n')
+ self.assertEqual(zipline, line + b'\n')
zipfp.close()
@@ -764,41 +759,41 @@
zipfp = zipfile.ZipFile(f, "r")
for sep, fn in self.arcfiles.items():
for line, zipline in zip(self.line_gen, zipfp.open(fn, "rU")):
- self.assertEqual(zipline, line + '\n')
+ self.assertEqual(zipline, line + b'\n')
zipfp.close()
def testReadStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readTest(f, zipfile.ZIP_STORED)
def testReadlineStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlineTest(f, zipfile.ZIP_STORED)
def testReadlinesStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlinesTest(f, zipfile.ZIP_STORED)
def testIterlinesStored(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.iterlinesTest(f, zipfile.ZIP_STORED)
if zlib:
def testReadDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readTest(f, zipfile.ZIP_DEFLATED)
def testReadlineDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlineTest(f, zipfile.ZIP_DEFLATED)
def testReadlinesDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlinesTest(f, zipfile.ZIP_DEFLATED)
def testIterlinesDeflated(self):
- for f in (TESTFN2, TemporaryFile(), StringIO()):
+ for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.iterlinesTest(f, zipfile.ZIP_DEFLATED)
def tearDown(self):
Modified: python/branches/py3k-struni/Lib/zipfile.py
==============================================================================
--- python/branches/py3k-struni/Lib/zipfile.py (original)
+++ python/branches/py3k-struni/Lib/zipfile.py Tue May 22 03:29:33 2007
@@ -1,5 +1,7 @@
"""
Read and write ZIP files.
+
+XXX references to utf-8 need further investigation.
"""
import struct, os, time, sys
import binascii, io
@@ -33,15 +35,15 @@
# Here are some struct module formats for reading headers
structEndArchive = "<4s4H2lH" # 9 items, end of archive, 22 bytes
-stringEndArchive = "PK\005\006" # magic number for end of archive record
+stringEndArchive = b"PK\005\006" # magic number for end of archive record
structCentralDir = "<4s4B4HlLL5HLl"# 19 items, central directory, 46 bytes
-stringCentralDir = "PK\001\002" # magic number for central directory
+stringCentralDir = b"PK\001\002" # magic number for central directory
structFileHeader = "<4s2B4HlLL2H" # 12 items, file header record, 30 bytes
-stringFileHeader = "PK\003\004" # magic number for file header
+stringFileHeader = b"PK\003\004" # magic number for file header
structEndArchive64Locator = "<4slql" # 4 items, locate Zip64 header, 20 bytes
-stringEndArchive64Locator = "PK\x06\x07" # magic token for locator header
+stringEndArchive64Locator = b"PK\x06\x07" # magic token for locator header
structEndArchive64 = "<4sqhhllqqqq" # 10 items, end of archive (Zip64), 56 bytes
-stringEndArchive64 = "PK\x06\x06" # magic token for Zip64 header
+stringEndArchive64 = b"PK\x06\x06" # magic token for Zip64 header
# indexes of entries in the central directory structure
@@ -82,7 +84,7 @@
def is_zipfile(filename):
"""Quickly see if file is a ZIP file by checking the magic number."""
try:
- fpin = open(filename, "rb")
+ fpin = io.open(filename, "rb")
endrec = _EndRecData(fpin)
fpin.close()
if endrec:
@@ -206,8 +208,8 @@
self.date_time = date_time # year, month, day, hour, min, sec
# Standard values:
self.compress_type = ZIP_STORED # Type of compression for the file
- self.comment = "" # Comment for each file
- self.extra = "" # ZIP extra data
+ self.comment = b"" # Comment for each file
+ self.extra = b"" # ZIP extra data
if sys.platform == 'win32':
self.create_system = 0 # System which created ZIP archive
else:
@@ -257,7 +259,7 @@
self.compress_type, dostime, dosdate, CRC,
compress_size, file_size,
len(self.filename), len(extra))
- return header + self.filename + extra
+ return header + self.filename.encode("utf-8") + extra
def _decodeExtra(self):
# Try to decode the extra field.
@@ -331,7 +333,7 @@
def _crc32(self, ch, crc):
"""Compute the CRC32 primitive on one byte."""
- return ((crc >> 8) & 0xffffff) ^ self.crctable[(crc ^ ord(ch)) & 0xff]
+ return ((crc >> 8) & 0xffffff) ^ self.crctable[(crc ^ ch) & 0xff]
def __init__(self, pwd):
self.key0 = 305419896
@@ -344,20 +346,13 @@
self.key0 = self._crc32(c, self.key0)
self.key1 = (self.key1 + (self.key0 & 255)) & 4294967295
self.key1 = (self.key1 * 134775813 + 1) & 4294967295
- self.key2 = self._crc32(chr((self.key1 >> 24) & 255), self.key2)
+ self.key2 = self._crc32((self.key1 >> 24) & 255, self.key2)
def __call__(self, c):
"""Decrypt a single character."""
- # XXX When this is called with a byte instead of a char, ord()
- # isn't needed. Don't die in that case. In the future we should
- # just leave this out, once we're always using bytes.
- try:
- c = ord(c)
- except TypeError:
- pass
+ assert isinstance(c, int)
k = self.key2 | 2
c = c ^ (((k * (k^1)) >> 8) & 255)
- c = chr(c)
self._UpdateKeys(c)
return c
@@ -370,13 +365,13 @@
self.fileobj = fileobj
self.decrypter = decrypt
self.bytes_read = 0
- self.rawbuffer = ''
- self.readbuffer = ''
- self.linebuffer = ''
+ self.rawbuffer = b''
+ self.readbuffer = b''
+ self.linebuffer = b''
self.eof = False
self.univ_newlines = False
- self.nlSeps = ("\n", )
- self.lastdiscard = ''
+ self.nlSeps = (b"\n", )
+ self.lastdiscard = b''
self.compress_type = zipinfo.compress_type
self.compress_size = zipinfo.compress_size
@@ -394,9 +389,9 @@
self.univ_newlines = univ_newlines
# pick line separator char(s) based on universal newlines flag
- self.nlSeps = ("\n", )
+ self.nlSeps = (b"\n", )
if self.univ_newlines:
- self.nlSeps = ("\r\n", "\r", "\n")
+ self.nlSeps = (b"\r\n", b"\r", b"\n")
def __iter__(self):
return self
@@ -417,7 +412,7 @@
# ugly check for cases where half of an \r\n pair was
# read on the last pass, and the \r was discarded. In this
# case we just throw away the \n at the start of the buffer.
- if (self.lastdiscard, self.linebuffer[0]) == ('\r','\n'):
+ if (self.lastdiscard, self.linebuffer[0]) == (b'\r', b'\n'):
self.linebuffer = self.linebuffer[1:]
for sep in self.nlSeps:
@@ -435,7 +430,7 @@
if size < 0:
size = sys.maxint
elif size == 0:
- return ''
+ return b''
# check for a newline already in buffer
nl, nllen = self._checkfornewline()
@@ -461,7 +456,7 @@
# so return current buffer
if nl < 0:
s = self.linebuffer
- self.linebuffer = ''
+ self.linebuffer = b''
return s
buf = self.linebuffer[:nl]
@@ -470,7 +465,7 @@
# line is always returned with \n as newline char (except possibly
# for a final incomplete line in the file, which is handled above).
- return buf + "\n"
+ return buf + b"\n"
def readlines(self, sizehint = -1):
"""Return a list with all (following) lines. The sizehint parameter
@@ -516,18 +511,23 @@
# try to read from file (if necessary)
if bytesToRead > 0:
- bytes = self.fileobj.read(bytesToRead)
- self.bytes_read += len(bytes)
- self.rawbuffer += bytes
+ data = self.fileobj.read(bytesToRead)
+ self.bytes_read += len(data)
+ try:
+ self.rawbuffer += data
+ except:
+ print(repr(self.fileobj), repr(self.rawbuffer),
+ repr(data))
+ raise
# handle contents of raw buffer
if self.rawbuffer:
newdata = self.rawbuffer
- self.rawbuffer = ''
+ self.rawbuffer = b''
# decrypt new data if we were given an object to handle that
if newdata and self.decrypter is not None:
- newdata = ''.join(map(self.decrypter, newdata))
+ newdata = bytes(map(self.decrypter, newdata))
# decompress newly read data if necessary
if newdata and self.compress_type == ZIP_DEFLATED:
@@ -546,13 +546,13 @@
# return what the user asked for
if size is None or len(self.readbuffer) <= size:
- bytes = self.readbuffer
- self.readbuffer = ''
+ data = self.readbuffer
+ self.readbuffer = b''
else:
- bytes = self.readbuffer[:size]
+ data = self.readbuffer[:size]
self.readbuffer = self.readbuffer[size:]
- return bytes
+ return data
class ZipFile:
@@ -593,15 +593,16 @@
# Check if we were passed a file-like object
if isinstance(file, basestring):
+ # No, it's a filename
self._filePassed = 0
self.filename = file
modeDict = {'r' : 'rb', 'w': 'wb', 'a' : 'r+b'}
try:
- self.fp = open(file, modeDict[mode])
+ self.fp = io.open(file, modeDict[mode])
except IOError:
if mode == 'a':
mode = key = 'w'
- self.fp = open(file, modeDict[mode])
+ self.fp = io.open(file, modeDict[mode])
else:
raise
else:
@@ -661,7 +662,7 @@
self.start_dir = offset_cd + concat
fp.seek(self.start_dir, 0)
data = fp.read(size_cd)
- fp = io.StringIO(data)
+ fp = io.BytesIO(data)
total = 0
while total < size_cd:
centdir = fp.read(46)
@@ -673,7 +674,7 @@
print(centdir)
filename = fp.read(centdir[_CD_FILENAME_LENGTH])
# Create ZipInfo instance to store file information
- x = ZipInfo(filename)
+ x = ZipInfo(str(filename))
x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH])
x.comment = fp.read(centdir[_CD_COMMENT_LENGTH])
total = (total + centdir[_CD_FILENAME_LENGTH]
@@ -708,12 +709,16 @@
archive."""
return self.filelist
- def printdir(self):
+ def printdir(self, file=None):
"""Print a table of contents for the zip file."""
- print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"))
+ if file is None:
+ file = sys.stdout
+ print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"),
+ file=file)
for zinfo in self.filelist:
date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time
- print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size))
+ print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size),
+ file=file)
def testzip(self):
"""Read all the files and check the CRC."""
@@ -730,6 +735,7 @@
def setpassword(self, pwd):
"""Set default password for encrypted files."""
+ assert isinstance(pwd, bytes)
self.pwd = pwd
def read(self, name, pwd=None):
@@ -749,7 +755,7 @@
if self._filePassed:
zef_file = self.fp
else:
- zef_file = open(self.filename, 'rb')
+ zef_file = io.open(self.filename, 'rb')
# Get info object for name
zinfo = self.getinfo(name)
@@ -768,9 +774,9 @@
if fheader[_FH_EXTRA_FIELD_LENGTH]:
zef_file.read(fheader[_FH_EXTRA_FIELD_LENGTH])
- if fname != zinfo.orig_filename:
+ if fname != zinfo.orig_filename.encode("utf-8"):
raise BadZipfile, \
- 'File name in directory "%s" and header "%s" differ.' % (
+ 'File name in directory %r and header %r differ.' % (
zinfo.orig_filename, fname)
# check for encrypted flag & handle password
@@ -790,7 +796,7 @@
# and is used to check the correctness of the password.
bytes = zef_file.read(12)
h = map(zd, bytes[0:12])
- if ord(h[11]) != ((zinfo.CRC>>24)&255):
+ if h[11] != ((zinfo.CRC>>24) & 255):
raise RuntimeError, "Bad password for file %s" % name
# build and return a ZipExtFile
@@ -852,7 +858,7 @@
self._writecheck(zinfo)
self._didModify = True
- fp = open(filename, "rb")
+ fp = io.open(filename, "rb")
# Must overwrite CRC and sizes with correct data later
zinfo.CRC = CRC = 0
zinfo.compress_size = compress_size = 0
@@ -982,7 +988,7 @@
0, zinfo.internal_attr, zinfo.external_attr,
header_offset)
self.fp.write(centdir)
- self.fp.write(zinfo.filename)
+ self.fp.write(zinfo.filename.encode("utf-8"))
self.fp.write(extra_data)
self.fp.write(zinfo.comment)
@@ -1163,7 +1169,7 @@
tgtdir = os.path.dirname(tgt)
if not os.path.exists(tgtdir):
os.makedirs(tgtdir)
- fp = open(tgt, 'wb')
+ fp = io.open(tgt, 'wb')
fp.write(zf.read(path))
fp.close()
zf.close()
Modified: python/branches/py3k-struni/Modules/_struct.c
==============================================================================
--- python/branches/py3k-struni/Modules/_struct.c (original)
+++ python/branches/py3k-struni/Modules/_struct.c Tue May 22 03:29:33 2007
@@ -1635,27 +1635,57 @@
const formatdef *e = code->fmtdef;
char *res = buf + code->offset;
if (e->format == 's') {
- if (!PyString_Check(v)) {
+ int isstring;
+ void *p;
+ if (PyUnicode_Check(v)) {
+ v = _PyUnicode_AsDefaultEncodedString(v, NULL);
+ if (v == NULL)
+ return -1;
+ }
+ isstring = PyString_Check(v);
+ if (!isstring && !PyBytes_Check(v)) {
PyErr_SetString(StructError,
"argument for 's' must be a string");
return -1;
}
- n = PyString_GET_SIZE(v);
+ if (isstring) {
+ n = PyString_GET_SIZE(v);
+ p = PyString_AS_STRING(v);
+ }
+ else {
+ n = PyBytes_GET_SIZE(v);
+ p = PyBytes_AS_STRING(v);
+ }
if (n > code->size)
n = code->size;
if (n > 0)
- memcpy(res, PyString_AS_STRING(v), n);
+ memcpy(res, p, n);
} else if (e->format == 'p') {
- if (!PyString_Check(v)) {
+ int isstring;
+ void *p;
+ if (PyUnicode_Check(v)) {
+ v = _PyUnicode_AsDefaultEncodedString(v, NULL);
+ if (v == NULL)
+ return -1;
+ }
+ isstring = PyString_Check(v);
+ if (!isstring && !PyBytes_Check(v)) {
PyErr_SetString(StructError,
"argument for 'p' must be a string");
return -1;
}
- n = PyString_GET_SIZE(v);
+ if (isstring) {
+ n = PyString_GET_SIZE(v);
+ p = PyString_AS_STRING(v);
+ }
+ else {
+ n = PyBytes_GET_SIZE(v);
+ p = PyBytes_AS_STRING(v);
+ }
if (n > (code->size - 1))
n = code->size - 1;
if (n > 0)
- memcpy(res + 1, PyString_AS_STRING(v), n);
+ memcpy(res + 1, p, n);
if (n > 255)
n = 255;
*res = Py_SAFE_DOWNCAST(n, Py_ssize_t, unsigned char);
More information about the Python-3000-checkins
mailing list