#---------------------------------------------------------
# tar constants
#---------------------------------------------------------
-NUL = "\0" # the null character
+NUL = b"\0" # the null character
BLOCKSIZE = 512 # length of processing blocks
RECORDSIZE = BLOCKSIZE * 20 # length of records
-GNU_MAGIC = "ustar \0" # magic gnu tar string
-POSIX_MAGIC = "ustar\x0000" # magic posix tar string
+GNU_MAGIC = b"ustar \0" # magic gnu tar string
+POSIX_MAGIC = b"ustar\x0000" # magic posix tar string
LENGTH_NAME = 100 # maximum length of a filename
LENGTH_LINK = 100 # maximum length of a linkname
LENGTH_PREFIX = 155 # maximum length of the prefix field
-REGTYPE = "0" # regular file
-AREGTYPE = "\0" # regular file
-LNKTYPE = "1" # link (inside tarfile)
-SYMTYPE = "2" # symbolic link
-CHRTYPE = "3" # character special device
-BLKTYPE = "4" # block special device
-DIRTYPE = "5" # directory
-FIFOTYPE = "6" # fifo special device
-CONTTYPE = "7" # contiguous file
+REGTYPE = b"0" # regular file
+AREGTYPE = b"\0" # regular file
+LNKTYPE = b"1" # link (inside tarfile)
+SYMTYPE = b"2" # symbolic link
+CHRTYPE = b"3" # character special device
+BLKTYPE = b"4" # block special device
+DIRTYPE = b"5" # directory
+FIFOTYPE = b"6" # fifo special device
+CONTTYPE = b"7" # contiguous file
-GNUTYPE_LONGNAME = "L" # GNU tar longname
-GNUTYPE_LONGLINK = "K" # GNU tar longlink
-GNUTYPE_SPARSE = "S" # GNU tar sparse file
+GNUTYPE_LONGNAME = b"L" # GNU tar longname
+GNUTYPE_LONGLINK = b"K" # GNU tar longlink
+GNUTYPE_SPARSE = b"S" # GNU tar sparse file
-XHDTYPE = "x" # POSIX.1-2001 extended header
-XGLTYPE = "g" # POSIX.1-2001 global header
-SOLARIS_XHDTYPE = "X" # Solaris extended header
+XHDTYPE = b"x" # POSIX.1-2001 extended header
+XGLTYPE = b"g" # POSIX.1-2001 global header
+SOLARIS_XHDTYPE = b"X" # Solaris extended header
USTAR_FORMAT = 0 # POSIX.1-1988 (ustar) format
GNU_FORMAT = 1 # GNU tar format
# Some useful functions
#---------------------------------------------------------
-def stn(s, length):
- """Convert a python string to a null-terminated string buffer.
+def stn(s, length, encoding, errors):
+ """Convert a string to a null-terminated bytes object.
"""
+ s = s.encode(encoding, errors)
return s[:length] + (length - len(s)) * NUL
-def nts(s):
- """Convert a null-terminated string field to a python string.
+def nts(s, encoding, errors):
+ """Convert a null-terminated bytes object to a string.
"""
- # Use the string up to the first null char.
- p = s.find("\0")
- if p == -1:
- return s
- return s[:p]
+ p = s.find(b"\0")
+ if p != -1:
+ s = s[:p]
+ return s.decode(encoding, errors)
def nti(s):
"""Convert a number field to a python number.
# itn() below.
if s[0] != chr(0o200):
try:
- n = int(nts(s) or "0", 8)
+ n = int(nts(s, "ascii", "strict") or "0", 8)
except ValueError:
raise HeaderError("invalid header")
else:
# encoding, the following digits-1 bytes are a big-endian
# representation. This allows values up to (256**(digits-1))-1.
if 0 <= n < 8 ** (digits - 1):
- s = "%0*o" % (digits - 1, n) + NUL
+ s = bytes("%0*o" % (digits - 1, n)) + NUL
else:
if format != GNU_FORMAT or n >= 256 ** (digits - 1):
raise ValueError("overflow in number field")
# this could raise OverflowError.
n = struct.unpack("L", struct.pack("l", n))[0]
- s = ""
+ s = b""
for i in range(digits - 1):
- s = chr(n & 0o377) + s
+ s.insert(0, n & 0o377)
n >>= 8
- s = chr(0o200) + s
+ s.insert(0, 0o200)
return s
-def uts(s, encoding, errors):
- """Convert a unicode object to a string.
- """
- if errors == "utf-8":
- # An extra error handler similar to the -o invalid=UTF-8 option
- # in POSIX.1-2001. Replace untranslatable characters with their
- # UTF-8 representation.
- try:
- return s.encode(encoding, "strict")
- except UnicodeEncodeError:
- x = []
- for c in s:
- try:
- x.append(c.encode(encoding, "strict"))
- except UnicodeEncodeError:
- x.append(c.encode("utf8"))
- return "".join(x)
- else:
- return s.encode(encoding, errors)
-
def calc_chksums(buf):
"""Calculate the checksum for a member's header by summing up all
characters except for the chksum field which is treated as if
self.comptype = comptype
self.fileobj = fileobj
self.bufsize = bufsize
- self.buf = ""
+ self.buf = b""
self.pos = 0
self.closed = False
except ImportError:
raise CompressionError("bz2 module is not available")
if mode == "r":
- self.dbuf = ""
+ self.dbuf = b""
self.cmp = bz2.BZ2Decompressor()
else:
self.cmp = bz2.BZ2Compressor()
self.zlib.DEF_MEM_LEVEL,
0)
timestamp = struct.pack("<L", int(time.time()))
- self.__write("\037\213\010\010%s\002\377" % timestamp)
+ self.__write(b"\037\213\010\010" + timestamp + b"\002\377")
if self.name.endswith(".gz"):
self.name = self.name[:-3]
- self.__write(self.name + NUL)
+ # RFC1952 says we must use ISO-8859-1 for the FNAME field.
+ self.__write(self.name.encode("iso-8859-1", "replace") + NUL)
def write(self, s):
"""Write string s to the stream.
if self.mode == "w" and self.buf:
self.fileobj.write(self.buf)
- self.buf = ""
+ self.buf = b""
if self.comptype == "gz":
# The native zlib crc is an unsigned 32-bit integer, but
# the Python wrapper implicitly casts that to a signed C
"""Initialize for reading a gzip compressed fileobj.
"""
self.cmp = self.zlib.decompressobj(-self.zlib.MAX_WBITS)
- self.dbuf = ""
+ self.dbuf = b""
# taken from gzip.GzipFile with some alterations
- if self.__read(2) != "\037\213":
+ if self.__read(2) != b"\037\213":
raise ReadError("not a gzip file")
- if self.__read(1) != "\010":
+ if self.__read(1) != b"\010":
raise CompressionError("unsupported compression method")
flag = ord(self.__read(1))
return self.__read(size)
c = len(self.dbuf)
- t = [self.dbuf]
while c < size:
buf = self.__read(self.bufsize)
if not buf:
buf = self.cmp.decompress(buf)
except IOError:
raise ReadError("invalid compressed data")
- t.append(buf)
+ self.dbuf += buf
c += len(buf)
- t = "".join(t)
- self.dbuf = t[size:]
- return t[:size]
+ buf = self.dbuf[:size]
+ self.dbuf = self.dbuf[size:]
+ return buf
def __read(self, size):
"""Return size bytes from stream. If internal buffer is empty,
read another block from the stream.
"""
c = len(self.buf)
- t = [self.buf]
while c < size:
buf = self.fileobj.read(self.bufsize)
if not buf:
break
- t.append(buf)
+ self.buf += buf
c += len(buf)
- t = "".join(t)
- self.buf = t[size:]
- return t[:size]
+ buf = self.buf[:size]
+ self.buf = self.buf[size:]
+ return buf
# class _Stream
class _StreamProxy(object):
return self.buf
def getcomptype(self):
- if self.buf.startswith("\037\213\010"):
+ if self.buf.startswith(b"\037\213\010"):
return "gz"
if self.buf.startswith("BZh91"):
return "bz2"
if self.mode == "r":
self.bz2obj = bz2.BZ2Decompressor()
self.fileobj.seek(0)
- self.buf = ""
+ self.buf = b""
else:
self.bz2obj = bz2.BZ2Compressor()
def read(self, size):
- b = [self.buf]
x = len(self.buf)
while x < size:
try:
raw = self.fileobj.read(self.blocksize)
data = self.bz2obj.decompress(raw)
- b.append(data)
+ self.buf += data
except EOFError:
break
x += len(data)
- self.buf = "".join(b)
buf = self.buf[:size]
self.buf = self.buf[size:]
self.sparse = sparse
self.position = 0
+ def seekable(self):
+ if not hasattr(self.fileobj, "seekable"):
+ # XXX gzip.GzipFile and bz2.BZ2File
+ return True
+ return self.fileobj.seekable()
+
def tell(self):
"""Return the current file position.
"""
def readsparse(self, size):
"""Read operation for sparse files.
"""
- data = []
+ data = b""
while size > 0:
buf = self.readsparsesection(size)
if not buf:
break
size -= len(buf)
- data.append(buf)
- return "".join(data)
+ data += buf
+ return data
def readsparsesection(self, size):
"""Read a single section of a sparse file.
section = self.sparse.find(self.position)
if section is None:
- return ""
+ return b""
size = min(size, section.offset + section.size - self.position)
self.size = tarinfo.size
self.position = 0
- self.buffer = ""
+ self.buffer = b""
+
+ def readable(self):
+ return True
+
+ def writable(self):
+ return False
+
+ def seekable(self):
+ return self.fileobj.seekable()
def read(self, size=None):
"""Read at most size bytes from the file. If size is not
if self.closed:
raise ValueError("I/O operation on closed file")
- buf = ""
+ buf = b""
if self.buffer:
if size is None:
buf = self.buffer
- self.buffer = ""
+ self.buffer = b""
else:
buf = self.buffer[:size]
self.buffer = self.buffer[size:]
self.position += len(buf)
return buf
+ # XXX TextIOWrapper uses the read1() method.
+ read1 = read
+
def readline(self, size=-1):
"""Read one entire line from the file. If size is present
and non-negative, return a string with at most that
if self.closed:
raise ValueError("I/O operation on closed file")
- if "\n" in self.buffer:
- pos = self.buffer.find("\n") + 1
- else:
- buffers = [self.buffer]
+ pos = self.buffer.find(b"\n") + 1
+ if pos == 0:
+ # no newline found.
while True:
buf = self.fileobj.read(self.blocksize)
- buffers.append(buf)
- if not buf or "\n" in buf:
- self.buffer = "".join(buffers)
- pos = self.buffer.find("\n") + 1
+ self.buffer += buf
+ if not buf or b"\n" in buf:
+ pos = self.buffer.find(b"\n") + 1
if pos == 0:
# no newline found.
pos = len(self.buffer)
else:
raise ValueError("Invalid argument")
- self.buffer = ""
+ self.buffer = b""
self.fileobj.seek(self.position)
def close(self):
def __repr__(self):
return "<%s %r at %#x>" % (self.__class__.__name__,self.name,id(self))
- def get_info(self, encoding, errors):
+ def get_info(self):
"""Return the TarInfo's attributes as a dictionary.
"""
info = {
if info["type"] == DIRTYPE and not info["name"].endswith("/"):
info["name"] += "/"
- for key in ("name", "linkname", "uname", "gname"):
- if isinstance(info[key], str):
- info[key] = info[key].encode(encoding, errors)
-
return info
def tobuf(self, format=DEFAULT_FORMAT, encoding=ENCODING, errors="strict"):
"""Return a tar header as a string of 512 byte blocks.
"""
- info = self.get_info(encoding, errors)
+ info = self.get_info()
if format == USTAR_FORMAT:
- return self.create_ustar_header(info)
+ return self.create_ustar_header(info, encoding, errors)
elif format == GNU_FORMAT:
- return self.create_gnu_header(info)
+ return self.create_gnu_header(info, encoding, errors)
elif format == PAX_FORMAT:
return self.create_pax_header(info, encoding, errors)
else:
raise ValueError("invalid format")
- def create_ustar_header(self, info):
+ def create_ustar_header(self, info, encoding, errors):
"""Return the object as a ustar header block.
"""
info["magic"] = POSIX_MAGIC
if len(info["name"]) > LENGTH_NAME:
info["prefix"], info["name"] = self._posix_split_name(info["name"])
- return self._create_header(info, USTAR_FORMAT)
+ return self._create_header(info, USTAR_FORMAT, encoding, errors)
- def create_gnu_header(self, info):
+ def create_gnu_header(self, info, encoding, errors):
"""Return the object as a GNU header block sequence.
"""
info["magic"] = GNU_MAGIC
- buf = ""
+ buf = b""
if len(info["linkname"]) > LENGTH_LINK:
- buf += self._create_gnu_long_header(info["linkname"], GNUTYPE_LONGLINK)
+ buf += self._create_gnu_long_header(info["linkname"], GNUTYPE_LONGLINK, encoding, errors)
if len(info["name"]) > LENGTH_NAME:
- buf += self._create_gnu_long_header(info["name"], GNUTYPE_LONGNAME)
+ buf += self._create_gnu_long_header(info["name"], GNUTYPE_LONGNAME, encoding, errors)
- return buf + self._create_header(info, GNU_FORMAT)
+ return buf + self._create_header(info, GNU_FORMAT, encoding, errors)
def create_pax_header(self, info, encoding, errors):
"""Return the object as a ustar header block. If it cannot be
# The pax header has priority.
continue
- val = info[name].decode(encoding, errors)
-
# Try to encode the string as ASCII.
try:
- val.encode("ascii")
+ info[name].encode("ascii", "strict")
except UnicodeEncodeError:
- pax_headers[hname] = val
+ pax_headers[hname] = info[name]
continue
if len(info[name]) > length:
- pax_headers[hname] = val
+ pax_headers[hname] = info[name]
# Test number fields for values that exceed the field limit or values
# that like to be stored as float.
# Create a pax extended header if necessary.
if pax_headers:
- buf = self._create_pax_generic_header(pax_headers)
+ buf = self._create_pax_generic_header(pax_headers, XHDTYPE, encoding, errors)
else:
- buf = ""
+ buf = b""
- return buf + self._create_header(info, USTAR_FORMAT)
+ return buf + self._create_header(info, USTAR_FORMAT, encoding, errors)
@classmethod
- def create_pax_global_header(cls, pax_headers):
+ def create_pax_global_header(cls, pax_headers, encoding, errors):
"""Return the object as a pax global header block sequence.
"""
- return cls._create_pax_generic_header(pax_headers, type=XGLTYPE)
+ return cls._create_pax_generic_header(pax_headers, XGLTYPE, encoding, errors)
def _posix_split_name(self, name):
"""Split a name longer than 100 chars into a prefix
return prefix, name
@staticmethod
- def _create_header(info, format):
+ def _create_header(info, format, encoding, errors):
"""Return a header block. info is a dictionary with file
information, format must be one of the *_FORMAT constants.
"""
parts = [
- stn(info.get("name", ""), 100),
+ stn(info.get("name", ""), 100, encoding, errors),
itn(info.get("mode", 0) & 0o7777, 8, format),
itn(info.get("uid", 0), 8, format),
itn(info.get("gid", 0), 8, format),
itn(info.get("size", 0), 12, format),
itn(info.get("mtime", 0), 12, format),
- " ", # checksum field
+ b" ", # checksum field
info.get("type", REGTYPE),
- stn(info.get("linkname", ""), 100),
- stn(info.get("magic", POSIX_MAGIC), 8),
- stn(info.get("uname", "root"), 32),
- stn(info.get("gname", "root"), 32),
+ stn(info.get("linkname", ""), 100, encoding, errors),
+ info.get("magic", POSIX_MAGIC),
+ stn(info.get("uname", "root"), 32, encoding, errors),
+ stn(info.get("gname", "root"), 32, encoding, errors),
itn(info.get("devmajor", 0), 8, format),
itn(info.get("devminor", 0), 8, format),
- stn(info.get("prefix", ""), 155)
+ stn(info.get("prefix", ""), 155, encoding, errors)
]
- buf = struct.pack("%ds" % BLOCKSIZE, "".join(parts))
+ buf = struct.pack("%ds" % BLOCKSIZE, b"".join(parts))
chksum = calc_chksums(buf[-BLOCKSIZE:])[0]
- buf = buf[:-364] + "%06o\0" % chksum + buf[-357:]
+ buf = buf[:-364] + bytes("%06o\0" % chksum) + buf[-357:]
return buf
@staticmethod
return payload
@classmethod
- def _create_gnu_long_header(cls, name, type):
+ def _create_gnu_long_header(cls, name, type, encoding, errors):
"""Return a GNUTYPE_LONGNAME or GNUTYPE_LONGLINK sequence
for name.
"""
- name += NUL
+ name = name.encode(encoding, errors) + NUL
info = {}
info["name"] = "././@LongLink"
info["magic"] = GNU_MAGIC
# create extended header + name blocks.
- return cls._create_header(info, USTAR_FORMAT) + \
+ return cls._create_header(info, USTAR_FORMAT, encoding, errors) + \
cls._create_payload(name)
@classmethod
- def _create_pax_generic_header(cls, pax_headers, type=XHDTYPE):
+ def _create_pax_generic_header(cls, pax_headers, type, encoding, errors):
"""Return a POSIX.1-2001 extended or global header sequence
that contains a list of keyword, value pairs. The values
- must be unicode objects.
+ must be strings.
"""
- records = []
+ records = b""
for keyword, value in pax_headers.items():
keyword = keyword.encode("utf8")
value = value.encode("utf8")
if n == p:
break
p = n
- records.append("%d %s=%s\n" % (p, keyword, value))
- records = "".join(records)
+ records += bytes(str(p)) + b" " + keyword + b"=" + value + b"\n"
# We use a hardcoded "././@PaxHeader" name like star does
# instead of the one that POSIX recommends.
info["magic"] = POSIX_MAGIC
# Create pax header + record blocks.
- return cls._create_header(info, USTAR_FORMAT) + \
+ return cls._create_header(info, USTAR_FORMAT, encoding, errors) + \
cls._create_payload(records)
@classmethod
- def frombuf(cls, buf):
- """Construct a TarInfo object from a 512 byte string buffer.
+ def frombuf(cls, buf, encoding, errors):
+ """Construct a TarInfo object from a 512 byte bytes object.
"""
if len(buf) != BLOCKSIZE:
raise HeaderError("truncated header")
obj = cls()
obj.buf = buf
- obj.name = nts(buf[0:100])
+ obj.name = nts(buf[0:100], encoding, errors)
obj.mode = nti(buf[100:108])
obj.uid = nti(buf[108:116])
obj.gid = nti(buf[116:124])
obj.mtime = nti(buf[136:148])
obj.chksum = chksum
obj.type = buf[156:157]
- obj.linkname = nts(buf[157:257])
- obj.uname = nts(buf[265:297])
- obj.gname = nts(buf[297:329])
+ obj.linkname = nts(buf[157:257], encoding, errors)
+ obj.uname = nts(buf[265:297], encoding, errors)
+ obj.gname = nts(buf[297:329], encoding, errors)
obj.devmajor = nti(buf[329:337])
obj.devminor = nti(buf[337:345])
- prefix = nts(buf[345:500])
+ prefix = nts(buf[345:500], encoding, errors)
# Old V7 tar format represents a directory as a regular
# file with a trailing slash.
buf = tarfile.fileobj.read(BLOCKSIZE)
if not buf:
return
- obj = cls.frombuf(buf)
+ obj = cls.frombuf(buf, tarfile.encoding, tarfile.errors)
obj.offset = tarfile.fileobj.tell() - BLOCKSIZE
return obj._proc_member(tarfile)
# the longname information.
next.offset = self.offset
if self.type == GNUTYPE_LONGNAME:
- next.name = nts(buf)
+ next.name = nts(buf, tarfile.encoding, tarfile.errors)
elif self.type == GNUTYPE_LONGLINK:
- next.linkname = nts(buf)
+ next.linkname = nts(buf, tarfile.encoding, tarfile.errors)
return next
lastpos = offset + numbytes
pos += 24
- isextended = ord(buf[482])
+ isextended = bool(buf[482])
origsize = nti(buf[483:495])
# If the isextended flag is given,
# there are extra headers to process.
- while isextended == 1:
+ while isextended:
buf = tarfile.fileobj.read(BLOCKSIZE)
pos = 0
for i in range(21):
realpos += numbytes
lastpos = offset + numbytes
pos += 24
- isextended = ord(buf[504])
+ isextended = bool(buf[504])
if lastpos < origsize:
sp.append(_hole(lastpos, origsize - lastpos))
value = PAX_NUMBER_FIELDS[keyword](value)
except ValueError:
value = 0
- else:
- value = uts(value, encoding, errors)
setattr(self, keyword, value)
if errors is not None:
self.errors = errors
elif mode == "r":
- self.errors = "utf-8"
+ self.errors = "replace"
else:
self.errors = "strict"
self.closed = False
self.members = [] # list of members as TarInfo objects
self._loaded = False # flag if all members have been read
- self.offset = 0 # current position in the archive file
+ self.offset = 0 # current position in the archive file
self.inodes = {} # dictionary caching the inodes of
# archive members already added
while True:
if self.next() is None:
if self.offset > 0:
- self.fileobj.seek(- BLOCKSIZE, 1)
+ self.fileobj.seek(self.fileobj.tell() - BLOCKSIZE)
break
if self.mode in "aw":
self._loaded = True
if self.pax_headers:
- buf = self.tarinfo.create_pax_global_header(self.pax_headers.copy())
+ buf = self.tarinfo.create_pax_global_header(
+ self.pax_headers.copy(), self.encoding, self.errors)
self.fileobj.write(buf)
self.offset += len(buf)
import sys
import os
+import io
import shutil
import tempfile
import StringIO
def test_fileobj_readlines(self):
self.tar.extract("ustar/regtype", TEMPDIR)
tarinfo = self.tar.getmember("ustar/regtype")
- fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "rU")
- fobj2 = self.tar.extractfile(tarinfo)
+ fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "r")
+ fobj2 = io.TextIOWrapper(self.tar.extractfile(tarinfo))
lines1 = fobj1.readlines()
lines2 = fobj2.readlines()
fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "rU")
fobj2 = self.tar.extractfile(tarinfo)
lines1 = fobj1.readlines()
- lines2 = [line for line in fobj2]
+ lines2 = list(io.TextIOWrapper(fobj2))
self.assert_(lines1 == lines2,
"fileobj.__iter__() failed")
fobj.seek(0, 2)
self.assertEqual(tarinfo.size, fobj.tell(),
"seek() to file's end failed")
- self.assert_(fobj.read() == "",
+ self.assert_(fobj.read() == b"",
"read() at file's end did not return empty string")
fobj.seek(-tarinfo.size, 2)
self.assertEqual(0, fobj.tell(),
- "relative seek() to file's start failed")
+ "relative seek() to file's end failed")
fobj.seek(512)
s1 = fobj.readlines()
fobj.seek(512)
def _testfunc_file(self, name, mode):
try:
tarfile.open(name, mode)
- except tarfile.ReadError:
+ except tarfile.ReadError as e:
self.fail()
def _testfunc_fileobj(self, name, mode):
try:
tarfile.open(name, mode, fileobj=open(name, "rb"))
- except tarfile.ReadError:
+ except tarfile.ReadError as e:
self.fail()
def _test_modes(self, testfunc):
tarinfo = self.tar.getmember(longname)
offset = tarinfo.offset
self.tar.fileobj.seek(offset)
- fobj = StringIO.StringIO(self.tar.fileobj.read(3 * 512))
+ fobj = io.BytesIO(self.tar.fileobj.read(3 * 512))
self.assertRaises(tarfile.ReadError, tarfile.open, name="foo.tar", fileobj=fobj)
def test_header_offset(self):
# the preceding extended header.
longname = self.subdir + "/" + "123/" * 125 + "longname"
offset = self.tar.getmember(longname).offset
- fobj = open(tarname)
+ fobj = open(tarname, "rb")
fobj.seek(offset)
- tarinfo = tarfile.TarInfo.frombuf(fobj.read(512))
+ tarinfo = tarfile.TarInfo.frombuf(fobj.read(512), "iso8859-1", "strict")
self.assertEqual(tarinfo.type, self.longnametype)
self.assertEqual(tar.pax_headers, pax_headers)
self.assertEqual(tar.getmembers()[0].pax_headers, pax_headers)
- # Test if all the fields are unicode.
+ # Test if all the fields are strings.
for key, val in tar.pax_headers.items():
- self.assert_(type(key) is unicode)
- self.assert_(type(val) is unicode)
+ self.assert_(type(key) is not bytes)
+ self.assert_(type(val) is not bytes)
if key in tarfile.PAX_NUMBER_FIELDS:
try:
tarfile.PAX_NUMBER_FIELDS[key](val)
tar.close()
tar = tarfile.open(tmpname, encoding=encoding)
- self.assert_(type(tar.getnames()[0]) is not unicode)
- self.assertEqual(tar.getmembers()[0].name, name.encode(encoding))
+ self.assert_(type(tar.getnames()[0]) is not bytes)
+ self.assertEqual(tar.getmembers()[0].name, name)
tar.close()
def test_unicode_filename_error(self):
tar = tarfile.open(tmpname, "w", format=self.format, encoding="ascii", errors="strict")
tarinfo = tarfile.TarInfo()
- tarinfo.name = "äöü"
- if self.format == tarfile.PAX_FORMAT:
- self.assertRaises(UnicodeError, tar.addfile, tarinfo)
- else:
- tar.addfile(tarinfo)
-
tarinfo.name = "äöü"
self.assertRaises(UnicodeError, tar.addfile, tarinfo)
t.uname = name
t.gname = name
- fobj = StringIO.StringIO()
+ fobj = io.BytesIO()
tar = tarfile.open("foo.tar", mode="w", fileobj=fobj, format=self.format, encoding="iso8859-1")
tar.addfile(t)
tar.close()
self.assertEqual(t.uname, "äöü")
self.assertEqual(t.gname, "äöü")
+
class GNUUnicodeTest(UstarUnicodeTest):
format = tarfile.GNU_FORMAT
-class PaxUnicodeTest(UstarUnicodeTest):
-
- format = tarfile.PAX_FORMAT
-
- def _create_unicode_name(self, name):
- tar = tarfile.open(tmpname, "w", format=self.format)
- t = tarfile.TarInfo()
- t.pax_headers["path"] = name
- tar.addfile(t)
- tar.close()
-
- def test_error_handlers(self):
- # Test if the unicode error handlers work correctly for characters
- # that cannot be expressed in a given encoding.
- self._create_unicode_name("äöü")
-
- for handler, name in (("utf-8", "äöü".encode("utf8")),
- ("replace", "???"), ("ignore", "")):
- tar = tarfile.open(tmpname, format=self.format, encoding="ascii",
- errors=handler)
- self.assertEqual(tar.getnames()[0], name)
-
- self.assertRaises(UnicodeError, tarfile.open, tmpname,
- encoding="ascii", errors="strict")
-
- def test_error_handler_utf8(self):
- # Create a pathname that has one component representable using
- # iso8859-1 and the other only in iso8859-15.
- self._create_unicode_name("äöü/¤")
-
- tar = tarfile.open(tmpname, format=self.format, encoding="iso8859-1",
- errors="utf-8")
- self.assertEqual(tar.getnames()[0], "äöü/" + "¤".encode("utf8"))
-
-
class AppendTest(unittest.TestCase):
# Test append mode (cp. patch #1652681).
tarinfo.tobuf(tarfile.PAX_FORMAT)
+class MiscTest(unittest.TestCase):
+
+ def test_char_fields(self):
+ self.assertEqual(tarfile.stn("foo", 8, "ascii", "strict"), b"foo\0\0\0\0\0")
+ self.assertEqual(tarfile.stn("foobar", 3, "ascii", "strict"), b"foo")
+ self.assertEqual(tarfile.nts(b"foo\0\0\0\0\0", "ascii", "strict"), "foo")
+ self.assertEqual(tarfile.nts(b"foo\0bar\0", "ascii", "strict"), "foo")
+
+ def test_number_fields(self):
+ self.assertEqual(tarfile.itn(1), b"0000001\x00")
+ self.assertEqual(tarfile.itn(0xffffffff), b"\x80\x00\x00\x00\xff\xff\xff\xff")
+
+
class GzipMiscReadTest(MiscReadTest):
tarname = gzipname
mode = "r:gz"
PaxWriteTest,
UstarUnicodeTest,
GNUUnicodeTest,
- PaxUnicodeTest,
AppendTest,
LimitsTest,
+ MiscTest,
]
if hasattr(os, "link"):
projects / python / commitdiff