diff options
| -rw-r--r-- | Doc/lib/libtarfile.tex | 157 | ||||
| -rw-r--r-- | Lib/tarfile.py | 980 | ||||
| -rw-r--r-- | Lib/test/test_tarfile.py | 1342 | ||||
| -rw-r--r-- | Lib/test/testtar.tar | bin | 133120 -> 256000 bytes | |||
| -rw-r--r-- | Misc/NEWS | 3 |
5 files changed, 1514 insertions, 968 deletions
diff --git a/Doc/lib/libtarfile.tex b/Doc/lib/libtarfile.tex index 96f835f..73c35ed 100644 --- a/Doc/lib/libtarfile.tex +++ b/Doc/lib/libtarfile.tex @@ -12,21 +12,24 @@ Some facts and figures: \begin{itemize} \item reads and writes \module{gzip} and \module{bzip2} compressed archives. -\item creates \POSIX{} 1003.1-1990 compliant or GNU tar compatible archives. -\item reads GNU tar extensions \emph{longname}, \emph{longlink} and - \emph{sparse}. -\item stores pathnames of unlimited length using GNU tar extensions. +\item read/write support for the \POSIX{}.1-1988 (ustar) format. +\item read/write support for the GNU tar format including \emph{longname} and + \emph{longlink} extensions, read-only support for the \emph{sparse} + extension. +\item read/write support for the \POSIX{}.1-2001 (pax) format. + \versionadded{2.6} \item handles directories, regular files, hardlinks, symbolic links, fifos, character devices and block devices and is able to acquire and restore file information like timestamp, access permissions and owner. \item can handle tape devices. \end{itemize} -\begin{funcdesc}{open}{\optional{name\optional{, mode - \optional{, fileobj\optional{, bufsize}}}}} +\begin{funcdesc}{open}{name\optional{, mode\optional{, + fileobj\optional{, bufsize}}}, **kwargs} Return a \class{TarFile} object for the pathname \var{name}. - For detailed information on \class{TarFile} objects, - see \citetitle{TarFile Objects} (section \ref{tarfile-objects}). + For detailed information on \class{TarFile} objects and the keyword + arguments that are allowed, see \citetitle{TarFile Objects} + (section \ref{tarfile-objects}). \var{mode} has to be a string of the form \code{'filemode[:compression]'}, it defaults to \code{'r'}. Here is a full list of mode combinations: @@ -130,6 +133,31 @@ Some facts and figures: \versionadded{2.6} \end{excdesc} +\begin{datadesc}{USTAR_FORMAT} + \POSIX{}.1-1988 (ustar) format. It supports filenames up to a length of + at best 256 characters and linknames up to 100 characters. The maximum + file size is 8 gigabytes. This is an old and limited but widely + supported format. +\end{datadesc} + +\begin{datadesc}{GNU_FORMAT} + GNU tar format. It supports arbitrarily long filenames and linknames and + files bigger than 8 gigabytes. It is the defacto standard on GNU/Linux + systems. +\end{datadesc} + +\begin{datadesc}{PAX_FORMAT} + \POSIX{}.1-2001 (pax) format. It is the most flexible format with + virtually no limits. It supports long filenames and linknames, large files + and stores pathnames in a portable way. However, not all tar + implementations today are able to handle pax archives properly. +\end{datadesc} + +\begin{datadesc}{DEFAULT_FORMAT} + The default format for creating archives. This is currently + \constant{GNU_FORMAT}. +\end{datadesc} + \begin{seealso} \seemodule{zipfile}{Documentation of the \refmodule{zipfile} standard module.} @@ -152,12 +180,21 @@ tar archive several times. Each archive member is represented by a \class{TarInfo} object, see \citetitle{TarInfo Objects} (section \ref{tarinfo-objects}) for details. -\begin{classdesc}{TarFile}{\optional{name - \optional{, mode\optional{, fileobj}}}} - Open an \emph{(uncompressed)} tar archive \var{name}. +\begin{classdesc}{TarFile}{name=None, mode='r', fileobj=None, + format=DEFAULT_FORMAT, tarinfo=TarInfo, dereference=False, + ignore_zeros=False, encoding=None, pax_headers=None, debug=0, + errorlevel=0} + + All following arguments are optional and can be accessed as instance + attributes as well. + + \var{name} is the pathname of the archive. It can be omitted if + \var{fileobj} is given. In this case, the file object's \member{name} + attribute is used if it exists. + \var{mode} is either \code{'r'} to read from an existing archive, \code{'a'} to append data to an existing file or \code{'w'} to create a new - file overwriting an existing one. \var{mode} defaults to \code{'r'}. + file overwriting an existing one. If \var{fileobj} is given, it is used for reading or writing data. If it can be determined, \var{mode} is overridden by \var{fileobj}'s mode. @@ -165,6 +202,48 @@ tar archive several times. Each archive member is represented by a \begin{notice} \var{fileobj} is not closed, when \class{TarFile} is closed. \end{notice} + + \var{format} controls the archive format. It must be one of the constants + \constant{USTAR_FORMAT}, \constant{GNU_FORMAT} or \constant{PAX_FORMAT} + that are defined at module level. + \versionadded{2.6} + + The \var{tarinfo} argument can be used to replace the default + \class{TarInfo} class with a different one. + \versionadded{2.6} + + If \var{dereference} is \code{False}, add symbolic and hard links to the + archive. If it is \code{True}, add the content of the target files to the + archive. This has no effect on systems that do not support symbolic links. + + If \var{ignore_zeros} is \code{False}, treat an empty block as the end of + the archive. If it is \var{True}, skip empty (and invalid) blocks and try + to get as many members as possible. This is only useful for reading + concatenated or damaged archives. + + \var{debug} can be set from \code{0} (no debug messages) up to \code{3} + (all debug messages). The messages are written to \code{sys.stderr}. + + If \var{errorlevel} is \code{0}, all errors are ignored when using + \method{extract()}. Nevertheless, they appear as error messages in the + debug output, when debugging is enabled. If \code{1}, all \emph{fatal} + errors are raised as \exception{OSError} or \exception{IOError} exceptions. + If \code{2}, all \emph{non-fatal} errors are raised as \exception{TarError} + exceptions as well. + + The \var{encoding} argument defines the local character encoding. It + defaults to the value from \function{sys.getfilesystemencoding()} or if + that is \code{None} to \code{"ascii"}. \var{encoding} is used only in + connection with the pax format which stores text data in \emph{UTF-8}. If + it is not set correctly, character conversion will fail with a + \exception{UnicodeError}. + \versionadded{2.6} + + The \var{pax_headers} argument must be a dictionary whose elements are + either unicode objects, numbers or strings that can be decoded to unicode + using \var{encoding}. This information will be added to the archive as a + pax global header. + \versionadded{2.6} \end{classdesc} \begin{methoddesc}{open}{...} @@ -279,43 +358,11 @@ tar archive several times. Each archive member is represented by a \end{methoddesc} \begin{memberdesc}{posix} - If true, create a \POSIX{} 1003.1-1990 compliant archive. GNU - extensions are not used, because they are not part of the \POSIX{} - standard. This limits the length of filenames to at most 256, - link names to 100 characters and the maximum file size to 8 - gigabytes. A \exception{ValueError} is raised if a file exceeds - this limit. If false, create a GNU tar compatible archive. It - will not be \POSIX{} compliant, but can store files without any - of the above restrictions. + Setting this to \constant{True} is equivalent to setting the + \member{format} attribute to \constant{USTAR_FORMAT}, + \constant{False} is equivalent to \constant{GNU_FORMAT}. \versionchanged[\var{posix} defaults to \constant{False}]{2.4} -\end{memberdesc} - -\begin{memberdesc}{dereference} - If false, add symbolic and hard links to archive. If true, add the - content of the target files to the archive. This has no effect on - systems that do not support symbolic links. -\end{memberdesc} - -\begin{memberdesc}{ignore_zeros} - If false, treat an empty block as the end of the archive. If true, - skip empty (and invalid) blocks and try to get as many members as - possible. This is only useful for concatenated or damaged - archives. -\end{memberdesc} - -\begin{memberdesc}{debug=0} - To be set from \code{0} (no debug messages; the default) up to - \code{3} (all debug messages). The messages are written to - \code{sys.stderr}. -\end{memberdesc} - -\begin{memberdesc}{errorlevel} - If \code{0} (the default), all errors are ignored when using - \method{extract()}. Nevertheless, they appear as error messages - in the debug output, when debugging is enabled. If \code{1}, all - \emph{fatal} errors are raised as \exception{OSError} or - \exception{IOError} exceptions. If \code{2}, all \emph{non-fatal} - errors are raised as \exception{TarError} exceptions as well. + \deprecated{2.6}{Use the \member{format} attribute instead.} \end{memberdesc} %----------------- @@ -343,12 +390,16 @@ the file's data itself. invalid.]{2.6} \end{methoddesc} -\begin{methoddesc}{tobuf}{posix} - Create a string buffer from a \class{TarInfo} object. - See \class{TarFile}'s \member{posix} attribute for information - on the \var{posix} argument. It defaults to \constant{False}. +\begin{methoddesc}{fromtarfile}{tarfile} + Read the next member from the \class{TarFile} object \var{tarfile} and + return it as a \class{TarInfo} object. + \versionadded{2.6} +\end{methoddesc} - \versionadded[The \var{posix} parameter]{2.5} +\begin{methoddesc}{tobuf}{\optional{format}} + Create a string buffer from a \class{TarInfo} object. See + \class{TarFile}'s \member{format} argument for information. + \versionchanged[The \var{format} parameter]{2.6} \end{methoddesc} A \code{TarInfo} object has the following public data attributes: diff --git a/Lib/tarfile.py b/Lib/tarfile.py index 54bb1b8..b6dc3ee 100644 --- a/Lib/tarfile.py +++ b/Lib/tarfile.py @@ -33,7 +33,7 @@ __version__ = "$Revision$" # $Source$ -version = "0.8.0" +version = "0.9.0" __author__ = "Lars Gustäbel (lars@gustaebel.de)" __date__ = "$Date$" __cvsid__ = "$Id$" @@ -50,6 +50,7 @@ import errno import time import struct import copy +import re if sys.platform == 'mac': # This module needs work for MacOS9, especially in the area of pathname @@ -69,42 +70,60 @@ __all__ = ["TarFile", "TarInfo", "is_tarfile", "TarError"] #--------------------------------------------------------- # tar constants #--------------------------------------------------------- -NUL = "\0" # the null character -BLOCKSIZE = 512 # length of processing blocks +NUL = "\0" # the null character +BLOCKSIZE = 512 # length of processing blocks RECORDSIZE = BLOCKSIZE * 20 # length of records -MAGIC = "ustar" # magic tar string -VERSION = "00" # version number +GNU_MAGIC = "ustar \0" # magic gnu tar string +POSIX_MAGIC = "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 -MAXSIZE_MEMBER = 077777777777L # maximum size of a file (11 octal digits) +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 +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 +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 -GNUTYPE_LONGNAME = "L" # GNU tar extension for longnames -GNUTYPE_LONGLINK = "K" # GNU tar extension for longlink -GNUTYPE_SPARSE = "S" # GNU tar extension for sparse file +GNUTYPE_LONGNAME = "L" # GNU tar longname +GNUTYPE_LONGLINK = "K" # GNU tar longlink +GNUTYPE_SPARSE = "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 + +USTAR_FORMAT = 0 # POSIX.1-1988 (ustar) format +GNU_FORMAT = 1 # GNU tar format +PAX_FORMAT = 2 # POSIX.1-2001 (pax) format +DEFAULT_FORMAT = GNU_FORMAT #--------------------------------------------------------- # tarfile constants #--------------------------------------------------------- -SUPPORTED_TYPES = (REGTYPE, AREGTYPE, LNKTYPE, # file types that tarfile - SYMTYPE, DIRTYPE, FIFOTYPE, # can cope with. +# File types that tarfile supports: +SUPPORTED_TYPES = (REGTYPE, AREGTYPE, LNKTYPE, + SYMTYPE, DIRTYPE, FIFOTYPE, CONTTYPE, CHRTYPE, BLKTYPE, GNUTYPE_LONGNAME, GNUTYPE_LONGLINK, GNUTYPE_SPARSE) -REGULAR_TYPES = (REGTYPE, AREGTYPE, # file types that somehow - CONTTYPE, GNUTYPE_SPARSE) # represent regular files +# File types that will be treated as a regular file. +REGULAR_TYPES = (REGTYPE, AREGTYPE, + CONTTYPE, GNUTYPE_SPARSE) + +# File types that are part of the GNU tar format. +GNU_TYPES = (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK, + GNUTYPE_SPARSE) + +# Fields from a pax header that override a TarInfo attribute. +PAX_FIELDS = ("path", "linkpath", "size", "mtime", + "uid", "gid", "uname", "gname") #--------------------------------------------------------- # Bits used in the mode field, values in octal. @@ -131,6 +150,13 @@ TOWRITE = 0002 # write by other TOEXEC = 0001 # execute/search by other #--------------------------------------------------------- +# initialization +#--------------------------------------------------------- +ENCODING = sys.getfilesystemencoding() +if ENCODING is None: + ENCODING = "ascii" + +#--------------------------------------------------------- # Some useful functions #--------------------------------------------------------- @@ -139,6 +165,15 @@ def stn(s, length): """ return s[:length] + (length - len(s)) * NUL +def nts(s): + """Convert a null-terminated string field to a python string. + """ + # Use the string up to the first null char. + p = s.find("\0") + if p == -1: + return s + return s[:p] + def nti(s): """Convert a number field to a python number. """ @@ -146,7 +181,7 @@ def nti(s): # itn() below. if s[0] != chr(0200): try: - n = int(s.rstrip(NUL + " ") or "0", 8) + n = int(nts(s) or "0", 8) except ValueError: raise HeaderError("invalid header") else: @@ -156,7 +191,7 @@ def nti(s): n += ord(s[i + 1]) return n -def itn(n, digits=8, posix=False): +def itn(n, digits=8, format=DEFAULT_FORMAT): """Convert a python number to a number field. """ # POSIX 1003.1-1988 requires numbers to be encoded as a string of @@ -168,7 +203,7 @@ def itn(n, digits=8, posix=False): if 0 <= n < 8 ** (digits - 1): s = "%0*o" % (digits - 1, n) + NUL else: - if posix: + if format != GNU_FORMAT or n >= 256 ** (digits - 1): raise ValueError("overflow in number field") if n < 0: @@ -514,7 +549,10 @@ class _Stream: buf = self.__read(self.bufsize) if not buf: break - buf = self.cmp.decompress(buf) + try: + buf = self.cmp.decompress(buf) + except IOError: + raise ReadError("invalid compressed data") t.append(buf) c += len(buf) t = "".join(t) @@ -575,6 +613,7 @@ class _BZ2Proxy(object): def __init__(self, fileobj, mode): self.fileobj = fileobj self.mode = mode + self.name = getattr(self.fileobj, "name", None) self.init() def init(self): @@ -847,8 +886,8 @@ class TarInfo(object): """Construct a TarInfo object. name is the optional name of the member. """ - self.name = name # member name (dirnames must end with '/') - self.mode = 0666 # file permissions + self.name = name # member name + self.mode = 0644 # file permissions self.uid = 0 # user id self.gid = 0 # group id self.size = 0 # file size @@ -856,17 +895,274 @@ class TarInfo(object): self.chksum = 0 # header checksum self.type = REGTYPE # member type self.linkname = "" # link name - self.uname = "user" # user name - self.gname = "group" # group name + self.uname = "root" # user name + self.gname = "root" # group name self.devmajor = 0 # device major number self.devminor = 0 # device minor number self.offset = 0 # the tar header starts here self.offset_data = 0 # the file's data starts here + self.pax_headers = {} # pax header information + + # In pax headers the "name" and "linkname" field are called + # "path" and "linkpath". + def _getpath(self): + return self.name + def _setpath(self, name): + self.name = name + path = property(_getpath, _setpath) + + def _getlinkpath(self): + return self.linkname + def _setlinkpath(self, linkname): + self.linkname = linkname + linkpath = property(_getlinkpath, _setlinkpath) + def __repr__(self): return "<%s %r at %#x>" % (self.__class__.__name__,self.name,id(self)) + def get_info(self): + """Return the TarInfo's attributes as a dictionary. + """ + info = { + "name": normpath(self.name), + "mode": self.mode & 07777, + "uid": self.uid, + "gid": self.gid, + "size": self.size, + "mtime": self.mtime, + "chksum": self.chksum, + "type": self.type, + "linkname": normpath(self.linkname) if self.linkname else "", + "uname": self.uname, + "gname": self.gname, + "devmajor": self.devmajor, + "devminor": self.devminor + } + + if info["type"] == DIRTYPE and not info["name"].endswith("/"): + info["name"] += "/" + + return info + + def tobuf(self, format=DEFAULT_FORMAT, encoding=ENCODING): + """Return a tar header as a string of 512 byte blocks. + """ + if format == USTAR_FORMAT: + return self.create_ustar_header() + elif format == GNU_FORMAT: + return self.create_gnu_header() + elif format == PAX_FORMAT: + return self.create_pax_header(encoding) + else: + raise ValueError("invalid format") + + def create_ustar_header(self): + """Return the object as a ustar header block. + """ + info = self.get_info() + info["magic"] = POSIX_MAGIC + + if len(info["linkname"]) > LENGTH_LINK: + raise ValueError("linkname is too long") + + if len(info["name"]) > LENGTH_NAME: + info["prefix"], info["name"] = self._posix_split_name(info["name"]) + + return self._create_header(info, USTAR_FORMAT) + + def create_gnu_header(self): + """Return the object as a GNU header block sequence. + """ + info = self.get_info() + info["magic"] = GNU_MAGIC + + buf = "" + if len(info["linkname"]) > LENGTH_LINK: + buf += self._create_gnu_long_header(info["linkname"], GNUTYPE_LONGLINK) + + if len(info["name"]) > LENGTH_NAME: + buf += self._create_gnu_long_header(info["name"], GNUTYPE_LONGNAME) + + return buf + self._create_header(info, GNU_FORMAT) + + def create_pax_header(self, encoding): + """Return the object as a ustar header block. If it cannot be + represented this way, prepend a pax extended header sequence + with supplement information. + """ + info = self.get_info() + info["magic"] = POSIX_MAGIC + pax_headers = self.pax_headers.copy() + + # Test string fields for values that exceed the field length or cannot + # be represented in ASCII encoding. + for name, hname, length in ( + ("name", "path", LENGTH_NAME), ("linkname", "linkpath", LENGTH_LINK), + ("uname", "uname", 32), ("gname", "gname", 32)): + + val = info[name].decode(encoding) + + # Try to encode the string as ASCII. + try: + val.encode("ascii") + except UnicodeEncodeError: + pax_headers[hname] = val + continue + + if len(val) > length: + if name == "name": + # Try to squeeze a longname in the prefix and name fields as in + # ustar format. + try: + info["prefix"], info["name"] = self._posix_split_name(info["name"]) + except ValueError: + pax_headers[hname] = val + else: + continue + else: + pax_headers[hname] = val + + # Test number fields for values that exceed the field limit or values + # that like to be stored as float. + for name, digits in (("uid", 8), ("gid", 8), ("size", 12), ("mtime", 12)): + val = info[name] + if not 0 <= val < 8 ** (digits - 1) or isinstance(val, float): + pax_headers[name] = unicode(val) + info[name] = 0 + + if pax_headers: + buf = self._create_pax_generic_header(pax_headers) + else: + buf = "" + + return buf + self._create_header(info, USTAR_FORMAT) + + @classmethod + def create_pax_global_header(cls, pax_headers, encoding): + """Return the object as a pax global header block sequence. + """ + new_headers = {} + for key, val in pax_headers.iteritems(): + key = cls._to_unicode(key, encoding) + val = cls._to_unicode(val, encoding) + new_headers[key] = val + return cls._create_pax_generic_header(new_headers, type=XGLTYPE) + + @staticmethod + def _to_unicode(value, encoding): + if isinstance(value, unicode): + return value + elif isinstance(value, (int, long, float)): + return unicode(value) + elif isinstance(value, str): + return unicode(value, encoding) + else: + raise ValueError("unable to convert to unicode: %r" % value) + + def _posix_split_name(self, name): + """Split a name longer than 100 chars into a prefix + and a name part. + """ + prefix = name[:LENGTH_PREFIX + 1] + while prefix and prefix[-1] != "/": + prefix = prefix[:-1] + + name = name[len(prefix):] + prefix = prefix[:-1] + + if not prefix or len(name) > LENGTH_NAME: + raise ValueError("name is too long") + return prefix, name + + @staticmethod + def _create_header(info, format): + """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), + itn(info.get("mode", 0) & 07777, 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 + info.get("type", REGTYPE), + stn(info.get("linkname", ""), 100), + stn(info.get("magic", ""), 8), + stn(info.get("uname", ""), 32), + stn(info.get("gname", ""), 32), + itn(info.get("devmajor", 0), 8, format), + itn(info.get("devminor", 0), 8, format), + stn(info.get("prefix", ""), 155) + ] + + buf = struct.pack("%ds" % BLOCKSIZE, "".join(parts)) + chksum = calc_chksums(buf[-BLOCKSIZE:])[0] + buf = buf[:-364] + "%06o\0" % chksum + buf[-357:] + return buf + + @staticmethod + def _create_payload(payload): + """Return the string payload filled with zero bytes + up to the next 512 byte border. + """ + blocks, remainder = divmod(len(payload), BLOCKSIZE) + if remainder > 0: + payload += (BLOCKSIZE - remainder) * NUL + return payload + + @classmethod + def _create_gnu_long_header(cls, name, type): + """Return a GNUTYPE_LONGNAME or GNUTYPE_LONGLINK sequence + for name. + """ + name += NUL + + info = {} + info["name"] = "././@LongLink" + info["type"] = type + info["size"] = len(name) + info["magic"] = GNU_MAGIC + + # create extended header + name blocks. + return cls._create_header(info, USTAR_FORMAT) + \ + cls._create_payload(name) + + @classmethod + def _create_pax_generic_header(cls, pax_headers, type=XHDTYPE): + """Return a POSIX.1-2001 extended or global header sequence + that contains a list of keyword, value pairs. The values + must be unicode objects. + """ + records = [] + for keyword, value in pax_headers.iteritems(): + keyword = keyword.encode("utf8") + value = value.encode("utf8") + l = len(keyword) + len(value) + 3 # ' ' + '=' + '\n' + n = p = 0 + while True: + n = l + len(str(p)) + if n == p: + break + p = n + records.append("%d %s=%s\n" % (p, keyword, value)) + records = "".join(records) + + # We use a hardcoded "././@PaxHeader" name like star does + # instead of the one that POSIX recommends. + info = {} + info["name"] = "././@PaxHeader" + info["type"] = type + info["size"] = len(records) + info["magic"] = POSIX_MAGIC + + # Create pax header + record blocks. + return cls._create_header(info, USTAR_FORMAT) + \ + cls._create_payload(records) + @classmethod def frombuf(cls, buf): """Construct a TarInfo object from a 512 byte string buffer. @@ -880,125 +1176,251 @@ class TarInfo(object): if chksum not in calc_chksums(buf): raise HeaderError("bad checksum") - tarinfo = cls() - tarinfo.buf = buf - tarinfo.name = buf[0:100].rstrip(NUL) - tarinfo.mode = nti(buf[100:108]) - tarinfo.uid = nti(buf[108:116]) - tarinfo.gid = nti(buf[116:124]) - tarinfo.size = nti(buf[124:136]) - tarinfo.mtime = nti(buf[136:148]) - tarinfo.chksum = chksum - tarinfo.type = buf[156:157] - tarinfo.linkname = buf[157:257].rstrip(NUL) - tarinfo.uname = buf[265:297].rstrip(NUL) - tarinfo.gname = buf[297:329].rstrip(NUL) - tarinfo.devmajor = nti(buf[329:337]) - tarinfo.devminor = nti(buf[337:345]) - prefix = buf[345:500].rstrip(NUL) - - if prefix and not tarinfo.issparse(): - tarinfo.name = prefix + "/" + tarinfo.name + obj = cls() + obj.buf = buf + obj.name = nts(buf[0:100]) + obj.mode = nti(buf[100:108]) + obj.uid = nti(buf[108:116]) + obj.gid = nti(buf[116:124]) + obj.size = nti(buf[124:136]) + 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.devmajor = nti(buf[329:337]) + obj.devminor = nti(buf[337:345]) + prefix = nts(buf[345:500]) + + # Old V7 tar format represents a directory as a regular + # file with a trailing slash. + if obj.type == AREGTYPE and obj.name.endswith("/"): + obj.type = DIRTYPE - return tarinfo + # Remove redundant slashes from directories. + if obj.isdir(): + obj.name = obj.name.rstrip("/") - def tobuf(self, posix=False): - """Return a tar header as a string of 512 byte blocks. - """ - buf = "" - type = self.type - prefix = "" + # Reconstruct a ustar longname. + if prefix and obj.type not in GNU_TYPES: + obj.name = prefix + "/" + obj.name + return obj - if self.name.endswith("/"): - type = DIRTYPE + @classmethod + def fromtarfile(cls, tarfile): + """Return the next TarInfo object from TarFile object + tarfile. + """ + buf = tarfile.fileobj.read(BLOCKSIZE) + if not buf: + return + obj = cls.frombuf(buf) + obj.offset = tarfile.fileobj.tell() - BLOCKSIZE + return obj._proc_member(tarfile) - if type in (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK): - # Prevent "././@LongLink" from being normalized. - name = self.name + #-------------------------------------------------------------------------- + # The following are methods that are called depending on the type of a + # member. The entry point is _proc_member() which can be overridden in a + # subclass to add custom _proc_*() methods. A _proc_*() method MUST + # implement the following + # operations: + # 1. Set self.offset_data to the position where the data blocks begin, + # if there is data that follows. + # 2. Set tarfile.offset to the position where the next member's header will + # begin. + # 3. Return self or another valid TarInfo object. + def _proc_member(self, tarfile): + """Choose the right processing method depending on + the type and call it. + """ + if self.type in (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK): + return self._proc_gnulong(tarfile) + elif self.type == GNUTYPE_SPARSE: + return self._proc_sparse(tarfile) + elif self.type in (XHDTYPE, XGLTYPE, SOLARIS_XHDTYPE): + return self._proc_pax(tarfile) else: - name = normpath(self.name) + return self._proc_builtin(tarfile) - if type == DIRTYPE: - # directories should end with '/' - name += "/" + def _proc_builtin(self, tarfile): + """Process a builtin type or an unknown type which + will be treated as a regular file. + """ + self.offset_data = tarfile.fileobj.tell() + offset = self.offset_data + if self.isreg() or self.type not in SUPPORTED_TYPES: + # Skip the following data blocks. + offset += self._block(self.size) + tarfile.offset = offset - linkname = self.linkname - if linkname: - # if linkname is empty we end up with a '.' - linkname = normpath(linkname) + # Patch the TarInfo object with saved extended + # header information. + for keyword, value in tarfile.pax_headers.iteritems(): + if keyword in PAX_FIELDS: + setattr(self, keyword, value) + self.pax_headers[keyword] = value - if posix: - if self.size > MAXSIZE_MEMBER: - raise ValueError("file is too large (>= 8 GB)") + return self - if len(self.linkname) > LENGTH_LINK: - raise ValueError("linkname is too long (>%d)" % (LENGTH_LINK)) + def _proc_gnulong(self, tarfile): + """Process the blocks that hold a GNU longname + or longlink member. + """ + buf = tarfile.fileobj.read(self._block(self.size)) - if len(name) > LENGTH_NAME: - prefix = name[:LENGTH_PREFIX + 1] - while prefix and prefix[-1] != "/": - prefix = prefix[:-1] + # Fetch the next header and process it. + b = tarfile.fileobj.read(BLOCKSIZE) + t = self.frombuf(b) + t.offset = self.offset + next = t._proc_member(tarfile) - name = name[len(prefix):] - prefix = prefix[:-1] + # Patch the TarInfo object from the next header with + # the longname information. + next.offset = self.offset + if self.type == GNUTYPE_LONGNAME: + next.name = buf.rstrip(NUL) + elif self.type == GNUTYPE_LONGLINK: + next.linkname = buf.rstrip(NUL) - if not prefix or len(name) > LENGTH_NAME: - raise ValueError("name is too long") + return next - else: - if len(self.linkname) > LENGTH_LINK: - buf += self._create_gnulong(self.linkname, GNUTYPE_LONGLINK) + def _proc_sparse(self, tarfile): + """Process a GNU sparse header plus extra headers. + """ + buf = self.buf + sp = _ringbuffer() + pos = 386 + lastpos = 0L + realpos = 0L + # There are 4 possible sparse structs in the + # first header. + for i in xrange(4): + try: + offset = nti(buf[pos:pos + 12]) + numbytes = nti(buf[pos + 12:pos + 24]) + except ValueError: + break + if offset > lastpos: + sp.append(_hole(lastpos, offset - lastpos)) + sp.append(_data(offset, numbytes, realpos)) + realpos += numbytes + lastpos = offset + numbytes + pos += 24 - if len(name) > LENGTH_NAME: - buf += self._create_gnulong(name, GNUTYPE_LONGNAME) + isextended = ord(buf[482]) + origsize = nti(buf[483:495]) - parts = [ - stn(name, 100), - itn(self.mode & 07777, 8, posix), - itn(self.uid, 8, posix), - itn(self.gid, 8, posix), - itn(self.size, 12, posix), - itn(self.mtime, 12, posix), - " ", # checksum field - type, - stn(self.linkname, 100), - stn(MAGIC, 6), - stn(VERSION, 2), - stn(self.uname, 32), - stn(self.gname, 32), - itn(self.devmajor, 8, posix), - itn(self.devminor, 8, posix), - stn(prefix, 155) - ] + # If the isextended flag is given, + # there are extra headers to process. + while isextended == 1: + buf = tarfile.fileobj.read(BLOCKSIZE) + pos = 0 + for i in xrange(21): + try: + offset = nti(buf[pos:pos + 12]) + numbytes = nti(buf[pos + 12:pos + 24]) + except ValueError: + break + if offset > lastpos: + sp.append(_hole(lastpos, offset - lastpos)) + sp.append(_data(offset, numbytes, realpos)) + realpos += numbytes + lastpos = offset + numbytes + pos += 24 + isextended = ord(buf[504]) - buf += struct.pack("%ds" % BLOCKSIZE, "".join(parts)) - chksum = calc_chksums(buf[-BLOCKSIZE:])[0] - buf = buf[:-364] + "%06o\0" % chksum + buf[-357:] - self.buf = buf - return buf + if lastpos < origsize: + sp.append(_hole(lastpos, origsize - lastpos)) + + self.sparse = sp - def _create_gnulong(self, name, type): - """Create a GNU longname/longlink header from name. - It consists of an extended tar header, with the length - of the longname as size, followed by data blocks, - which contain the longname as a null terminated string. + self.offset_data = tarfile.fileobj.tell() + tarfile.offset = self.offset_data + self._block(self.size) + self.size = origsize + + return self + + def _proc_pax(self, tarfile): + """Process an extended or global header as described in + POSIX.1-2001. """ - name += NUL + # Read the header information. + buf = tarfile.fileobj.read(self._block(self.size)) - tarinfo = self.__class__() - tarinfo.name = "././@LongLink" - tarinfo.type = type - tarinfo.mode = 0 - tarinfo.size = len(name) - - # create extended header - buf = tarinfo.tobuf() - # create name blocks - buf += name - blocks, remainder = divmod(len(name), BLOCKSIZE) - if remainder > 0: - buf += (BLOCKSIZE - remainder) * NUL - return buf + # A pax header stores supplemental information for either + # the following file (extended) or all following files + # (global). + if self.type == XGLTYPE: + pax_headers = tarfile.pax_headers + else: + pax_headers = tarfile.pax_headers.copy() + + # Fields in POSIX.1-2001 that are numbers, all other fields + # are treated as UTF-8 strings. + type_mapping = { + "atime": float, + "ctime": float, + "mtime": float, + "uid": int, + "gid": int, + "size": int + } + + # Parse pax header information. A record looks like that: + # "%d %s=%s\n" % (length, keyword, value). length is the size + # of the complete record including the length field itself and + # the newline. + regex = re.compile(r"(\d+) ([^=]+)=", re.U) + pos = 0 + while True: + match = regex.match(buf, pos) + if not match: + break + + length, keyword = match.groups() + length = int(length) + value = buf[match.end(2) + 1:match.start(1) + length - 1] + + keyword = keyword.decode("utf8") + keyword = keyword.encode(tarfile.encoding) + + value = value.decode("utf8") + if keyword in type_mapping: + try: + value = type_mapping[keyword](value) + except ValueError: + value = 0 + else: + value = value.encode(tarfile.encoding) + + pax_headers[keyword] = value + pos += length + + # Fetch the next header that will be patched with the + # supplement information from the pax header (extended + # only). + t = self.fromtarfile(tarfile) + + if self.type != XGLTYPE and t is not None: + # Patch the TarInfo object from the next header with + # the pax header's information. + for keyword, value in pax_headers.items(): + if keyword in PAX_FIELDS: + setattr(t, keyword, value) + pax_headers[keyword] = value + t.pax_headers = pax_headers.copy() + + return t + + def _block(self, count): + """Round up a byte count by BLOCKSIZE and return it, + e.g. _block(834) => 1024. + """ + blocks, remainder = divmod(count, BLOCKSIZE) + if remainder: + blocks += 1 + return blocks * BLOCKSIZE def isreg(self): return self.type in REGULAR_TYPES @@ -1038,12 +1460,18 @@ class TarFile(object): # messages (if debug >= 0). If > 0, errors # are passed to the caller as exceptions. - posix = False # If True, generates POSIX.1-1990-compliant - # archives (no GNU extensions!) + format = DEFAULT_FORMAT # The format to use when creating an archive. + + encoding = ENCODING # Transfer UTF-8 strings from POSIX.1-2001 + # headers to this encoding. + + tarinfo = TarInfo # The default TarInfo class to use. - fileobject = ExFileObject + fileobject = ExFileObject # The default ExFileObject class to use. - def __init__(self, name=None, mode="r", fileobj=None): + def __init__(self, name=None, mode="r", fileobj=None, format=None, + tarinfo=None, dereference=None, ignore_zeros=None, encoding=None, + pax_headers=None, debug=None, errorlevel=None): """Open an (uncompressed) tar archive `name'. `mode' is either 'r' to read from an existing archive, 'a' to append data to an existing file or 'w' to create a new file overwriting an existing one. `mode' @@ -1052,58 +1480,86 @@ class TarFile(object): can be determined, `mode' is overridden by `fileobj's mode. `fileobj' is not closed, when TarFile is closed. """ - self.name = os.path.abspath(name) - if len(mode) > 1 or mode not in "raw": raise ValueError("mode must be 'r', 'a' or 'w'") - self._mode = mode - self.mode = {"r": "rb", "a": "r+b", "w": "wb"}[mode] + self.mode = mode + self._mode = {"r": "rb", "a": "r+b", "w": "wb"}[mode] if not fileobj: - if self._mode == "a" and not os.path.exists(self.name): + if self.mode == "a" and not os.path.exists(name): # Create nonexistent files in append mode. - self._mode = "w" - self.mode = "wb" - fileobj = file(self.name, self.mode) + self.mode = "w" + self._mode = "wb" + fileobj = file(name, self._mode) self._extfileobj = False else: - if self.name is None and hasattr(fileobj, "name"): - self.name = os.path.abspath(fileobj.name) + if name is None and hasattr(fileobj, "name"): + name = fileobj.name if hasattr(fileobj, "mode"): - self.mode = fileobj.mode + self._mode = fileobj.mode self._extfileobj = True + self.name = os.path.abspath(name) self.fileobj = fileobj - # Init datastructures + # Init attributes. + if format is not None: + self.format = format + if tarinfo is not None: + self.tarinfo = tarinfo + if dereference is not None: + self.dereference = dereference + if ignore_zeros is not None: + self.ignore_zeros = ignore_zeros + if encoding is not None: + self.encoding = encoding + if debug is not None: + self.debug = debug + if errorlevel is not None: + self.errorlevel = errorlevel + + # Init datastructures. self.closed = False self.members = [] # list of members as TarInfo objects self._loaded = False # flag if all members have been read self.offset = 0L # current position in the archive file self.inodes = {} # dictionary caching the inodes of # archive members already added + self.pax_headers = {} # save contents of global pax headers - if self._mode == "r": + if self.mode == "r": self.firstmember = None self.firstmember = self.next() - if self._mode == "a": + if self.mode == "a": # Move to the end of the archive, # before the first empty block. self.firstmember = None while True: - try: - tarinfo = self.next() - except ReadError: - self.fileobj.seek(0) - break - if tarinfo is None: + if self.next() is None: if self.offset > 0: self.fileobj.seek(- BLOCKSIZE, 1) break - if self._mode in "aw": + if self.mode in "aw": self._loaded = True + if pax_headers: + buf = self.tarinfo.create_pax_global_header( + pax_headers.copy(), self.encoding) + self.fileobj.write(buf) + self.offset += len(buf) + + def _getposix(self): + return self.format == USTAR_FORMAT + def _setposix(self, value): + import warnings + warnings.warn("use the format attribute instead", DeprecationWarning) + if value: + self.format = USTAR_FORMAT + else: + self.format = GNU_FORMAT + posix = property(_getposix, _setposix) + #-------------------------------------------------------------------------- # Below are the classmethods which act as alternate constructors to the # TarFile class. The open() method is the only one that is needed for @@ -1116,7 +1572,7 @@ class TarFile(object): # by adding it to the mapping in OPEN_METH. @classmethod - def open(cls, name=None, mode="r", fileobj=None, bufsize=20*512): + def open(cls, name=None, mode="r", fileobj=None, bufsize=RECORDSIZE, **kwargs): """Open a tar archive for reading, writing or appending. Return an appropriate TarFile class. @@ -1149,8 +1605,8 @@ class TarFile(object): if fileobj is not None: saved_pos = fileobj.tell() try: - return func(name, "r", fileobj) - except (ReadError, CompressionError): + return func(name, "r", fileobj, **kwargs) + except (ReadError, CompressionError), e: if fileobj is not None: fileobj.seek(saved_pos) continue @@ -1167,7 +1623,7 @@ class TarFile(object): func = getattr(cls, cls.OPEN_METH[comptype]) else: raise CompressionError("unknown compression type %r" % comptype) - return func(name, filemode, fileobj) + return func(name, filemode, fileobj, **kwargs) elif "|" in mode: filemode, comptype = mode.split("|", 1) @@ -1178,25 +1634,26 @@ class TarFile(object): raise ValueError("mode must be 'r' or 'w'") t = cls(name, filemode, - _Stream(name, filemode, comptype, fileobj, bufsize)) + _Stream(name, filemode, comptype, fileobj, bufsize), + **kwargs) t._extfileobj = False return t elif mode in "aw": - return cls.taropen(name, mode, fileobj) + return cls.taropen(name, mode, fileobj, **kwargs) raise ValueError("undiscernible mode") @classmethod - def taropen(cls, name, mode="r", fileobj=None): + def taropen(cls, name, mode="r", fileobj=None, **kwargs): """Open uncompressed tar archive name for reading or writing. """ if len(mode) > 1 or mode not in "raw": raise ValueError("mode must be 'r', 'a' or 'w'") - return cls(name, mode, fileobj) + return cls(name, mode, fileobj, **kwargs) @classmethod - def gzopen(cls, name, mode="r", fileobj=None, compresslevel=9): + def gzopen(cls, name, mode="r", fileobj=None, compresslevel=9, **kwargs): """Open gzip compressed tar archive name for reading or writing. Appending is not allowed. """ @@ -1214,14 +1671,15 @@ class TarFile(object): try: t = cls.taropen(name, mode, - gzip.GzipFile(name, mode, compresslevel, fileobj)) + gzip.GzipFile(name, mode, compresslevel, fileobj), + **kwargs) except IOError: raise ReadError("not a gzip file") t._extfileobj = False return t @classmethod - def bz2open(cls, name, mode="r", fileobj=None, compresslevel=9): + def bz2open(cls, name, mode="r", fileobj=None, compresslevel=9, **kwargs): """Open bzip2 compressed tar archive name for reading or writing. Appending is not allowed. """ @@ -1239,7 +1697,7 @@ class TarFile(object): fileobj = bz2.BZ2File(name, mode, compresslevel=compresslevel) try: - t = cls.taropen(name, mode, fileobj) + t = cls.taropen(name, mode, fileobj, **kwargs) except IOError: raise ReadError("not a bzip2 file") t._extfileobj = False @@ -1262,7 +1720,7 @@ class TarFile(object): if self.closed: return - if self._mode in "aw": + if self.mode in "aw": self.fileobj.write(NUL * (BLOCKSIZE * 2)) self.offset += (BLOCKSIZE * 2) # fill up the end with zero-blocks @@ -1328,7 +1786,8 @@ class TarFile(object): # Now, fill the TarInfo object with # information specific for the file. - tarinfo = TarInfo() + tarinfo = self.tarinfo() + tarinfo.tarfile = self # Use os.stat or os.lstat, depending on platform # and if symlinks shall be resolved. @@ -1344,8 +1803,8 @@ class TarFile(object): stmd = statres.st_mode if stat.S_ISREG(stmd): inode = (statres.st_ino, statres.st_dev) - if not self.dereference and \ - statres.st_nlink > 1 and inode in self.inodes: + if not self.dereference and statres.st_nlink > 1 and \ + inode in self.inodes and arcname != self.inodes[inode]: # Is it a hardlink to an already # archived file? type = LNKTYPE @@ -1422,7 +1881,7 @@ class TarFile(object): print "%d-%02d-%02d %02d:%02d:%02d" \ % time.localtime(tarinfo.mtime)[:6], - print tarinfo.name, + print tarinfo.name + ("/" if tarinfo.isdir() else ""), if verbose: if tarinfo.issym(): @@ -1454,7 +1913,7 @@ class TarFile(object): if recursive: if arcname == ".": arcname = "" - for f in os.listdir("."): + for f in os.listdir(name): self.add(f, os.path.join(arcname, f)) return @@ -1493,7 +1952,7 @@ class TarFile(object): tarinfo = copy.copy(tarinfo) - buf = tarinfo.tobuf(self.posix) + buf = tarinfo.tobuf(self.format, self.encoding) self.fileobj.write(buf) self.offset += len(buf) @@ -1525,7 +1984,7 @@ class TarFile(object): # Extract directory with a safe mode, so that # all files below can be extracted as well. try: - os.makedirs(os.path.join(path, tarinfo.name), 0777) + os.makedirs(os.path.join(path, tarinfo.name), 0700) except EnvironmentError: pass directories.append(tarinfo) @@ -1557,10 +2016,10 @@ class TarFile(object): """ self._check("r") - if isinstance(member, TarInfo): - tarinfo = member - else: + if isinstance(member, basestring): tarinfo = self.getmember(member) + else: + tarinfo = member # Prepare the link target for makelink(). if tarinfo.islnk(): @@ -1593,10 +2052,10 @@ class TarFile(object): """ self._check("r") - if isinstance(member, TarInfo): - tarinfo = member - else: + if isinstance(member, basestring): tarinfo = self.getmember(member) + else: + tarinfo = member if tarinfo.isreg(): return self.fileobject(self, tarinfo) @@ -1809,20 +2268,11 @@ class TarFile(object): # Read the next block. self.fileobj.seek(self.offset) while True: - buf = self.fileobj.read(BLOCKSIZE) - if not buf: - return None - try: - tarinfo = TarInfo.frombuf(buf) - - # Set the TarInfo object's offset to the current position of the - # TarFile and set self.offset to the position where the data blocks - # should begin. - tarinfo.offset = self.offset - self.offset += BLOCKSIZE - - tarinfo = self.proc_member(tarinfo) + tarinfo = self.tarinfo.fromtarfile(self) + if tarinfo is None: + return + self.members.append(tarinfo) except HeaderError, e: if self.ignore_zeros: @@ -1835,149 +2285,11 @@ class TarFile(object): return None break - # Some old tar programs represent a directory as a regular - # file with a trailing slash. - if tarinfo.isreg() and tarinfo.name.endswith("/"): - tarinfo.type = DIRTYPE - - # Directory names should have a '/' at the end. - if tarinfo.isdir(): - tarinfo.name += "/" - - self.members.append(tarinfo) - return tarinfo - - #-------------------------------------------------------------------------- - # The following are methods that are called depending on the type of a - # member. The entry point is proc_member() which is called with a TarInfo - # object created from the header block from the current offset. The - # proc_member() method can be overridden in a subclass to add custom - # proc_*() methods. A proc_*() method MUST implement the following - # operations: - # 1. Set tarinfo.offset_data to the position where the data blocks begin, - # if there is data that follows. - # 2. Set self.offset to the position where the next member's header will - # begin. - # 3. Return tarinfo or another valid TarInfo object. - def proc_member(self, tarinfo): - """Choose the right processing method for tarinfo depending - on its type and call it. - """ - if tarinfo.type in (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK): - return self.proc_gnulong(tarinfo) - elif tarinfo.type == GNUTYPE_SPARSE: - return self.proc_sparse(tarinfo) - else: - return self.proc_builtin(tarinfo) - - def proc_builtin(self, tarinfo): - """Process a builtin type member or an unknown member - which will be treated as a regular file. - """ - tarinfo.offset_data = self.offset - if tarinfo.isreg() or tarinfo.type not in SUPPORTED_TYPES: - # Skip the following data blocks. - self.offset += self._block(tarinfo.size) - return tarinfo - - def proc_gnulong(self, tarinfo): - """Process the blocks that hold a GNU longname - or longlink member. - """ - buf = "" - count = tarinfo.size - while count > 0: - block = self.fileobj.read(BLOCKSIZE) - buf += block - self.offset += BLOCKSIZE - count -= BLOCKSIZE - - # Fetch the next header and process it. - b = self.fileobj.read(BLOCKSIZE) - t = TarInfo.frombuf(b) - t.offset = self.offset - self.offset += BLOCKSIZE - next = self.proc_member(t) - - # Patch the TarInfo object from the next header with - # the longname information. - next.offset = tarinfo.offset - if tarinfo.type == GNUTYPE_LONGNAME: - next.name = buf.rstrip(NUL) - elif tarinfo.type == GNUTYPE_LONGLINK: - next.linkname = buf.rstrip(NUL) - - return next - - def proc_sparse(self, tarinfo): - """Process a GNU sparse header plus extra headers. - """ - buf = tarinfo.buf - sp = _ringbuffer() - pos = 386 - lastpos = 0L - realpos = 0L - # There are 4 possible sparse structs in the - # first header. - for i in xrange(4): - try: - offset = nti(buf[pos:pos + 12]) - numbytes = nti(buf[pos + 12:pos + 24]) - except ValueError: - break - if offset > lastpos: - sp.append(_hole(lastpos, offset - lastpos)) - sp.append(_data(offset, numbytes, realpos)) - realpos += numbytes - lastpos = offset + numbytes - pos += 24 - - isextended = ord(buf[482]) - origsize = nti(buf[483:495]) - - # If the isextended flag is given, - # there are extra headers to process. - while isextended == 1: - buf = self.fileobj.read(BLOCKSIZE) - self.offset += BLOCKSIZE - pos = 0 - for i in xrange(21): - try: - offset = nti(buf[pos:pos + 12]) - numbytes = nti(buf[pos + 12:pos + 24]) - except ValueError: - break - if offset > lastpos: - sp.append(_hole(lastpos, offset - lastpos)) - sp.append(_data(offset, numbytes, realpos)) - realpos += numbytes - lastpos = offset + numbytes - pos += 24 - isextended = ord(buf[504]) - - if lastpos < origsize: - sp.append(_hole(lastpos, origsize - lastpos)) - - tarinfo.sparse = sp - - tarinfo.offset_data = self.offset - self.offset += self._block(tarinfo.size) - tarinfo.size = origsize - return tarinfo #-------------------------------------------------------------------------- # Little helper methods: - def _block(self, count): - """Round up a byte count by BLOCKSIZE and return it, - e.g. _block(834) => 1024. - """ - blocks, remainder = divmod(count, BLOCKSIZE) - if remainder: - blocks += 1 - return blocks * BLOCKSIZE - def _getmember(self, name, tarinfo=None): """Find an archive member by name from bottom to top. If tarinfo is given, it is used as the starting point. @@ -2010,8 +2322,8 @@ class TarFile(object): """ if self.closed: raise IOError("%s is closed" % self.__class__.__name__) - if mode is not None and self._mode not in mode: - raise IOError("bad operation for mode %r" % self._mode) + if mode is not None and self.mode not in mode: + raise IOError("bad operation for mode %r" % self.mode) def __iter__(self): """Provide an iterator object. diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index f2dd97d..573489a 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -1,8 +1,12 @@ +# encoding: iso8859-15 + import sys import os import shutil import tempfile import StringIO +import md5 +import errno import unittest import tarfile @@ -20,452 +24,546 @@ try: except ImportError: bz2 = None +def md5sum(data): + return md5.new(data).hexdigest() + def path(path): return test_support.findfile(path) -testtar = path("testtar.tar") -tempdir = os.path.join(tempfile.gettempdir(), "testtar" + os.extsep + "dir") -tempname = test_support.TESTFN -membercount = 12 - -def tarname(comp=""): - if not comp: - return testtar - return os.path.join(tempdir, "%s%s%s" % (testtar, os.extsep, comp)) +TEMPDIR = os.path.join(tempfile.gettempdir(), "test_tarfile_tmp") +tarname = path("testtar.tar") +gzipname = os.path.join(TEMPDIR, "testtar.tar.gz") +bz2name = os.path.join(TEMPDIR, "testtar.tar.bz2") +tmpname = os.path.join(TEMPDIR, "tmp.tar") -def dirname(): - if not os.path.exists(tempdir): - os.mkdir(tempdir) - return tempdir +md5_regtype = "65f477c818ad9e15f7feab0c6d37742f" +md5_sparse = "a54fbc4ca4f4399a90e1b27164012fc6" -def tmpname(): - return tempname +class ReadTest(unittest.TestCase): -class BaseTest(unittest.TestCase): - comp = '' - mode = 'r' - sep = ':' + tarname = tarname + mode = "r:" def setUp(self): - mode = self.mode + self.sep + self.comp - self.tar = tarfile.open(tarname(self.comp), mode) + self.tar = tarfile.open(self.tarname, mode=self.mode) def tearDown(self): self.tar.close() -class ReadTest(BaseTest): - def test(self): - """Test member extraction. - """ - members = 0 +class UstarReadTest(ReadTest): + + def test_fileobj_regular_file(self): + tarinfo = self.tar.getmember("ustar/regtype") + fobj = self.tar.extractfile(tarinfo) + data = fobj.read() + self.assert_((len(data), md5sum(data)) == (tarinfo.size, md5_regtype), + "regular file extraction failed") + + 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) + + lines1 = fobj1.readlines() + lines2 = fobj2.readlines() + self.assert_(lines1 == lines2, + "fileobj.readlines() failed") + self.assert_(len(lines2) == 114, + "fileobj.readlines() failed") + self.assert_(lines2[83] == \ + "I will gladly admit that Python is not the fastest running scripting language.\n", + "fileobj.readlines() failed") + + def test_fileobj_iter(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) + lines1 = fobj1.readlines() + lines2 = [line for line in fobj2] + self.assert_(lines1 == lines2, + "fileobj.__iter__() failed") + + def test_fileobj_seek(self): + self.tar.extract("ustar/regtype", TEMPDIR) + fobj = open(os.path.join(TEMPDIR, "ustar/regtype"), "rb") + data = fobj.read() + fobj.close() + + tarinfo = self.tar.getmember("ustar/regtype") + fobj = self.tar.extractfile(tarinfo) + + text = fobj.read() + fobj.seek(0) + self.assert_(0 == fobj.tell(), + "seek() to file's start failed") + fobj.seek(2048, 0) + self.assert_(2048 == fobj.tell(), + "seek() to absolute position failed") + fobj.seek(-1024, 1) + self.assert_(1024 == fobj.tell(), + "seek() to negative relative position failed") + fobj.seek(1024, 1) + self.assert_(2048 == fobj.tell(), + "seek() to positive relative position failed") + s = fobj.read(10) + self.assert_(s == data[2048:2058], + "read() after seek failed") + fobj.seek(0, 2) + self.assert_(tarinfo.size == fobj.tell(), + "seek() to file's end failed") + self.assert_(fobj.read() == "", + "read() at file's end did not return empty string") + fobj.seek(-tarinfo.size, 2) + self.assert_(0 == fobj.tell(), + "relative seek() to file's start failed") + fobj.seek(512) + s1 = fobj.readlines() + fobj.seek(512) + s2 = fobj.readlines() + self.assert_(s1 == s2, + "readlines() after seek failed") + fobj.seek(0) + self.assert_(len(fobj.readline()) == fobj.tell(), + "tell() after readline() failed") + fobj.seek(512) + self.assert_(len(fobj.readline()) + 512 == fobj.tell(), + "tell() after seek() and readline() failed") + fobj.seek(0) + line = fobj.readline() + self.assert_(fobj.read() == data[len(line):], + "read() after readline() failed") + fobj.close() + + +class MiscReadTest(ReadTest): + + def test_no_filename(self): + fobj = open(self.tarname, "rb") + tar = tarfile.open(fileobj=fobj, mode=self.mode) + self.assertEqual(tar.name, os.path.abspath(fobj.name)) + + def test_fail_comp(self): + # For Gzip and Bz2 Tests: fail with a ReadError on an uncompressed file. + if self.mode == "r:": + return + self.assertRaises(tarfile.ReadError, tarfile.open, tarname, self.mode) + fobj = open(tarname, "rb") + self.assertRaises(tarfile.ReadError, tarfile.open, fileobj=fobj, mode=self.mode) + + def test_v7_dirtype(self): + # Test old style dirtype member (bug #1336623): + # Old V7 tars create directory members using an AREGTYPE + # header with a "/" appended to the filename field. + tarinfo = self.tar.getmember("misc/dirtype-old-v7") + self.assert_(tarinfo.type == tarfile.DIRTYPE, + "v7 dirtype failed") + + def test_check_members(self): for tarinfo in self.tar: - members += 1 - if not tarinfo.isreg(): + self.assert_(int(tarinfo.mtime) == 07606136617, + "wrong mtime for %s" % tarinfo.name) + if not tarinfo.name.startswith("ustar/"): continue - f = self.tar.extractfile(tarinfo) - self.assert_(len(f.read()) == tarinfo.size, - "size read does not match expected size") - f.close() - - self.assert_(members == membercount, - "could not find all members") - - def test_sparse(self): - """Test sparse member extraction. - """ - if self.sep != "|": - f1 = self.tar.extractfile("S-SPARSE") - f2 = self.tar.extractfile("S-SPARSE-WITH-NULLS") - self.assert_(f1.read() == f2.read(), - "_FileObject failed on sparse file member") - - def test_readlines(self): - """Test readlines() method of _FileObject. - """ - if self.sep != "|": - filename = "0-REGTYPE-TEXT" - self.tar.extract(filename, dirname()) - f = open(os.path.join(dirname(), filename), "rU") - lines1 = f.readlines() - f.close() - lines2 = self.tar.extractfile(filename).readlines() - self.assert_(lines1 == lines2, - "_FileObject.readline() does not work correctly") - - def test_iter(self): - # Test iteration over ExFileObject. - if self.sep != "|": - filename = "0-REGTYPE-TEXT" - self.tar.extract(filename, dirname()) - f = open(os.path.join(dirname(), filename), "rU") - lines1 = f.readlines() - f.close() - lines2 = [line for line in self.tar.extractfile(filename)] - self.assert_(lines1 == lines2, - "ExFileObject iteration does not work correctly") - - def test_seek(self): - """Test seek() method of _FileObject, incl. random reading. - """ - if self.sep != "|": - filename = "0-REGTYPE-TEXT" - self.tar.extract(filename, dirname()) - f = open(os.path.join(dirname(), filename), "rb") - data = f.read() - f.close() - - tarinfo = self.tar.getmember(filename) - fobj = self.tar.extractfile(tarinfo) - - text = fobj.read() - fobj.seek(0) - self.assert_(0 == fobj.tell(), - "seek() to file's start failed") - fobj.seek(2048, 0) - self.assert_(2048 == fobj.tell(), - "seek() to absolute position failed") - fobj.seek(-1024, 1) - self.assert_(1024 == fobj.tell(), - "seek() to negative relative position failed") - fobj.seek(1024, 1) - self.assert_(2048 == fobj.tell(), - "seek() to positive relative position failed") - s = fobj.read(10) - self.assert_(s == data[2048:2058], - "read() after seek failed") - fobj.seek(0, 2) - self.assert_(tarinfo.size == fobj.tell(), - "seek() to file's end failed") - self.assert_(fobj.read() == "", - "read() at file's end did not return empty string") - fobj.seek(-tarinfo.size, 2) - self.assert_(0 == fobj.tell(), - "relative seek() to file's start failed") - fobj.seek(512) - s1 = fobj.readlines() - fobj.seek(512) - s2 = fobj.readlines() - self.assert_(s1 == s2, - "readlines() after seek failed") - fobj.seek(0) - self.assert_(len(fobj.readline()) == fobj.tell(), - "tell() after readline() failed") - fobj.seek(512) - self.assert_(len(fobj.readline()) + 512 == fobj.tell(), - "tell() after seek() and readline() failed") - fobj.seek(0) - line = fobj.readline() - self.assert_(fobj.read() == data[len(line):], - "read() after readline() failed") - fobj.close() + self.assert_(tarinfo.uname == "tarfile", + "wrong uname for %s" % tarinfo.name) - def test_old_dirtype(self): - """Test old style dirtype member (bug #1336623). - """ - # Old tars create directory members using a REGTYPE - # header with a "/" appended to the filename field. + def test_find_members(self): + self.assert_(self.tar.getmembers()[-1].name == "misc/eof", + "could not find all members") - # Create an old tar style directory entry. - filename = tmpname() - tarinfo = tarfile.TarInfo("directory/") - tarinfo.type = tarfile.REGTYPE + def test_extract_hardlink(self): + # Test hardlink extraction (e.g. bug #857297). + tar = tarfile.open(tarname, errorlevel=1) - fobj = open(filename, "w") - fobj.write(tarinfo.tobuf()) - fobj.close() + tar.extract("ustar/regtype", TEMPDIR) + try: + tar.extract("ustar/lnktype", TEMPDIR) + except EnvironmentError, e: + if e.errno == errno.ENOENT: + self.fail("hardlink not extracted properly") + + data = open(os.path.join(TEMPDIR, "ustar/lnktype"), "rb").read() + self.assertEqual(md5sum(data), md5_regtype) try: - # Test if it is still a directory entry when - # read back. - tar = tarfile.open(filename) - tarinfo = tar.getmembers()[0] - tar.close() - - self.assert_(tarinfo.type == tarfile.DIRTYPE) - self.assert_(tarinfo.name.endswith("/")) - finally: - try: - os.unlink(filename) - except: - pass - -class ReadStreamTest(ReadTest): - sep = "|" - - def test(self): - """Test member extraction, and for StreamError when - seeking backwards. - """ - ReadTest.test(self) - tarinfo = self.tar.getmembers()[0] - f = self.tar.extractfile(tarinfo) + tar.extract("ustar/symtype", TEMPDIR) + except EnvironmentError, e: + if e.errno == errno.ENOENT: + self.fail("symlink not extracted properly") + + data = open(os.path.join(TEMPDIR, "ustar/symtype"), "rb").read() + self.assertEqual(md5sum(data), md5_regtype) + + +class StreamReadTest(ReadTest): + + mode="r|" + + def test_fileobj_regular_file(self): + tarinfo = self.tar.next() # get "regtype" (can't use getmember) + fobj = self.tar.extractfile(tarinfo) + data = fobj.read() + self.assert_((len(data), md5sum(data)) == (tarinfo.size, md5_regtype), + "regular file extraction failed") + + def test_provoke_stream_error(self): + tarinfos = self.tar.getmembers() + f = self.tar.extractfile(tarinfos[0]) # read the first member self.assertRaises(tarfile.StreamError, f.read) - def test_stream(self): - """Compare the normal tar and the stream tar. - """ - stream = self.tar - tar = tarfile.open(tarname(), 'r') + def test_compare_members(self): + tar1 = tarfile.open(tarname) + tar2 = self.tar - while 1: - t1 = tar.next() - t2 = stream.next() + while True: + t1 = tar1.next() + t2 = tar2.next() if t1 is None: break self.assert_(t2 is not None, "stream.next() failed.") if t2.islnk() or t2.issym(): - self.assertRaises(tarfile.StreamError, stream.extractfile, t2) + self.assertRaises(tarfile.StreamError, tar2.extractfile, t2) continue - v1 = tar.extractfile(t1) - v2 = stream.extractfile(t2) + + v1 = tar1.extractfile(t1) + v2 = tar2.extractfile(t2) if v1 is None: continue self.assert_(v2 is not None, "stream.extractfile() failed") self.assert_(v1.read() == v2.read(), "stream extraction failed") - tar.close() - stream.close() + tar1.close() -class ReadDetectTest(ReadTest): - def setUp(self): - self.tar = tarfile.open(tarname(self.comp), self.mode) +class DetectReadTest(unittest.TestCase): -class ReadDetectFileobjTest(ReadTest): + def _testfunc_file(self, name, mode): + try: + tarfile.open(name, mode) + except tarfile.ReadError: + self.fail() - def setUp(self): - name = tarname(self.comp) - self.tar = tarfile.open(name, mode=self.mode, - fileobj=open(name, "rb")) + def _testfunc_fileobj(self, name, mode): + try: + tarfile.open(name, mode, fileobj=open(name, "rb")) + except tarfile.ReadError: + self.fail() -class ReadAsteriskTest(ReadTest): + def _test_modes(self, testfunc): + testfunc(tarname, "r") + testfunc(tarname, "r:") + testfunc(tarname, "r:*") + testfunc(tarname, "r|") + testfunc(tarname, "r|*") - def setUp(self): - mode = self.mode + self.sep + "*" - self.tar = tarfile.open(tarname(self.comp), mode) + if gzip: + self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r:gz") + self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r|gz") + self.assertRaises(tarfile.ReadError, tarfile.open, gzipname, mode="r:") + self.assertRaises(tarfile.ReadError, tarfile.open, gzipname, mode="r|") -class ReadStreamAsteriskTest(ReadStreamTest): + testfunc(gzipname, "r") + testfunc(gzipname, "r:*") + testfunc(gzipname, "r:gz") + testfunc(gzipname, "r|*") + testfunc(gzipname, "r|gz") - def setUp(self): - mode = self.mode + self.sep + "*" - self.tar = tarfile.open(tarname(self.comp), mode) + if bz2: + self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r:bz2") + self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r|bz2") + self.assertRaises(tarfile.ReadError, tarfile.open, bz2name, mode="r:") + self.assertRaises(tarfile.ReadError, tarfile.open, bz2name, mode="r|") -class WriteTest(BaseTest): - mode = 'w' + testfunc(bz2name, "r") + testfunc(bz2name, "r:*") + testfunc(bz2name, "r:bz2") + testfunc(bz2name, "r|*") + testfunc(bz2name, "r|bz2") - def setUp(self): - mode = self.mode + self.sep + self.comp - self.src = tarfile.open(tarname(self.comp), 'r') - self.dstname = tmpname() - self.dst = tarfile.open(self.dstname, mode) + def test_detect_file(self): + self._test_modes(self._testfunc_file) - def tearDown(self): - self.src.close() - self.dst.close() + def test_detect_fileobj(self): + self._test_modes(self._testfunc_fileobj) - def test_posix(self): - self.dst.posix = 1 - self._test() - def test_nonposix(self): - self.dst.posix = 0 - self._test() +class MemberReadTest(ReadTest): - def test_small(self): - self.dst.add(os.path.join(os.path.dirname(__file__),"cfgparser.1")) - self.dst.close() - self.assertNotEqual(os.stat(self.dstname).st_size, 0) + def _test_member(self, tarinfo, chksum=None, **kwargs): + if chksum is not None: + self.assert_(md5sum(self.tar.extractfile(tarinfo).read()) == chksum, + "wrong md5sum for %s" % tarinfo.name) - def _test(self): - for tarinfo in self.src: - if not tarinfo.isreg(): - continue - f = self.src.extractfile(tarinfo) - if self.dst.posix and len(tarinfo.name) > tarfile.LENGTH_NAME and "/" not in tarinfo.name: - self.assertRaises(ValueError, self.dst.addfile, - tarinfo, f) - else: - self.dst.addfile(tarinfo, f) + kwargs["mtime"] = 07606136617 + kwargs["uid"] = 1000 + kwargs["gid"] = 100 + if "old-v7" not in tarinfo.name: + # V7 tar can't handle alphabetic owners. + kwargs["uname"] = "tarfile" + kwargs["gname"] = "tarfile" + for k, v in kwargs.iteritems(): + self.assert_(getattr(tarinfo, k) == v, + "wrong value in %s field of %s" % (k, tarinfo.name)) - def test_add_self(self): - dstname = os.path.abspath(self.dstname) + def test_find_regtype(self): + tarinfo = self.tar.getmember("ustar/regtype") + self._test_member(tarinfo, size=7011, chksum=md5_regtype) - self.assertEqual(self.dst.name, dstname, "archive name must be absolute") + def test_find_conttype(self): + tarinfo = self.tar.getmember("ustar/conttype") + self._test_member(tarinfo, size=7011, chksum=md5_regtype) - self.dst.add(dstname) - self.assertEqual(self.dst.getnames(), [], "added the archive to itself") + def test_find_dirtype(self): + tarinfo = self.tar.getmember("ustar/dirtype") + self._test_member(tarinfo, size=0) - cwd = os.getcwd() - os.chdir(dirname()) - self.dst.add(dstname) - os.chdir(cwd) - self.assertEqual(self.dst.getnames(), [], "added the archive to itself") + def test_find_dirtype_with_size(self): + tarinfo = self.tar.getmember("ustar/dirtype-with-size") + self._test_member(tarinfo, size=255) + def test_find_lnktype(self): + tarinfo = self.tar.getmember("ustar/lnktype") + self._test_member(tarinfo, size=0, linkname="ustar/regtype") -class AppendTest(unittest.TestCase): - # Test append mode (cp. patch #1652681). + def test_find_symtype(self): + tarinfo = self.tar.getmember("ustar/symtype") + self._test_member(tarinfo, size=0, linkname="regtype") - def setUp(self): - self.tarname = tmpname() - if os.path.exists(self.tarname): - os.remove(self.tarname) + def test_find_blktype(self): + tarinfo = self.tar.getmember("ustar/blktype") + self._test_member(tarinfo, size=0, devmajor=3, devminor=0) - def _add_testfile(self, fileobj=None): - tar = tarfile.open(self.tarname, "a", fileobj=fileobj) - tar.addfile(tarfile.TarInfo("bar")) - tar.close() + def test_find_chrtype(self): + tarinfo = self.tar.getmember("ustar/chrtype") + self._test_member(tarinfo, size=0, devmajor=1, devminor=3) - def _create_testtar(self): - src = tarfile.open(tarname()) - t = src.getmember("0-REGTYPE") - t.name = "foo" - f = src.extractfile(t) - tar = tarfile.open(self.tarname, "w") - tar.addfile(t, f) - tar.close() + def test_find_fifotype(self): + tarinfo = self.tar.getmember("ustar/fifotype") + self._test_member(tarinfo, size=0) - def _test(self, names=["bar"], fileobj=None): - tar = tarfile.open(self.tarname, fileobj=fileobj) - self.assert_(tar.getnames() == names) + def test_find_sparse(self): + tarinfo = self.tar.getmember("ustar/sparse") + self._test_member(tarinfo, size=86016, chksum=md5_sparse) - def test_non_existing(self): - self._add_testfile() - self._test() + def test_find_umlauts(self): + tarinfo = self.tar.getmember("ustar/umlauts-ÄÖÜäöüß") + self._test_member(tarinfo, size=7011, chksum=md5_regtype) - def test_empty(self): - open(self.tarname, "wb").close() - self._add_testfile() - self._test() + def test_find_ustar_longname(self): + name = "ustar/" + "12345/" * 39 + "1234567/longname" + self.assert_(name in self.tar.getnames()) - def test_empty_fileobj(self): - fobj = StringIO.StringIO() - self._add_testfile(fobj) - fobj.seek(0) - self._test(fileobj=fobj) + def test_find_regtype_oldv7(self): + tarinfo = self.tar.getmember("misc/regtype-old-v7") + self._test_member(tarinfo, size=7011, chksum=md5_regtype) - def test_fileobj(self): - self._create_testtar() - data = open(self.tarname, "rb").read() - fobj = StringIO.StringIO(data) - self._add_testfile(fobj) - fobj.seek(0) - self._test(names=["foo", "bar"], fileobj=fobj) + def test_find_pax_umlauts(self): + tarinfo = self.tar.getmember("pax/umlauts-ÄÖÜäöüß") + self._test_member(tarinfo, size=7011, chksum=md5_regtype) - def test_existing(self): - self._create_testtar() - self._add_testfile() - self._test(names=["foo", "bar"]) +class LongnameTest(ReadTest): -class Write100Test(BaseTest): - # The name field in a tar header stores strings of at most 100 chars. - # If a string is shorter than 100 chars it has to be padded with '\0', - # which implies that a string of exactly 100 chars is stored without - # a trailing '\0'. + def test_read_longname(self): + # Test reading of longname (bug #1471427). + name = self.subdir + "/" + "123/" * 125 + "longname" + try: + tarinfo = self.tar.getmember(name) + except KeyError: + self.fail("longname not found") + self.assert_(tarinfo.type != tarfile.DIRTYPE, "read longname as dirtype") - def setUp(self): - self.name = "01234567890123456789012345678901234567890123456789" - self.name += "01234567890123456789012345678901234567890123456789" + def test_read_longlink(self): + longname = self.subdir + "/" + "123/" * 125 + "longname" + longlink = self.subdir + "/" + "123/" * 125 + "longlink" + try: + tarinfo = self.tar.getmember(longlink) + except KeyError: + self.fail("longlink not found") + self.assert_(tarinfo.linkname == longname, "linkname wrong") - self.tar = tarfile.open(tmpname(), "w") - t = tarfile.TarInfo(self.name) - self.tar.addfile(t) - self.tar.close() + def test_truncated_longname(self): + longname = self.subdir + "/" + "123/" * 125 + "longname" + tarinfo = self.tar.getmember(longname) + offset = tarinfo.offset + self.tar.fileobj.seek(offset) + fobj = StringIO.StringIO(self.tar.fileobj.read(1536)) + self.assertRaises(tarfile.ReadError, tarfile.open, name="foo.tar", fileobj=fobj) - self.tar = tarfile.open(tmpname()) - def tearDown(self): - self.tar.close() +class GNUReadTest(LongnameTest): - def test(self): - self.assertEqual(self.tar.getnames()[0], self.name, - "failed to store 100 char filename") + subdir = "gnu" + def test_sparse_file(self): + tarinfo1 = self.tar.getmember("ustar/sparse") + fobj1 = self.tar.extractfile(tarinfo1) + tarinfo2 = self.tar.getmember("gnu/sparse") + fobj2 = self.tar.extractfile(tarinfo2) + self.assert_(fobj1.read() == fobj2.read(), + "sparse file extraction failed") -class WriteSize0Test(BaseTest): - mode = 'w' - def setUp(self): - self.tmpdir = dirname() - self.dstname = tmpname() - self.dst = tarfile.open(self.dstname, "w") +class PaxReadTest(ReadTest): - def tearDown(self): - self.dst.close() + subdir = "pax" + + def test_pax_globheaders(self): + tar = tarfile.open(tarname) + tarinfo = tar.getmember("pax/regtype1") + self.assertEqual(tarinfo.uname, "foo") + self.assertEqual(tarinfo.gname, "bar") + self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), "ÄÖÜäöüß") + + tarinfo = tar.getmember("pax/regtype2") + self.assertEqual(tarinfo.uname, "") + self.assertEqual(tarinfo.gname, "bar") + self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), "ÄÖÜäöüß") + + tarinfo = tar.getmember("pax/regtype3") + self.assertEqual(tarinfo.uname, "tarfile") + self.assertEqual(tarinfo.gname, "tarfile") + self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), "ÄÖÜäöüß") + + +class WriteTest(unittest.TestCase): + + mode = "w:" + + def test_100_char_name(self): + # The name field in a tar header stores strings of at most 100 chars. + # If a string is shorter than 100 chars it has to be padded with '\0', + # which implies that a string of exactly 100 chars is stored without + # a trailing '\0'. + name = "0123456789" * 10 + tar = tarfile.open(tmpname, self.mode) + t = tarfile.TarInfo(name) + tar.addfile(t) + tar.close() + + tar = tarfile.open(tmpname) + self.assert_(tar.getnames()[0] == name, + "failed to store 100 char filename") + tar.close() + + def test_tar_size(self): + # Test for bug #1013882. + tar = tarfile.open(tmpname, self.mode) + path = os.path.join(TEMPDIR, "file") + fobj = open(path, "wb") + fobj.write("aaa") + fobj.close() + tar.add(path) + tar.close() + self.assert_(os.path.getsize(tmpname) > 0, + "tarfile is empty") + + # The test_*_size tests test for bug #1167128. + def test_file_size(self): + tar = tarfile.open(tmpname, self.mode) - def test_file(self): - path = os.path.join(self.tmpdir, "file") - f = open(path, "w") - f.close() - tarinfo = self.dst.gettarinfo(path) + path = os.path.join(TEMPDIR, "file") + fobj = open(path, "wb") + fobj.close() + tarinfo = tar.gettarinfo(path) self.assertEqual(tarinfo.size, 0) - f = open(path, "w") - f.write("aaa") - f.close() - tarinfo = self.dst.gettarinfo(path) + + fobj = open(path, "wb") + fobj.write("aaa") + fobj.close() + tarinfo = tar.gettarinfo(path) self.assertEqual(tarinfo.size, 3) - def test_directory(self): - path = os.path.join(self.tmpdir, "directory") - if os.path.exists(path): - # This shouldn't be necessary, but is <wink> if a previous - # run was killed in mid-stream. - shutil.rmtree(path) - os.mkdir(path) - tarinfo = self.dst.gettarinfo(path) - self.assertEqual(tarinfo.size, 0) + tar.close() - def test_symlink(self): + def test_directory_size(self): + path = os.path.join(TEMPDIR, "directory") + os.mkdir(path) + try: + tar = tarfile.open(tmpname, self.mode) + tarinfo = tar.gettarinfo(path) + self.assertEqual(tarinfo.size, 0) + finally: + os.rmdir(path) + + def test_link_size(self): + if hasattr(os, "link"): + link = os.path.join(TEMPDIR, "link") + target = os.path.join(TEMPDIR, "link_target") + open(target, "wb").close() + os.link(target, link) + try: + tar = tarfile.open(tmpname, self.mode) + tarinfo = tar.gettarinfo(link) + self.assertEqual(tarinfo.size, 0) + finally: + os.remove(target) + os.remove(link) + + def test_symlink_size(self): if hasattr(os, "symlink"): - path = os.path.join(self.tmpdir, "symlink") + path = os.path.join(TEMPDIR, "symlink") os.symlink("link_target", path) - tarinfo = self.dst.gettarinfo(path) - self.assertEqual(tarinfo.size, 0) + try: + tar = tarfile.open(tmpname, self.mode) + tarinfo = tar.gettarinfo(path) + self.assertEqual(tarinfo.size, 0) + finally: + os.remove(path) + def test_add_self(self): + # Test for #1257255. + dstname = os.path.abspath(tmpname) -class WriteStreamTest(WriteTest): - sep = '|' + tar = tarfile.open(tmpname, self.mode) + self.assert_(tar.name == dstname, "archive name must be absolute") - def test_padding(self): - self.dst.close() + tar.add(dstname) + self.assert_(tar.getnames() == [], "added the archive to itself") - if self.comp == "gz": - f = gzip.GzipFile(self.dstname) - s = f.read() - f.close() - elif self.comp == "bz2": - f = bz2.BZ2Decompressor() - s = file(self.dstname).read() - s = f.decompress(s) - self.assertEqual(len(f.unused_data), 0, "trailing data") - else: - f = file(self.dstname) - s = f.read() - f.close() + cwd = os.getcwd() + os.chdir(TEMPDIR) + tar.add(dstname) + os.chdir(cwd) + self.assert_(tar.getnames() == [], "added the archive to itself") - self.assertEqual(s.count("\0"), tarfile.RECORDSIZE, - "incorrect zero padding") +class StreamWriteTest(unittest.TestCase): -class WriteGNULongTest(unittest.TestCase): - """This testcase checks for correct creation of GNU Longname - and Longlink extensions. + mode = "w|" - It creates a tarfile and adds empty members with either - long names, long linknames or both and compares the size - of the tarfile with the expected size. + def test_stream_padding(self): + # Test for bug #1543303. + tar = tarfile.open(tmpname, self.mode) + tar.close() - It checks for SF bug #812325 in TarFile._create_gnulong(). + if self.mode.endswith("gz"): + fobj = gzip.GzipFile(tmpname) + data = fobj.read() + fobj.close() + elif self.mode.endswith("bz2"): + dec = bz2.BZ2Decompressor() + data = open(tmpname, "rb").read() + data = dec.decompress(data) + self.assert_(len(dec.unused_data) == 0, + "found trailing data") + else: + fobj = open(tmpname, "rb") + data = fobj.read() + fobj.close() + + self.assert_(data.count("\0") == tarfile.RECORDSIZE, + "incorrect zero padding") - While I was writing this testcase, I noticed a second bug - in the same method: - Long{names,links} weren't null-terminated which lead to - bad tarfiles when their length was a multiple of 512. This - is tested as well. - """ + +class GNUWriteTest(unittest.TestCase): + # This testcase checks for correct creation of GNU Longname + # and Longlink extended headers (cp. bug #812325). def _length(self, s): blocks, remainder = divmod(len(s) + 1, 512) @@ -474,19 +572,17 @@ class WriteGNULongTest(unittest.TestCase): return blocks * 512 def _calc_size(self, name, link=None): - # initial tar header + # Initial tar header count = 512 if len(name) > tarfile.LENGTH_NAME: - # gnu longname extended header + longname + # GNU longname extended header + longname count += 512 count += self._length(name) - if link is not None and len(link) > tarfile.LENGTH_LINK: - # gnu longlink extended header + longlink + # GNU longlink extended header + longlink count += 512 count += self._length(link) - return count def _test(self, name, link=None): @@ -495,17 +591,17 @@ class WriteGNULongTest(unittest.TestCase): tarinfo.linkname = link tarinfo.type = tarfile.LNKTYPE - tar = tarfile.open(tmpname(), "w") - tar.posix = False + tar = tarfile.open(tmpname, "w") + tar.format = tarfile.GNU_FORMAT tar.addfile(tarinfo) v1 = self._calc_size(name, link) v2 = tar.offset - self.assertEqual(v1, v2, "GNU longname/longlink creation failed") + self.assert_(v1 == v2, "GNU longname/longlink creation failed") tar.close() - tar = tarfile.open(tmpname()) + tar = tarfile.open(tmpname) member = tar.next() self.failIf(member is None, "unable to read longname member") self.assert_(tarinfo.name == member.name and \ @@ -542,268 +638,352 @@ class WriteGNULongTest(unittest.TestCase): self._test(("longnam/" * 127) + "longname_", ("longlnk/" * 127) + "longlink_") -class ReadGNULongTest(unittest.TestCase): + +class HardlinkTest(unittest.TestCase): + # Test the creation of LNKTYPE (hardlink) members in an archive. def setUp(self): - self.tar = tarfile.open(tarname()) + self.foo = os.path.join(TEMPDIR, "foo") + self.bar = os.path.join(TEMPDIR, "bar") + + fobj = open(self.foo, "wb") + fobj.write("foo") + fobj.close() + + os.link(self.foo, self.bar) + + self.tar = tarfile.open(tmpname, "w") + self.tar.add(self.foo) def tearDown(self): - self.tar.close() + os.remove(self.foo) + os.remove(self.bar) - def test_1471427(self): - """Test reading of longname (bug #1471427). - """ - name = "test/" * 20 + "0-REGTYPE" - try: - tarinfo = self.tar.getmember(name) - except KeyError: - tarinfo = None - self.assert_(tarinfo is not None, "longname not found") - self.assert_(tarinfo.type != tarfile.DIRTYPE, "read longname as dirtype") + def test_add_twice(self): + # The same name will be added as a REGTYPE every + # time regardless of st_nlink. + tarinfo = self.tar.gettarinfo(self.foo) + self.assert_(tarinfo.type == tarfile.REGTYPE, + "add file as regular failed") - def test_read_name(self): - name = ("0-LONGNAME-" * 10)[:101] - try: - tarinfo = self.tar.getmember(name) - except KeyError: - tarinfo = None - self.assert_(tarinfo is not None, "longname not found") + def test_add_hardlink(self): + tarinfo = self.tar.gettarinfo(self.bar) + self.assert_(tarinfo.type == tarfile.LNKTYPE, + "add file as hardlink failed") - def test_read_link(self): - link = ("1-LONGLINK-" * 10)[:101] - name = ("0-LONGNAME-" * 10)[:101] - try: - tarinfo = self.tar.getmember(link) - except KeyError: - tarinfo = None - self.assert_(tarinfo is not None, "longlink not found") - self.assert_(tarinfo.linkname == name, "linkname wrong") + def test_dereference_hardlink(self): + self.tar.dereference = True + tarinfo = self.tar.gettarinfo(self.bar) + self.assert_(tarinfo.type == tarfile.REGTYPE, + "dereferencing hardlink failed") - def test_truncated_longname(self): - f = open(tarname()) - fobj = StringIO.StringIO(f.read(1024)) - f.close() - tar = tarfile.open(name="foo.tar", fileobj=fobj) - self.assert_(len(tar.getmembers()) == 0, "") + +class PaxWriteTest(GNUWriteTest): + + def _test(self, name, link=None): + # See GNUWriteTest. + tarinfo = tarfile.TarInfo(name) + if link: + tarinfo.linkname = link + tarinfo.type = tarfile.LNKTYPE + + tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT) + tar.addfile(tarinfo) tar.close() + tar = tarfile.open(tmpname) + if link: + l = tar.getmembers()[0].linkname + self.assert_(link == l, "PAX longlink creation failed") + else: + n = tar.getmembers()[0].name + self.assert_(name == n, "PAX longname creation failed") -class ExtractHardlinkTest(BaseTest): + def test_iso8859_15_filename(self): + self._test_unicode_filename("iso8859-15") - def test_hardlink(self): - """Test hardlink extraction (bug #857297) - """ - # Prevent errors from being caught - self.tar.errorlevel = 1 + def test_utf8_filename(self): + self._test_unicode_filename("utf8") - self.tar.extract("0-REGTYPE", dirname()) - try: - # Extract 1-LNKTYPE which is a hardlink to 0-REGTYPE - self.tar.extract("1-LNKTYPE", dirname()) - except EnvironmentError, e: - import errno - if e.errno == errno.ENOENT: - self.fail("hardlink not extracted properly") + def test_utf16_filename(self): + self._test_unicode_filename("utf16") -class CreateHardlinkTest(BaseTest): - """Test the creation of LNKTYPE (hardlink) members in an archive. - In this respect tarfile.py mimics the behaviour of GNU tar: If - a file has a st_nlink > 1, it will be added a REGTYPE member - only the first time. - """ + def _test_unicode_filename(self, encoding): + tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT) + name = u"\u20ac".encode(encoding) # Euro sign + tar.encoding = encoding + tar.addfile(tarfile.TarInfo(name)) + tar.close() - def setUp(self): - self.tar = tarfile.open(tmpname(), "w") + tar = tarfile.open(tmpname, encoding=encoding) + self.assertEqual(tar.getmembers()[0].name, name) + tar.close() - self.foo = os.path.join(dirname(), "foo") - self.bar = os.path.join(dirname(), "bar") + def test_unicode_filename_error(self): + # The euro sign filename cannot be translated to iso8859-1 encoding. + tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT, encoding="utf8") + name = u"\u20ac".encode("utf8") # Euro sign + tar.addfile(tarfile.TarInfo(name)) + tar.close() - if os.path.exists(self.foo): - os.remove(self.foo) - if os.path.exists(self.bar): - os.remove(self.bar) + self.assertRaises(UnicodeError, tarfile.open, tmpname, encoding="iso8859-1") - f = open(self.foo, "w") - f.write("foo") - f.close() - self.tar.add(self.foo) + def test_pax_headers(self): + self._test_pax_headers({"foo": "bar", "uid": 0, "mtime": 1.23}) - def test_add_twice(self): - # If st_nlink == 1 then the same file will be added as - # REGTYPE every time. - tarinfo = self.tar.gettarinfo(self.foo) - self.assertEqual(tarinfo.type, tarfile.REGTYPE, - "add file as regular failed") + self._test_pax_headers({"euro": u"\u20ac".encode("utf8")}) - def test_add_hardlink(self): - # If st_nlink > 1 then the same file will be added as - # LNKTYPE. - os.link(self.foo, self.bar) - tarinfo = self.tar.gettarinfo(self.foo) - self.assertEqual(tarinfo.type, tarfile.LNKTYPE, - "add file as hardlink failed") + self._test_pax_headers({"euro": u"\u20ac"}, + {"euro": u"\u20ac".encode("utf8")}) - tarinfo = self.tar.gettarinfo(self.bar) - self.assertEqual(tarinfo.type, tarfile.LNKTYPE, - "add file as hardlink failed") + self._test_pax_headers({u"\u20ac": "euro"}, + {u"\u20ac".encode("utf8"): "euro"}) - def test_dereference_hardlink(self): - self.tar.dereference = True - os.link(self.foo, self.bar) - tarinfo = self.tar.gettarinfo(self.bar) - self.assertEqual(tarinfo.type, tarfile.REGTYPE, - "dereferencing hardlink failed") + def _test_pax_headers(self, pax_headers, cmp_headers=None): + if cmp_headers is None: + cmp_headers = pax_headers + tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT, \ + pax_headers=pax_headers, encoding="utf8") + tar.addfile(tarfile.TarInfo("test")) + tar.close() -# Gzip TestCases -class ReadTestGzip(ReadTest): - comp = "gz" -class ReadStreamTestGzip(ReadStreamTest): - comp = "gz" -class WriteTestGzip(WriteTest): - comp = "gz" -class WriteStreamTestGzip(WriteStreamTest): - comp = "gz" -class ReadDetectTestGzip(ReadDetectTest): - comp = "gz" -class ReadDetectFileobjTestGzip(ReadDetectFileobjTest): - comp = "gz" -class ReadAsteriskTestGzip(ReadAsteriskTest): - comp = "gz" -class ReadStreamAsteriskTestGzip(ReadStreamAsteriskTest): - comp = "gz" - -# Filemode test cases - -class FileModeTest(unittest.TestCase): - def test_modes(self): - self.assertEqual(tarfile.filemode(0755), '-rwxr-xr-x') - self.assertEqual(tarfile.filemode(07111), '---s--s--t') - -class HeaderErrorTest(unittest.TestCase): + tar = tarfile.open(tmpname, encoding="utf8") + self.assertEqual(tar.pax_headers, cmp_headers) def test_truncated_header(self): - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, "") - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, "filename\0") - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, "\0" * 511) - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, "\0" * 513) - - def test_empty_header(self): - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, "\0" * 512) - - def test_invalid_header(self): - buf = tarfile.TarInfo("filename").tobuf() - buf = buf[:148] + "foo\0\0\0\0\0" + buf[156:] # invalid number field. - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, buf) - - def test_bad_checksum(self): - buf = tarfile.TarInfo("filename").tobuf() - b = buf[:148] + " " + buf[156:] # clear the checksum field. - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, b) - b = "a" + buf[1:] # manipulate the buffer, so checksum won't match. - self.assertRaises(tarfile.HeaderError, tarfile.TarInfo.frombuf, b) - -class OpenFileobjTest(BaseTest): - # Test for SF bug #1496501. - - def test_opener(self): - fobj = StringIO.StringIO("foo\n") - try: - tarfile.open("", "r", fileobj=fobj) - except tarfile.ReadError: - self.assertEqual(fobj.tell(), 0, "fileobj's position has moved") - -if bz2: - # Bzip2 TestCases - class ReadTestBzip2(ReadTestGzip): - comp = "bz2" - class ReadStreamTestBzip2(ReadStreamTestGzip): - comp = "bz2" - class WriteTestBzip2(WriteTest): - comp = "bz2" - class WriteStreamTestBzip2(WriteStreamTestGzip): - comp = "bz2" - class ReadDetectTestBzip2(ReadDetectTest): - comp = "bz2" - class ReadDetectFileobjTestBzip2(ReadDetectFileobjTest): - comp = "bz2" - class ReadAsteriskTestBzip2(ReadAsteriskTest): - comp = "bz2" - class ReadStreamAsteriskTestBzip2(ReadStreamAsteriskTest): - comp = "bz2" - -# If importing gzip failed, discard the Gzip TestCases. -if not gzip: - del ReadTestGzip - del ReadStreamTestGzip - del WriteTestGzip - del WriteStreamTestGzip + tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT) + tarinfo = tarfile.TarInfo("123/" * 126 + "longname") + tar.addfile(tarinfo) + tar.close() -def test_main(): - # Create archive. - f = open(tarname(), "rb") - fguts = f.read() - f.close() - if gzip: - # create testtar.tar.gz - tar = gzip.open(tarname("gz"), "wb") - tar.write(fguts) + # Simulate a premature EOF. + open(tmpname, "rb+").truncate(1536) + tar = tarfile.open(tmpname) + self.assertEqual(tar.getmembers(), []) + + +class AppendTest(unittest.TestCase): + # Test append mode (cp. patch #1652681). + + def setUp(self): + self.tarname = tmpname + if os.path.exists(self.tarname): + os.remove(self.tarname) + + def _add_testfile(self, fileobj=None): + tar = tarfile.open(self.tarname, "a", fileobj=fileobj) + tar.addfile(tarfile.TarInfo("bar")) tar.close() - if bz2: - # create testtar.tar.bz2 - tar = bz2.BZ2File(tarname("bz2"), "wb") - tar.write(fguts) + + def _create_testtar(self, mode="w:"): + src = tarfile.open(tarname) + t = src.getmember("ustar/regtype") + t.name = "foo" + f = src.extractfile(t) + tar = tarfile.open(self.tarname, mode) + tar.addfile(t, f) tar.close() + def _test(self, names=["bar"], fileobj=None): + tar = tarfile.open(self.tarname, fileobj=fileobj) + self.assertEqual(tar.getnames(), names) + + def test_non_existing(self): + self._add_testfile() + self._test() + + def test_empty(self): + open(self.tarname, "w").close() + self._add_testfile() + self._test() + + def test_empty_fileobj(self): + fobj = StringIO.StringIO() + self._add_testfile(fobj) + fobj.seek(0) + self._test(fileobj=fobj) + + def test_fileobj(self): + self._create_testtar() + data = open(self.tarname).read() + fobj = StringIO.StringIO(data) + self._add_testfile(fobj) + fobj.seek(0) + self._test(names=["foo", "bar"], fileobj=fobj) + + def test_existing(self): + self._create_testtar() + self._add_testfile() + self._test(names=["foo", "bar"]) + + def test_append_gz(self): + if gzip is None: + return + self._create_testtar("w:gz") + self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, "a") + + def test_append_bz2(self): + if bz2 is None: + return + self._create_testtar("w:bz2") + self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, "a") + + +class LimitsTest(unittest.TestCase): + + def test_ustar_limits(self): + # 100 char name + tarinfo = tarfile.TarInfo("0123456789" * 10) + tarinfo.create_ustar_header() + + # 101 char name that cannot be stored + tarinfo = tarfile.TarInfo("0123456789" * 10 + "0") + self.assertRaises(ValueError, tarinfo.create_ustar_header) + + # 256 char name with a slash at pos 156 + tarinfo = tarfile.TarInfo("123/" * 62 + "longname") + tarinfo.create_ustar_header() + + # 256 char name that cannot be stored + tarinfo = tarfile.TarInfo("1234567/" * 31 + "longname") + self.assertRaises(ValueError, tarinfo.create_ustar_header) + + # 512 char name + tarinfo = tarfile.TarInfo("123/" * 126 + "longname") + self.assertRaises(ValueError, tarinfo.create_ustar_header) + + # 512 char linkname + tarinfo = tarfile.TarInfo("longlink") + tarinfo.linkname = "123/" * 126 + "longname" + self.assertRaises(ValueError, tarinfo.create_ustar_header) + + # uid > 8 digits + tarinfo = tarfile.TarInfo("name") + tarinfo.uid = 010000000 + self.assertRaises(ValueError, tarinfo.create_ustar_header) + + def test_gnu_limits(self): + tarinfo = tarfile.TarInfo("123/" * 126 + "longname") + tarinfo.create_gnu_header() + + tarinfo = tarfile.TarInfo("longlink") + tarinfo.linkname = "123/" * 126 + "longname" + tarinfo.create_gnu_header() + + # uid >= 256 ** 7 + tarinfo = tarfile.TarInfo("name") + tarinfo.uid = 04000000000000000000L + self.assertRaises(ValueError, tarinfo.create_gnu_header) + + def test_pax_limits(self): + # A 256 char name that can be stored without an extended header. + tarinfo = tarfile.TarInfo("123/" * 62 + "longname") + self.assert_(len(tarinfo.create_pax_header("utf8")) == 512, + "create_pax_header attached superfluous extended header") + + tarinfo = tarfile.TarInfo("123/" * 126 + "longname") + tarinfo.create_pax_header("utf8") + + tarinfo = tarfile.TarInfo("longlink") + tarinfo.linkname = "123/" * 126 + "longname" + tarinfo.create_pax_header("utf8") + + tarinfo = tarfile.TarInfo("name") + tarinfo.uid = 04000000000000000000L + tarinfo.create_pax_header("utf8") + + +class GzipMiscReadTest(MiscReadTest): + tarname = gzipname + mode = "r:gz" +class GzipUstarReadTest(UstarReadTest): + tarname = gzipname + mode = "r:gz" +class GzipStreamReadTest(StreamReadTest): + tarname = gzipname + mode = "r|gz" +class GzipWriteTest(WriteTest): + mode = "w:gz" +class GzipStreamWriteTest(StreamWriteTest): + mode = "w|gz" + + +class Bz2MiscReadTest(MiscReadTest): + tarname = bz2name + mode = "r:bz2" +class Bz2UstarReadTest(UstarReadTest): + tarname = bz2name + mode = "r:bz2" +class Bz2StreamReadTest(StreamReadTest): + tarname = bz2name + mode = "r|bz2" +class Bz2WriteTest(WriteTest): + mode = "w:bz2" +class Bz2StreamWriteTest(StreamWriteTest): + mode = "w|bz2" + +def test_main(): + if not os.path.exists(TEMPDIR): + os.mkdir(TEMPDIR) + tests = [ - FileModeTest, - HeaderErrorTest, - OpenFileobjTest, - ReadTest, - ReadStreamTest, - ReadDetectTest, - ReadDetectFileobjTest, - ReadAsteriskTest, - ReadStreamAsteriskTest, + UstarReadTest, + MiscReadTest, + StreamReadTest, + DetectReadTest, + MemberReadTest, + GNUReadTest, + PaxReadTest, WriteTest, + StreamWriteTest, + GNUWriteTest, + PaxWriteTest, AppendTest, - Write100Test, - WriteSize0Test, - WriteStreamTest, - WriteGNULongTest, - ReadGNULongTest, + LimitsTest, ] if hasattr(os, "link"): - tests.append(ExtractHardlinkTest) - tests.append(CreateHardlinkTest) + tests.append(HardlinkTest) + + fobj = open(tarname, "rb") + data = fobj.read() + fobj.close() if gzip: - tests.extend([ - ReadTestGzip, ReadStreamTestGzip, - WriteTestGzip, WriteStreamTestGzip, - ReadDetectTestGzip, ReadDetectFileobjTestGzip, - ReadAsteriskTestGzip, ReadStreamAsteriskTestGzip - ]) + # Create testtar.tar.gz and add gzip-specific tests. + tar = gzip.open(gzipname, "wb") + tar.write(data) + tar.close() + + tests += [ + GzipMiscReadTest, + GzipUstarReadTest, + GzipStreamReadTest, + GzipWriteTest, + GzipStreamWriteTest, + ] if bz2: - tests.extend([ - ReadTestBzip2, ReadStreamTestBzip2, - WriteTestBzip2, WriteStreamTestBzip2, - ReadDetectTestBzip2, ReadDetectFileobjTestBzip2, - ReadAsteriskTestBzip2, ReadStreamAsteriskTestBzip2 - ]) + # Create testtar.tar.bz2 and add bz2-specific tests. + tar = bz2.BZ2File(bz2name, "wb") + tar.write(data) + tar.close() + + tests += [ + Bz2MiscReadTest, + Bz2UstarReadTest, + Bz2StreamReadTest, + Bz2WriteTest, + Bz2StreamWriteTest, + ] + try: test_support.run_unittest(*tests) finally: - if gzip: - os.remove(tarname("gz")) - if bz2: - os.remove(tarname("bz2")) - if os.path.exists(dirname()): - shutil.rmtree(dirname()) - if os.path.exists(tmpname()): - os.remove(tmpname()) + if os.path.exists(TEMPDIR): + shutil.rmtree(TEMPDIR) if __name__ == "__main__": test_main() + diff --git a/Lib/test/testtar.tar b/Lib/test/testtar.tar Binary files differindex 1f4493f..c4c82b8 100644 --- a/Lib/test/testtar.tar +++ b/Lib/test/testtar.tar @@ -168,6 +168,9 @@ Core and builtins Library ------- +- Added support for the POSIX.1-2001 (pax) format to tarfile.py. Extended + and cleaned up the test suite. Added a new testtar.tar. + - Patch #1449244: Support Unicode strings in email.message.Message.{set_charset,get_content_charset}. |