[Python-checkins] r54335 - in python/trunk: Doc/lib/libtarfile.tex Lib/tarfile.py Lib/test/test_tarfile.py Lib/test/testtar.tar Misc/NEWS
lars.gustaebel
python-checkins at python.org
Tue Mar 13 11:47:29 CET 2007
Author: lars.gustaebel
Date: Tue Mar 13 11:47:19 2007
New Revision: 54335
Modified:
python/trunk/Doc/lib/libtarfile.tex
python/trunk/Lib/tarfile.py
python/trunk/Lib/test/test_tarfile.py
python/trunk/Lib/test/testtar.tar
python/trunk/Misc/NEWS
Log:
This is the implementation of POSIX.1-2001 (pax) format read/write
support.
The TarInfo class now contains all necessary logic to process and
create tar header data which has been moved there from the TarFile
class. The fromtarfile() method was added. The new path and linkpath
properties are aliases for the name and linkname attributes in
correspondence to the pax naming scheme.
The TarFile constructor and classmethods now accept a number of
keyword arguments which could only be set as attributes before (e.g.
dereference, ignore_zeros). The encoding and pax_headers arguments
were added for pax support. There is a new tarinfo keyword argument
that allows using subclassed TarInfo objects in TarFile.
The boolean TarFile.posix attribute is deprecated, because now three
tar formats are supported. Instead, the desired format for writing is
specified using the constants USTAR_FORMAT, GNU_FORMAT and PAX_FORMAT
as the format keyword argument. This change affects TarInfo.tobuf()
as well.
The test suite has been heavily reorganized and partially rewritten.
A new testtar.tar was added that contains sample data in many formats
from 4 different tar programs.
Some bugs and quirks that also have been fixed:
Directory names do no longer have a trailing slash in TarInfo.name or
TarFile.getnames().
Adding the same file twice does not create a hardlink file member.
The TarFile constructor does no longer need a name argument.
The TarFile._mode attribute was renamed to mode and contains either
'r', 'w' or 'a'.
Modified: python/trunk/Doc/lib/libtarfile.tex
==============================================================================
--- python/trunk/Doc/lib/libtarfile.tex (original)
+++ python/trunk/Doc/lib/libtarfile.tex Tue Mar 13 11:47:19 2007
@@ -12,21 +12,24 @@
\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 @@
\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 @@
\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 @@
\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 @@
\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 @@
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:
Modified: python/trunk/Lib/tarfile.py
==============================================================================
--- python/trunk/Lib/tarfile.py (original)
+++ python/trunk/Lib/tarfile.py Tue Mar 13 11:47:19 2007
@@ -33,7 +33,7 @@
__version__ = "$Revision$"
# $Source$
-version = "0.8.0"
+version = "0.9.0"
__author__ = "Lars Gustäbel (lars at gustaebel.de)"
__date__ = "$Date$"
__cvsid__ = "$Id$"
@@ -50,6 +50,7 @@
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 @@
#---------------------------------------------------------
# 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 @@
TOEXEC = 0001 # execute/search by other
#---------------------------------------------------------
+# initialization
+#---------------------------------------------------------
+ENCODING = sys.getfilesystemencoding()
+if ENCODING is None:
+ ENCODING = "ascii"
+
+#---------------------------------------------------------
# Some useful functions
#---------------------------------------------------------
@@ -139,6 +165,15 @@
"""
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 @@
# 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 @@
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 @@
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 @@
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 @@
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 @@
"""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 @@
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 @@
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)
+ 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])
- if prefix and not tarinfo.issparse():
- tarinfo.name = prefix + "/" + tarinfo.name
+ # 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("/")
+
+ # Reconstruct a ustar longname.
+ if prefix and obj.type not in GNU_TYPES:
+ obj.name = prefix + "/" + obj.name
+ return obj
- def tobuf(self, posix=False):
- """Return a tar header as a string of 512 byte blocks.
+ @classmethod
+ def fromtarfile(cls, tarfile):
+ """Return the next TarInfo object from TarFile object
+ tarfile.
"""
- buf = ""
- type = self.type
- prefix = ""
+ 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 self.name.endswith("/"):
- type = DIRTYPE
+ #--------------------------------------------------------------------------
+ # 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:
+ return self._proc_builtin(tarfile)
+
+ 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
- if type in (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK):
- # Prevent "././@LongLink" from being normalized.
- name = self.name
- else:
- name = normpath(self.name)
+ # 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 type == DIRTYPE:
- # directories should end with '/'
- name += "/"
+ return self
- linkname = self.linkname
- if linkname:
- # if linkname is empty we end up with a '.'
- linkname = normpath(linkname)
+ 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 posix:
- if self.size > MAXSIZE_MEMBER:
- raise ValueError("file is too large (>= 8 GB)")
+ # 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)
- if len(self.linkname) > LENGTH_LINK:
- raise ValueError("linkname is too long (>%d)" % (LENGTH_LINK))
+ # 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 len(name) > LENGTH_NAME:
- prefix = name[:LENGTH_PREFIX + 1]
- while prefix and prefix[-1] != "/":
- prefix = prefix[:-1]
+ return next
- name = name[len(prefix):]
- prefix = prefix[:-1]
+ 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 not prefix or len(name) > LENGTH_NAME:
- raise ValueError("name is too long")
+ isextended = ord(buf[482])
+ origsize = nti(buf[483:495])
- else:
- if len(self.linkname) > LENGTH_LINK:
- buf += self._create_gnulong(self.linkname, GNUTYPE_LONGLINK)
+ # 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])
- if len(name) > LENGTH_NAME:
- buf += self._create_gnulong(name, GNUTYPE_LONGNAME)
+ if lastpos < origsize:
+ sp.append(_hole(lastpos, origsize - lastpos))
- 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)
- ]
+ self.sparse = sp
- 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
+ self.offset_data = tarfile.fileobj.tell()
+ tarfile.offset = self.offset_data + self._block(self.size)
+ self.size = origsize
- 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.
- """
- name += NUL
+ return self
- tarinfo = self.__class__()
- tarinfo.name = "././@LongLink"
- tarinfo.type = type
- tarinfo.mode = 0
- tarinfo.size = len(name)
+ def _proc_pax(self, tarfile):
+ """Process an extended or global header as described in
+ POSIX.1-2001.
+ """
+ # Read the header information.
+ buf = tarfile.fileobj.read(self._block(self.size))
+
+ # 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
- # 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
+ 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 @@
# 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.
- fileobject = ExFileObject
+ encoding = ENCODING # Transfer UTF-8 strings from POSIX.1-2001
+ # headers to this encoding.
- def __init__(self, name=None, mode="r", fileobj=None):
+ tarinfo = TarInfo # The default TarInfo class to use.
+
+ fileobject = ExFileObject # The default ExFileObject class to use.
+
+ 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 @@
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 @@
# 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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
# 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 @@
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 @@
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 @@
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 @@
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 @@
# 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 @@
"""
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 @@
"""
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 @@
# 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 @@
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 @@
"""
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.
Modified: python/trunk/Lib/test/test_tarfile.py
==============================================================================
--- python/trunk/Lib/test/test_tarfile.py (original)
+++ python/trunk/Lib/test/test_tarfile.py Tue Mar 13 11:47:19 2007
@@ -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 @@
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))
-
-def dirname():
- if not os.path.exists(tempdir):
- os.mkdir(tempdir)
- return tempdir
-
-def tmpname():
- return tempname
-
-
-class BaseTest(unittest.TestCase):
- comp = ''
- mode = 'r'
- sep = ':'
+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")
+
+md5_regtype = "65f477c818ad9e15f7feab0c6d37742f"
+md5_sparse = "a54fbc4ca4f4399a90e1b27164012fc6"
+
+
+class ReadTest(unittest.TestCase):
+
+ 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
- for tarinfo in self.tar:
- members += 1
- if not tarinfo.isreg():
- 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()
+class UstarReadTest(ReadTest):
- 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_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()
- # Create an old tar style directory entry.
- filename = tmpname()
- tarinfo = tarfile.TarInfo("directory/")
- tarinfo.type = tarfile.REGTYPE
+ tarinfo = self.tar.getmember("ustar/regtype")
+ fobj = self.tar.extractfile(tarinfo)
- fobj = open(filename, "w")
- fobj.write(tarinfo.tobuf())
+ 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()
- 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)
- self.assertRaises(tarfile.StreamError, f.read)
+class MiscReadTest(ReadTest):
- def test_stream(self):
- """Compare the normal tar and the stream tar.
- """
- stream = self.tar
- tar = tarfile.open(tarname(), 'r')
-
- while 1:
- t1 = tar.next()
- t2 = stream.next()
- if t1 is None:
- break
- self.assert_(t2 is not None, "stream.next() failed.")
+ 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")
- if t2.islnk() or t2.issym():
- self.assertRaises(tarfile.StreamError, stream.extractfile, t2)
- continue
- v1 = tar.extractfile(t1)
- v2 = stream.extractfile(t2)
- if v1 is None:
+ def test_check_members(self):
+ for tarinfo in self.tar:
+ self.assert_(int(tarinfo.mtime) == 07606136617,
+ "wrong mtime for %s" % tarinfo.name)
+ if not tarinfo.name.startswith("ustar/"):
continue
- self.assert_(v2 is not None, "stream.extractfile() failed")
- self.assert_(v1.read() == v2.read(), "stream extraction failed")
+ self.assert_(tarinfo.uname == "tarfile",
+ "wrong uname for %s" % tarinfo.name)
- tar.close()
- stream.close()
+ def test_find_members(self):
+ self.assert_(self.tar.getmembers()[-1].name == "misc/eof",
+ "could not find all members")
+
+ def test_extract_hardlink(self):
+ # Test hardlink extraction (e.g. bug #857297).
+ tar = tarfile.open(tarname, errorlevel=1)
-class ReadDetectTest(ReadTest):
+ 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")
- def setUp(self):
- self.tar = tarfile.open(tarname(self.comp), self.mode)
+ data = open(os.path.join(TEMPDIR, "ustar/lnktype"), "rb").read()
+ self.assertEqual(md5sum(data), md5_regtype)
-class ReadDetectFileobjTest(ReadTest):
+ try:
+ tar.extract("ustar/symtype", TEMPDIR)
+ except EnvironmentError, e:
+ if e.errno == errno.ENOENT:
+ self.fail("symlink not extracted properly")
- def setUp(self):
- name = tarname(self.comp)
- self.tar = tarfile.open(name, mode=self.mode,
- fileobj=open(name, "rb"))
+ data = open(os.path.join(TEMPDIR, "ustar/symtype"), "rb").read()
+ self.assertEqual(md5sum(data), md5_regtype)
-class ReadAsteriskTest(ReadTest):
- def setUp(self):
- mode = self.mode + self.sep + "*"
- self.tar = tarfile.open(tarname(self.comp), mode)
+class StreamReadTest(ReadTest):
-class ReadStreamAsteriskTest(ReadStreamTest):
+ mode="r|"
- def setUp(self):
- mode = self.mode + self.sep + "*"
- self.tar = tarfile.open(tarname(self.comp), mode)
+ 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")
-class WriteTest(BaseTest):
- mode = 'w'
+ 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 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_compare_members(self):
+ tar1 = tarfile.open(tarname)
+ tar2 = self.tar
+
+ while True:
+ t1 = tar1.next()
+ t2 = tar2.next()
+ if t1 is None:
+ break
+ self.assert_(t2 is not None, "stream.next() failed.")
- def tearDown(self):
- self.src.close()
- self.dst.close()
+ if t2.islnk() or t2.issym():
+ self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
+ continue
- def test_posix(self):
- self.dst.posix = 1
- self._test()
+ 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")
- def test_nonposix(self):
- self.dst.posix = 0
- self._test()
+ tar1.close()
- 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(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)
- def test_add_self(self):
- dstname = os.path.abspath(self.dstname)
+class DetectReadTest(unittest.TestCase):
- self.assertEqual(self.dst.name, dstname, "archive name must be absolute")
+ def _testfunc_file(self, name, mode):
+ try:
+ tarfile.open(name, mode)
+ except tarfile.ReadError:
+ self.fail()
- self.dst.add(dstname)
- self.assertEqual(self.dst.getnames(), [], "added the archive to itself")
+ def _testfunc_fileobj(self, name, mode):
+ try:
+ tarfile.open(name, mode, fileobj=open(name, "rb"))
+ except tarfile.ReadError:
+ self.fail()
- 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_modes(self, testfunc):
+ testfunc(tarname, "r")
+ testfunc(tarname, "r:")
+ testfunc(tarname, "r:*")
+ testfunc(tarname, "r|")
+ testfunc(tarname, "r|*")
+ 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|")
+
+ testfunc(gzipname, "r")
+ testfunc(gzipname, "r:*")
+ testfunc(gzipname, "r:gz")
+ testfunc(gzipname, "r|*")
+ testfunc(gzipname, "r|gz")
-class AppendTest(unittest.TestCase):
- # Test append mode (cp. patch #1652681).
+ 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|")
+
+ testfunc(bz2name, "r")
+ testfunc(bz2name, "r:*")
+ testfunc(bz2name, "r:bz2")
+ testfunc(bz2name, "r|*")
+ testfunc(bz2name, "r|bz2")
+
+ def test_detect_file(self):
+ self._test_modes(self._testfunc_file)
+
+ def test_detect_fileobj(self):
+ self._test_modes(self._testfunc_fileobj)
+
+
+class MemberReadTest(ReadTest):
+
+ 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)
+
+ 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_find_regtype(self):
+ tarinfo = self.tar.getmember("ustar/regtype")
+ self._test_member(tarinfo, size=7011, chksum=md5_regtype)
+
+ def test_find_conttype(self):
+ tarinfo = self.tar.getmember("ustar/conttype")
+ self._test_member(tarinfo, size=7011, chksum=md5_regtype)
+
+ def test_find_dirtype(self):
+ tarinfo = self.tar.getmember("ustar/dirtype")
+ self._test_member(tarinfo, size=0)
+
+ 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")
+
+ def test_find_symtype(self):
+ tarinfo = self.tar.getmember("ustar/symtype")
+ self._test_member(tarinfo, size=0, linkname="regtype")
+
+ def test_find_blktype(self):
+ tarinfo = self.tar.getmember("ustar/blktype")
+ self._test_member(tarinfo, size=0, devmajor=3, devminor=0)
+
+ def test_find_chrtype(self):
+ tarinfo = self.tar.getmember("ustar/chrtype")
+ self._test_member(tarinfo, size=0, devmajor=1, devminor=3)
+
+ def test_find_fifotype(self):
+ tarinfo = self.tar.getmember("ustar/fifotype")
+ self._test_member(tarinfo, size=0)
+
+ def test_find_sparse(self):
+ tarinfo = self.tar.getmember("ustar/sparse")
+ self._test_member(tarinfo, size=86016, chksum=md5_sparse)
+
+ def test_find_umlauts(self):
+ tarinfo = self.tar.getmember("ustar/umlauts-ÄÖÜäöüß")
+ self._test_member(tarinfo, size=7011, chksum=md5_regtype)
+
+ def test_find_ustar_longname(self):
+ name = "ustar/" + "12345/" * 39 + "1234567/longname"
+ self.assert_(name in self.tar.getnames())
+
+ 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_find_pax_umlauts(self):
+ tarinfo = self.tar.getmember("pax/umlauts-ÄÖÜäöüß")
+ self._test_member(tarinfo, size=7011, chksum=md5_regtype)
+
+
+class LongnameTest(ReadTest):
+
+ 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.tarname = tmpname()
- if os.path.exists(self.tarname):
- os.remove(self.tarname)
+ 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")
- def _add_testfile(self, fileobj=None):
- tar = tarfile.open(self.tarname, "a", fileobj=fileobj)
- tar.addfile(tarfile.TarInfo("bar"))
+ 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)
+
+
+class GNUReadTest(LongnameTest):
+
+ 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 PaxReadTest(ReadTest):
+
+ 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()
- 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 = tarfile.open(tmpname)
+ self.assert_(tar.getnames()[0] == name,
+ "failed to store 100 char filename")
tar.close()
- def _test(self, names=["bar"], fileobj=None):
- tar = tarfile.open(self.tarname, fileobj=fileobj)
- self.assert_(tar.getnames() == names)
-
- def test_non_existing(self):
- self._add_testfile()
- self._test()
-
- def test_empty(self):
- open(self.tarname, "wb").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, "rb").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"])
-
-
-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_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")
- def setUp(self):
- self.name = "01234567890123456789012345678901234567890123456789"
- self.name += "01234567890123456789012345678901234567890123456789"
+ # The test_*_size tests test for bug #1167128.
+ def test_file_size(self):
+ tar = tarfile.open(tmpname, self.mode)
- self.tar = tarfile.open(tmpname(), "w")
- t = tarfile.TarInfo(self.name)
- self.tar.addfile(t)
- self.tar.close()
+ path = os.path.join(TEMPDIR, "file")
+ fobj = open(path, "wb")
+ fobj.close()
+ tarinfo = tar.gettarinfo(path)
+ self.assertEqual(tarinfo.size, 0)
- self.tar = tarfile.open(tmpname())
+ fobj = open(path, "wb")
+ fobj.write("aaa")
+ fobj.close()
+ tarinfo = tar.gettarinfo(path)
+ self.assertEqual(tarinfo.size, 3)
- def tearDown(self):
- self.tar.close()
+ tar.close()
- def test(self):
- self.assertEqual(self.tar.getnames()[0], self.name,
- "failed to store 100 char filename")
+ 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)
-class WriteSize0Test(BaseTest):
- mode = 'w'
+ def test_symlink_size(self):
+ if hasattr(os, "symlink"):
+ path = os.path.join(TEMPDIR, "symlink")
+ os.symlink("link_target", path)
+ try:
+ tar = tarfile.open(tmpname, self.mode)
+ tarinfo = tar.gettarinfo(path)
+ self.assertEqual(tarinfo.size, 0)
+ finally:
+ os.remove(path)
- def setUp(self):
- self.tmpdir = dirname()
- self.dstname = tmpname()
- self.dst = tarfile.open(self.dstname, "w")
+ def test_add_self(self):
+ # Test for #1257255.
+ dstname = os.path.abspath(tmpname)
- def tearDown(self):
- self.dst.close()
+ tar = tarfile.open(tmpname, self.mode)
+ self.assert_(tar.name == dstname, "archive name must be absolute")
- def test_file(self):
- path = os.path.join(self.tmpdir, "file")
- f = open(path, "w")
- f.close()
- tarinfo = self.dst.gettarinfo(path)
- self.assertEqual(tarinfo.size, 0)
- f = open(path, "w")
- f.write("aaa")
- f.close()
- tarinfo = self.dst.gettarinfo(path)
- self.assertEqual(tarinfo.size, 3)
+ tar.add(dstname)
+ self.assert_(tar.getnames() == [], "added the archive to itself")
- 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)
+ cwd = os.getcwd()
+ os.chdir(TEMPDIR)
+ tar.add(dstname)
+ os.chdir(cwd)
+ self.assert_(tar.getnames() == [], "added the archive to itself")
- def test_symlink(self):
- if hasattr(os, "symlink"):
- path = os.path.join(self.tmpdir, "symlink")
- os.symlink("link_target", path)
- tarinfo = self.dst.gettarinfo(path)
- self.assertEqual(tarinfo.size, 0)
+class StreamWriteTest(unittest.TestCase):
-class WriteStreamTest(WriteTest):
- sep = '|'
+ mode = "w|"
- def test_padding(self):
- self.dst.close()
+ def test_stream_padding(self):
+ # Test for bug #1543303.
+ tar = tarfile.open(tmpname, self.mode)
+ tar.close()
- 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")
+ 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:
- f = file(self.dstname)
- s = f.read()
- f.close()
+ fobj = open(tmpname, "rb")
+ data = fobj.read()
+ fobj.close()
- self.assertEqual(s.count("\0"), tarfile.RECORDSIZE,
+ self.assert_(data.count("\0") == tarfile.RECORDSIZE,
"incorrect zero padding")
-class WriteGNULongTest(unittest.TestCase):
- """This testcase checks for correct creation of GNU Longname
- and Longlink extensions.
-
- 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.
-
- It checks for SF bug #812325 in TarFile._create_gnulong().
-
- 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 @@
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 @@
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 @@
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, "")
- tar.close()
+class PaxWriteTest(GNUWriteTest):
-class ExtractHardlinkTest(BaseTest):
+ def _test(self, name, link=None):
+ # See GNUWriteTest.
+ tarinfo = tarfile.TarInfo(name)
+ if link:
+ tarinfo.linkname = link
+ tarinfo.type = tarfile.LNKTYPE
- def test_hardlink(self):
- """Test hardlink extraction (bug #857297)
- """
- # Prevent errors from being caught
- self.tar.errorlevel = 1
+ tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT)
+ tar.addfile(tarinfo)
+ tar.close()
- 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")
+ 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 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_iso8859_15_filename(self):
+ self._test_unicode_filename("iso8859-15")
- def setUp(self):
- self.tar = tarfile.open(tmpname(), "w")
+ def test_utf8_filename(self):
+ self._test_unicode_filename("utf8")
- self.foo = os.path.join(dirname(), "foo")
- self.bar = os.path.join(dirname(), "bar")
+ def test_utf16_filename(self):
+ self._test_unicode_filename("utf16")
+
+ 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()
- if os.path.exists(self.foo):
- os.remove(self.foo)
- if os.path.exists(self.bar):
- os.remove(self.bar)
-
- f = open(self.foo, "w")
- f.write("foo")
- f.close()
- self.tar.add(self.foo)
+ tar = tarfile.open(tmpname, encoding=encoding)
+ self.assertEqual(tar.getmembers()[0].name, name)
+ tar.close()
- 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")
+ 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()
- 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.assertRaises(UnicodeError, tarfile.open, tmpname, encoding="iso8859-1")
- tarinfo = self.tar.gettarinfo(self.bar)
- self.assertEqual(tarinfo.type, tarfile.LNKTYPE,
- "add file as hardlink failed")
+ def test_pax_headers(self):
+ self._test_pax_headers({"foo": "bar", "uid": 0, "mtime": 1.23})
- 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")
+ self._test_pax_headers({"euro": u"\u20ac".encode("utf8")})
+ self._test_pax_headers({"euro": u"\u20ac"},
+ {"euro": u"\u20ac".encode("utf8")})
-# 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')
+ self._test_pax_headers({u"\u20ac": "euro"},
+ {u"\u20ac".encode("utf8"): "euro"})
-class HeaderErrorTest(unittest.TestCase):
+ 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()
+
+ 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)
+ tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT)
+ tarinfo = tarfile.TarInfo("123/" * 126 + "longname")
+ tar.addfile(tarinfo)
+ tar.close()
-class OpenFileobjTest(BaseTest):
- # Test for SF bug #1496501.
+ # Simulate a premature EOF.
+ open(tmpname, "rb+").truncate(1536)
+ tar = tarfile.open(tmpname)
+ self.assertEqual(tar.getmembers(), [])
- 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
+class AppendTest(unittest.TestCase):
+ # Test append mode (cp. patch #1652681).
-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)
+ 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()
+
Modified: python/trunk/Lib/test/testtar.tar
==============================================================================
Binary files. No diff available.
Modified: python/trunk/Misc/NEWS
==============================================================================
--- python/trunk/Misc/NEWS (original)
+++ python/trunk/Misc/NEWS Tue Mar 13 11:47:19 2007
@@ -168,6 +168,9 @@
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}.
More information about the Python-checkins
mailing list