Kaydet (Commit) c64e4021 authored tarafından Lars Gustäbel's avatar Lars Gustäbel

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