[Jython-checkins] jython: Repatched the automatically patchable modules against CPythonLib 2.7
alex.gronholm
jython-checkins at python.org
Wed Mar 14 22:02:53 CET 2012
http://hg.python.org/jython/rev/940c47216bdb
changeset: 6372:940c47216bdb
user: Alex Grönholm <alex.gronholm at nextday.fi>
date: Wed Mar 14 14:00:06 2012 -0700
summary:
Repatched the automatically patchable modules against CPythonLib 2.7
(instead of 2.6 like before)
files:
Lib/decimal.py | 798 ++++++++-
Lib/distutils/ccompiler.py | 391 ++--
Lib/distutils/command/bdist_dumb.py | 41 +-
Lib/distutils/command/install_scripts.py | 4 +-
Lib/distutils/file_util.py | 158 +-
Lib/distutils/spawn.py | 92 +-
Lib/gettext.py | 6 +-
Lib/mailbox.py | 94 +-
Lib/netrc.py | 9 +-
Lib/robotparser.py | 7 +-
Lib/test/list_tests.py | 6 +-
Lib/test/pickletester.py | 361 ++++-
Lib/test/test_array.py | 110 +-
Lib/test/test_code.py | 45 +-
Lib/test/test_codeccallbacks.py | 64 +-
Lib/test/test_hashlib.py | 164 +-
Lib/test/test_hmac.py | 28 +-
Lib/test/test_logging.py | 894 ++++++++++-
Lib/test/test_new.py | 2 +-
Lib/test/test_os.py | 344 ++-
Lib/test/test_random.py | 68 +-
Lib/test/test_repr.py | 57 +-
Lib/test/test_robotparser.py | 51 +-
Lib/test/test_shutil.py | 360 ++++-
Lib/test/test_time.py | 72 +-
Lib/test/test_urllib2.py | 153 +-
Lib/timeit.py | 6 +-
Lib/types.py | 19 +-
Lib/warnings.py | 40 +-
Lib/weakref.py | 26 +-
Lib/zipfile.py | 446 ++--
31 files changed, 3660 insertions(+), 1256 deletions(-)
diff --git a/Lib/decimal.py b/Lib/decimal.py
--- a/Lib/decimal.py
+++ b/Lib/decimal.py
@@ -134,7 +134,10 @@
'setcontext', 'getcontext', 'localcontext'
]
+__version__ = '1.70' # Highest version of the spec this complies with
+
import copy as _copy
+import math as _math
import numbers as _numbers
try:
@@ -645,11 +648,55 @@
return self
if isinstance(value, float):
- raise TypeError("Cannot convert float to Decimal. " +
- "First convert the float to a string")
+ value = Decimal.from_float(value)
+ self._exp = value._exp
+ self._sign = value._sign
+ self._int = value._int
+ self._is_special = value._is_special
+ return self
raise TypeError("Cannot convert %r to Decimal" % value)
+ # @classmethod, but @decorator is not valid Python 2.3 syntax, so
+ # don't use it (see notes on Py2.3 compatibility at top of file)
+ def from_float(cls, f):
+ """Converts a float to a decimal number, exactly.
+
+ Note that Decimal.from_float(0.1) is not the same as Decimal('0.1').
+ Since 0.1 is not exactly representable in binary floating point, the
+ value is stored as the nearest representable value which is
+ 0x1.999999999999ap-4. The exact equivalent of the value in decimal
+ is 0.1000000000000000055511151231257827021181583404541015625.
+
+ >>> Decimal.from_float(0.1)
+ Decimal('0.1000000000000000055511151231257827021181583404541015625')
+ >>> Decimal.from_float(float('nan'))
+ Decimal('NaN')
+ >>> Decimal.from_float(float('inf'))
+ Decimal('Infinity')
+ >>> Decimal.from_float(-float('inf'))
+ Decimal('-Infinity')
+ >>> Decimal.from_float(-0.0)
+ Decimal('-0')
+
+ """
+ if isinstance(f, (int, long)): # handle integer inputs
+ return cls(f)
+ if _math.isinf(f) or _math.isnan(f): # raises TypeError if not a float
+ return cls(repr(f))
+ if _math.copysign(1.0, f) == 1.0:
+ sign = 0
+ else:
+ sign = 1
+ n, d = abs(f).as_integer_ratio()
+ k = d.bit_length() - 1
+ result = _dec_from_triple(sign, str(n*5**k), -k)
+ if cls is Decimal:
+ return result
+ else:
+ return cls(result)
+ from_float = classmethod(from_float)
+
def _isnan(self):
"""Returns whether the number is not actually one.
@@ -802,8 +849,11 @@
# subject of what should happen for a comparison involving a NaN.
# We take the following approach:
#
- # == comparisons involving a NaN always return False
- # != comparisons involving a NaN always return True
+ # == comparisons involving a quiet NaN always return False
+ # != comparisons involving a quiet NaN always return True
+ # == or != comparisons involving a signaling NaN signal
+ # InvalidOperation, and return False or True as above if the
+ # InvalidOperation is not trapped.
# <, >, <= and >= comparisons involving a (quiet or signaling)
# NaN signal InvalidOperation, and return False if the
# InvalidOperation is not trapped.
@@ -811,24 +861,24 @@
# This behavior is designed to conform as closely as possible to
# that specified by IEEE 754.
- def __eq__(self, other):
- other = _convert_other(other)
+ def __eq__(self, other, context=None):
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
- if self.is_nan() or other.is_nan():
+ if self._check_nans(other, context):
return False
return self._cmp(other) == 0
- def __ne__(self, other):
- other = _convert_other(other)
+ def __ne__(self, other, context=None):
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
- if self.is_nan() or other.is_nan():
+ if self._check_nans(other, context):
return True
return self._cmp(other) != 0
def __lt__(self, other, context=None):
- other = _convert_other(other)
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
ans = self._compare_check_nans(other, context)
@@ -837,7 +887,7 @@
return self._cmp(other) < 0
def __le__(self, other, context=None):
- other = _convert_other(other)
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
ans = self._compare_check_nans(other, context)
@@ -846,7 +896,7 @@
return self._cmp(other) <= 0
def __gt__(self, other, context=None):
- other = _convert_other(other)
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
ans = self._compare_check_nans(other, context)
@@ -855,7 +905,7 @@
return self._cmp(other) > 0
def __ge__(self, other, context=None):
- other = _convert_other(other)
+ other = _convert_other(other, allow_float=True)
if other is NotImplemented:
return other
ans = self._compare_check_nans(other, context)
@@ -889,12 +939,34 @@
# The hash of a nonspecial noninteger Decimal must depend only
# on the value of that Decimal, and not on its representation.
# For example: hash(Decimal('100E-1')) == hash(Decimal('10')).
+
+ # Equality comparisons involving signaling nans can raise an
+ # exception; since equality checks are implicitly and
+ # unpredictably used when checking set and dict membership, we
+ # prevent signaling nans from being used as set elements or
+ # dict keys by making __hash__ raise an exception.
if self._is_special:
- if self._isnan():
- raise TypeError('Cannot hash a NaN value.')
- return hash(str(self))
- if not self:
- return 0
+ if self.is_snan():
+ raise TypeError('Cannot hash a signaling NaN value.')
+ elif self.is_nan():
+ # 0 to match hash(float('nan'))
+ return 0
+ else:
+ # values chosen to match hash(float('inf')) and
+ # hash(float('-inf')).
+ if self._sign:
+ return -271828
+ else:
+ return 314159
+
+ # In Python 2.7, we're allowing comparisons (but not
+ # arithmetic operations) between floats and Decimals; so if
+ # a Decimal instance is exactly representable as a float then
+ # its hash should match that of the float.
+ self_as_float = float(self)
+ if Decimal.from_float(self_as_float) == self:
+ return hash(self_as_float)
+
if self._isinteger():
op = _WorkRep(self.to_integral_value())
# to make computation feasible for Decimals with large
@@ -2813,6 +2885,7 @@
def copy_sign(self, other):
"""Returns self with the sign of other."""
+ other = _convert_other(other, raiseit=True)
return _dec_from_triple(other._sign, self._int,
self._exp, self._is_special)
@@ -3517,27 +3590,25 @@
return (self.__class__, (str(self),))
def __copy__(self):
- if type(self) == Decimal:
+ if type(self) is Decimal:
return self # I'm immutable; therefore I am my own clone
return self.__class__(str(self))
def __deepcopy__(self, memo):
- if type(self) == Decimal:
+ if type(self) is Decimal:
return self # My components are also immutable
return self.__class__(str(self))
- # PEP 3101 support. See also _parse_format_specifier and _format_align
- def __format__(self, specifier, context=None):
+ # PEP 3101 support. the _localeconv keyword argument should be
+ # considered private: it's provided for ease of testing only.
+ def __format__(self, specifier, context=None, _localeconv=None):
"""Format a Decimal instance according to the given specifier.
The specifier should be a standard format specifier, with the
form described in PEP 3101. Formatting types 'e', 'E', 'f',
- 'F', 'g', 'G', and '%' are supported. If the formatting type
- is omitted it defaults to 'g' or 'G', depending on the value
- of context.capitals.
-
- At this time the 'n' format specifier type (which is supposed
- to use the current locale) is not supported.
+ 'F', 'g', 'G', 'n' and '%' are supported. If the formatting
+ type is omitted it defaults to 'g' or 'G', depending on the
+ value of context.capitals.
"""
# Note: PEP 3101 says that if the type is not present then
@@ -3548,17 +3619,20 @@
if context is None:
context = getcontext()
- spec = _parse_format_specifier(specifier)
-
- # special values don't care about the type or precision...
+ spec = _parse_format_specifier(specifier, _localeconv=_localeconv)
+
+ # special values don't care about the type or precision
if self._is_special:
- return _format_align(str(self), spec)
+ sign = _format_sign(self._sign, spec)
+ body = str(self.copy_abs())
+ return _format_align(sign, body, spec)
# a type of None defaults to 'g' or 'G', depending on context
- # if type is '%', adjust exponent of self accordingly
if spec['type'] is None:
spec['type'] = ['g', 'G'][context.capitals]
- elif spec['type'] == '%':
+
+ # if type is '%', adjust exponent of self accordingly
+ if spec['type'] == '%':
self = _dec_from_triple(self._sign, self._int, self._exp+2)
# round if necessary, taking rounding mode from the context
@@ -3567,53 +3641,45 @@
if precision is not None:
if spec['type'] in 'eE':
self = self._round(precision+1, rounding)
- elif spec['type'] in 'gG':
- if len(self._int) > precision:
- self = self._round(precision, rounding)
elif spec['type'] in 'fF%':
self = self._rescale(-precision, rounding)
+ elif spec['type'] in 'gG' and len(self._int) > precision:
+ self = self._round(precision, rounding)
# special case: zeros with a positive exponent can't be
# represented in fixed point; rescale them to 0e0.
- elif not self and self._exp > 0 and spec['type'] in 'fF%':
+ if not self and self._exp > 0 and spec['type'] in 'fF%':
self = self._rescale(0, rounding)
# figure out placement of the decimal point
leftdigits = self._exp + len(self._int)
- if spec['type'] in 'fF%':
- dotplace = leftdigits
- elif spec['type'] in 'eE':
+ if spec['type'] in 'eE':
if not self and precision is not None:
dotplace = 1 - precision
else:
dotplace = 1
+ elif spec['type'] in 'fF%':
+ dotplace = leftdigits
elif spec['type'] in 'gG':
if self._exp <= 0 and leftdigits > -6:
dotplace = leftdigits
else:
dotplace = 1
- # figure out main part of numeric string...
- if dotplace <= 0:
- num = '0.' + '0'*(-dotplace) + self._int
- elif dotplace >= len(self._int):
- # make sure we're not padding a '0' with extra zeros on the right
- assert dotplace==len(self._int) or self._int != '0'
- num = self._int + '0'*(dotplace-len(self._int))
+ # find digits before and after decimal point, and get exponent
+ if dotplace < 0:
+ intpart = '0'
+ fracpart = '0'*(-dotplace) + self._int
+ elif dotplace > len(self._int):
+ intpart = self._int + '0'*(dotplace-len(self._int))
+ fracpart = ''
else:
- num = self._int[:dotplace] + '.' + self._int[dotplace:]
-
- # ...then the trailing exponent, or trailing '%'
- if leftdigits != dotplace or spec['type'] in 'eE':
- echar = {'E': 'E', 'e': 'e', 'G': 'E', 'g': 'e'}[spec['type']]
- num = num + "{0}{1:+}".format(echar, leftdigits-dotplace)
- elif spec['type'] == '%':
- num = num + '%'
-
- # add sign
- if self._sign == 1:
- num = '-' + num
- return _format_align(num, spec)
-
+ intpart = self._int[:dotplace] or '0'
+ fracpart = self._int[dotplace:]
+ exp = leftdigits-dotplace
+
+ # done with the decimal-specific stuff; hand over the rest
+ # of the formatting to the _format_number function
+ return _format_number(self._sign, intpart, fracpart, exp, spec)
def _dec_from_triple(sign, coefficient, exponent, special=False):
"""Create a decimal instance directly, without any validation,
@@ -3842,6 +3908,23 @@
"diagnostic info too long in NaN")
return d._fix(self)
+ def create_decimal_from_float(self, f):
+ """Creates a new Decimal instance from a float but rounding using self
+ as the context.
+
+ >>> context = Context(prec=5, rounding=ROUND_DOWN)
+ >>> context.create_decimal_from_float(3.1415926535897932)
+ Decimal('3.1415')
+ >>> context = Context(prec=5, traps=[Inexact])
+ >>> context.create_decimal_from_float(3.1415926535897932)
+ Traceback (most recent call last):
+ ...
+ Inexact: None
+
+ """
+ d = Decimal.from_float(f) # An exact conversion
+ return d._fix(self) # Apply the context rounding
+
# Methods
def abs(self, a):
"""Returns the absolute value of the operand.
@@ -3858,7 +3941,10 @@
Decimal('101.5')
>>> ExtendedContext.abs(Decimal('-101.5'))
Decimal('101.5')
+ >>> ExtendedContext.abs(-1)
+ Decimal('1')
"""
+ a = _convert_other(a, raiseit=True)
return a.__abs__(context=self)
def add(self, a, b):
@@ -3868,8 +3954,19 @@
Decimal('19.00')
>>> ExtendedContext.add(Decimal('1E+2'), Decimal('1.01E+4'))
Decimal('1.02E+4')
+ >>> ExtendedContext.add(1, Decimal(2))
+ Decimal('3')
+ >>> ExtendedContext.add(Decimal(8), 5)
+ Decimal('13')
+ >>> ExtendedContext.add(5, 5)
+ Decimal('10')
"""
- return a.__add__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__add__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def _apply(self, a):
return str(a._fix(self))
@@ -3911,7 +4008,14 @@
Decimal('1')
>>> ExtendedContext.compare(Decimal('-3'), Decimal('2.1'))
Decimal('-1')
+ >>> ExtendedContext.compare(1, 2)
+ Decimal('-1')
+ >>> ExtendedContext.compare(Decimal(1), 2)
+ Decimal('-1')
+ >>> ExtendedContext.compare(1, Decimal(2))
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.compare(b, context=self)
def compare_signal(self, a, b):
@@ -3939,7 +4043,14 @@
Decimal('NaN')
>>> print c.flags[InvalidOperation]
1
+ >>> c.compare_signal(-1, 2)
+ Decimal('-1')
+ >>> c.compare_signal(Decimal(-1), 2)
+ Decimal('-1')
+ >>> c.compare_signal(-1, Decimal(2))
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.compare_signal(b, context=self)
def compare_total(self, a, b):
@@ -3961,7 +4072,14 @@
Decimal('1')
>>> ExtendedContext.compare_total(Decimal('12.3'), Decimal('NaN'))
Decimal('-1')
+ >>> ExtendedContext.compare_total(1, 2)
+ Decimal('-1')
+ >>> ExtendedContext.compare_total(Decimal(1), 2)
+ Decimal('-1')
+ >>> ExtendedContext.compare_total(1, Decimal(2))
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.compare_total(b)
def compare_total_mag(self, a, b):
@@ -3969,6 +4087,7 @@
Like compare_total, but with operand's sign ignored and assumed to be 0.
"""
+ a = _convert_other(a, raiseit=True)
return a.compare_total_mag(b)
def copy_abs(self, a):
@@ -3978,17 +4097,23 @@
Decimal('2.1')
>>> ExtendedContext.copy_abs(Decimal('-100'))
Decimal('100')
+ >>> ExtendedContext.copy_abs(-1)
+ Decimal('1')
"""
+ a = _convert_other(a, raiseit=True)
return a.copy_abs()
def copy_decimal(self, a):
- """Returns a copy of the decimal objet.
+ """Returns a copy of the decimal object.
>>> ExtendedContext.copy_decimal(Decimal('2.1'))
Decimal('2.1')
>>> ExtendedContext.copy_decimal(Decimal('-1.00'))
Decimal('-1.00')
+ >>> ExtendedContext.copy_decimal(1)
+ Decimal('1')
"""
+ a = _convert_other(a, raiseit=True)
return Decimal(a)
def copy_negate(self, a):
@@ -3998,7 +4123,10 @@
Decimal('-101.5')
>>> ExtendedContext.copy_negate(Decimal('-101.5'))
Decimal('101.5')
+ >>> ExtendedContext.copy_negate(1)
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.copy_negate()
def copy_sign(self, a, b):
@@ -4015,7 +4143,14 @@
Decimal('-1.50')
>>> ExtendedContext.copy_sign(Decimal('-1.50'), Decimal('-7.33'))
Decimal('-1.50')
+ >>> ExtendedContext.copy_sign(1, -2)
+ Decimal('-1')
+ >>> ExtendedContext.copy_sign(Decimal(1), -2)
+ Decimal('-1')
+ >>> ExtendedContext.copy_sign(1, Decimal(-2))
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.copy_sign(b)
def divide(self, a, b):
@@ -4041,8 +4176,19 @@
Decimal('1000')
>>> ExtendedContext.divide(Decimal('2.40E+6'), Decimal('2'))
Decimal('1.20E+6')
+ >>> ExtendedContext.divide(5, 5)
+ Decimal('1')
+ >>> ExtendedContext.divide(Decimal(5), 5)
+ Decimal('1')
+ >>> ExtendedContext.divide(5, Decimal(5))
+ Decimal('1')
"""
- return a.__div__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__div__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def divide_int(self, a, b):
"""Divides two numbers and returns the integer part of the result.
@@ -4053,18 +4199,40 @@
Decimal('3')
>>> ExtendedContext.divide_int(Decimal('1'), Decimal('0.3'))
Decimal('3')
+ >>> ExtendedContext.divide_int(10, 3)
+ Decimal('3')
+ >>> ExtendedContext.divide_int(Decimal(10), 3)
+ Decimal('3')
+ >>> ExtendedContext.divide_int(10, Decimal(3))
+ Decimal('3')
"""
- return a.__floordiv__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__floordiv__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def divmod(self, a, b):
- """Return (a // b, a % b)
+ """Return (a // b, a % b).
>>> ExtendedContext.divmod(Decimal(8), Decimal(3))
(Decimal('2'), Decimal('2'))
>>> ExtendedContext.divmod(Decimal(8), Decimal(4))
(Decimal('2'), Decimal('0'))
+ >>> ExtendedContext.divmod(8, 4)
+ (Decimal('2'), Decimal('0'))
+ >>> ExtendedContext.divmod(Decimal(8), 4)
+ (Decimal('2'), Decimal('0'))
+ >>> ExtendedContext.divmod(8, Decimal(4))
+ (Decimal('2'), Decimal('0'))
"""
- return a.__divmod__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__divmod__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def exp(self, a):
"""Returns e ** a.
@@ -4084,7 +4252,10 @@
Decimal('2.00000000')
>>> c.exp(Decimal('+Infinity'))
Decimal('Infinity')
+ >>> c.exp(10)
+ Decimal('22026.4658')
"""
+ a =_convert_other(a, raiseit=True)
return a.exp(context=self)
def fma(self, a, b, c):
@@ -4100,7 +4271,14 @@
Decimal('-8')
>>> ExtendedContext.fma(Decimal('888565290'), Decimal('1557.96930'), Decimal('-86087.7578'))
Decimal('1.38435736E+12')
+ >>> ExtendedContext.fma(1, 3, 4)
+ Decimal('7')
+ >>> ExtendedContext.fma(1, Decimal(3), 4)
+ Decimal('7')
+ >>> ExtendedContext.fma(1, 3, Decimal(4))
+ Decimal('7')
"""
+ a = _convert_other(a, raiseit=True)
return a.fma(b, c, context=self)
def is_canonical(self, a):
@@ -4130,7 +4308,10 @@
False
>>> ExtendedContext.is_finite(Decimal('NaN'))
False
+ >>> ExtendedContext.is_finite(1)
+ True
"""
+ a = _convert_other(a, raiseit=True)
return a.is_finite()
def is_infinite(self, a):
@@ -4142,7 +4323,10 @@
True
>>> ExtendedContext.is_infinite(Decimal('NaN'))
False
+ >>> ExtendedContext.is_infinite(1)
+ False
"""
+ a = _convert_other(a, raiseit=True)
return a.is_infinite()
def is_nan(self, a):
@@ -4155,7 +4339,10 @@
True
>>> ExtendedContext.is_nan(Decimal('-sNaN'))
True
+ >>> ExtendedContext.is_nan(1)
+ False
"""
+ a = _convert_other(a, raiseit=True)
return a.is_nan()
def is_normal(self, a):
@@ -4175,7 +4362,10 @@
False
>>> c.is_normal(Decimal('NaN'))
False
+ >>> c.is_normal(1)
+ True
"""
+ a = _convert_other(a, raiseit=True)
return a.is_normal(context=self)
def is_qnan(self, a):
@@ -4187,7 +4377,10 @@
True
>>> ExtendedContext.is_qnan(Decimal('sNaN'))
False
+ >>> ExtendedContext.is_qnan(1)
+ False
"""
+ a = _convert_other(a, raiseit=True)
return a.is_qnan()
def is_signed(self, a):
@@ -4199,7 +4392,12 @@
True
>>> ExtendedContext.is_signed(Decimal('-0'))
True
+ >>> ExtendedContext.is_signed(8)
+ False
+ >>> ExtendedContext.is_signed(-8)
+ True
"""
+ a = _convert_other(a, raiseit=True)
return a.is_signed()
def is_snan(self, a):
@@ -4212,7 +4410,10 @@
False
>>> ExtendedContext.is_snan(Decimal('sNaN'))
True
+ >>> ExtendedContext.is_snan(1)
+ False
"""
+ a = _convert_other(a, raiseit=True)
return a.is_snan()
def is_subnormal(self, a):
@@ -4231,7 +4432,10 @@
False
>>> c.is_subnormal(Decimal('NaN'))
False
+ >>> c.is_subnormal(1)
+ False
"""
+ a = _convert_other(a, raiseit=True)
return a.is_subnormal(context=self)
def is_zero(self, a):
@@ -4243,7 +4447,12 @@
False
>>> ExtendedContext.is_zero(Decimal('-0E+2'))
True
+ >>> ExtendedContext.is_zero(1)
+ False
+ >>> ExtendedContext.is_zero(0)
+ True
"""
+ a = _convert_other(a, raiseit=True)
return a.is_zero()
def ln(self, a):
@@ -4262,7 +4471,10 @@
Decimal('2.30258509')
>>> c.ln(Decimal('+Infinity'))
Decimal('Infinity')
+ >>> c.ln(1)
+ Decimal('0')
"""
+ a = _convert_other(a, raiseit=True)
return a.ln(context=self)
def log10(self, a):
@@ -4285,7 +4497,12 @@
Decimal('1.84509804')
>>> c.log10(Decimal('+Infinity'))
Decimal('Infinity')
+ >>> c.log10(0)
+ Decimal('-Infinity')
+ >>> c.log10(1)
+ Decimal('0')
"""
+ a = _convert_other(a, raiseit=True)
return a.log10(context=self)
def logb(self, a):
@@ -4304,7 +4521,14 @@
Decimal('-2')
>>> ExtendedContext.logb(Decimal('0'))
Decimal('-Infinity')
+ >>> ExtendedContext.logb(1)
+ Decimal('0')
+ >>> ExtendedContext.logb(10)
+ Decimal('1')
+ >>> ExtendedContext.logb(100)
+ Decimal('2')
"""
+ a = _convert_other(a, raiseit=True)
return a.logb(context=self)
def logical_and(self, a, b):
@@ -4324,7 +4548,14 @@
Decimal('1000')
>>> ExtendedContext.logical_and(Decimal('1111'), Decimal('10'))
Decimal('10')
+ >>> ExtendedContext.logical_and(110, 1101)
+ Decimal('100')
+ >>> ExtendedContext.logical_and(Decimal(110), 1101)
+ Decimal('100')
+ >>> ExtendedContext.logical_and(110, Decimal(1101))
+ Decimal('100')
"""
+ a = _convert_other(a, raiseit=True)
return a.logical_and(b, context=self)
def logical_invert(self, a):
@@ -4340,7 +4571,10 @@
Decimal('0')
>>> ExtendedContext.logical_invert(Decimal('101010101'))
Decimal('10101010')
+ >>> ExtendedContext.logical_invert(1101)
+ Decimal('111110010')
"""
+ a = _convert_other(a, raiseit=True)
return a.logical_invert(context=self)
def logical_or(self, a, b):
@@ -4360,7 +4594,14 @@
Decimal('1110')
>>> ExtendedContext.logical_or(Decimal('1110'), Decimal('10'))
Decimal('1110')
+ >>> ExtendedContext.logical_or(110, 1101)
+ Decimal('1111')
+ >>> ExtendedContext.logical_or(Decimal(110), 1101)
+ Decimal('1111')
+ >>> ExtendedContext.logical_or(110, Decimal(1101))
+ Decimal('1111')
"""
+ a = _convert_other(a, raiseit=True)
return a.logical_or(b, context=self)
def logical_xor(self, a, b):
@@ -4380,10 +4621,17 @@
Decimal('110')
>>> ExtendedContext.logical_xor(Decimal('1111'), Decimal('10'))
Decimal('1101')
+ >>> ExtendedContext.logical_xor(110, 1101)
+ Decimal('1011')
+ >>> ExtendedContext.logical_xor(Decimal(110), 1101)
+ Decimal('1011')
+ >>> ExtendedContext.logical_xor(110, Decimal(1101))
+ Decimal('1011')
"""
+ a = _convert_other(a, raiseit=True)
return a.logical_xor(b, context=self)
- def max(self, a,b):
+ def max(self, a, b):
"""max compares two values numerically and returns the maximum.
If either operand is a NaN then the general rules apply.
@@ -4400,14 +4648,34 @@
Decimal('1')
>>> ExtendedContext.max(Decimal('7'), Decimal('NaN'))
Decimal('7')
+ >>> ExtendedContext.max(1, 2)
+ Decimal('2')
+ >>> ExtendedContext.max(Decimal(1), 2)
+ Decimal('2')
+ >>> ExtendedContext.max(1, Decimal(2))
+ Decimal('2')
"""
+ a = _convert_other(a, raiseit=True)
return a.max(b, context=self)
def max_mag(self, a, b):
- """Compares the values numerically with their sign ignored."""
+ """Compares the values numerically with their sign ignored.
+
+ >>> ExtendedContext.max_mag(Decimal('7'), Decimal('NaN'))
+ Decimal('7')
+ >>> ExtendedContext.max_mag(Decimal('7'), Decimal('-10'))
+ Decimal('-10')
+ >>> ExtendedContext.max_mag(1, -2)
+ Decimal('-2')
+ >>> ExtendedContext.max_mag(Decimal(1), -2)
+ Decimal('-2')
+ >>> ExtendedContext.max_mag(1, Decimal(-2))
+ Decimal('-2')
+ """
+ a = _convert_other(a, raiseit=True)
return a.max_mag(b, context=self)
- def min(self, a,b):
+ def min(self, a, b):
"""min compares two values numerically and returns the minimum.
If either operand is a NaN then the general rules apply.
@@ -4424,11 +4692,31 @@
Decimal('1.0')
>>> ExtendedContext.min(Decimal('7'), Decimal('NaN'))
Decimal('7')
+ >>> ExtendedContext.min(1, 2)
+ Decimal('1')
+ >>> ExtendedContext.min(Decimal(1), 2)
+ Decimal('1')
+ >>> ExtendedContext.min(1, Decimal(29))
+ Decimal('1')
"""
+ a = _convert_other(a, raiseit=True)
return a.min(b, context=self)
def min_mag(self, a, b):
- """Compares the values numerically with their sign ignored."""
+ """Compares the values numerically with their sign ignored.
+
+ >>> ExtendedContext.min_mag(Decimal('3'), Decimal('-2'))
+ Decimal('-2')
+ >>> ExtendedContext.min_mag(Decimal('-3'), Decimal('NaN'))
+ Decimal('-3')
+ >>> ExtendedContext.min_mag(1, -2)
+ Decimal('1')
+ >>> ExtendedContext.min_mag(Decimal(1), -2)
+ Decimal('1')
+ >>> ExtendedContext.min_mag(1, Decimal(-2))
+ Decimal('1')
+ """
+ a = _convert_other(a, raiseit=True)
return a.min_mag(b, context=self)
def minus(self, a):
@@ -4442,16 +4730,19 @@
Decimal('-1.3')
>>> ExtendedContext.minus(Decimal('-1.3'))
Decimal('1.3')
+ >>> ExtendedContext.minus(1)
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.__neg__(context=self)
def multiply(self, a, b):
"""multiply multiplies two operands.
If either operand is a special value then the general rules apply.
- Otherwise, the operands are multiplied together ('long multiplication'),
- resulting in a number which may be as long as the sum of the lengths
- of the two operands.
+ Otherwise, the operands are multiplied together
+ ('long multiplication'), resulting in a number which may be as long as
+ the sum of the lengths of the two operands.
>>> ExtendedContext.multiply(Decimal('1.20'), Decimal('3'))
Decimal('3.60')
@@ -4463,8 +4754,19 @@
Decimal('-0.0')
>>> ExtendedContext.multiply(Decimal('654321'), Decimal('654321'))
Decimal('4.28135971E+11')
+ >>> ExtendedContext.multiply(7, 7)
+ Decimal('49')
+ >>> ExtendedContext.multiply(Decimal(7), 7)
+ Decimal('49')
+ >>> ExtendedContext.multiply(7, Decimal(7))
+ Decimal('49')
"""
- return a.__mul__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__mul__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def next_minus(self, a):
"""Returns the largest representable number smaller than a.
@@ -4480,7 +4782,10 @@
Decimal('-1.00000004')
>>> c.next_minus(Decimal('Infinity'))
Decimal('9.99999999E+999')
+ >>> c.next_minus(1)
+ Decimal('0.999999999')
"""
+ a = _convert_other(a, raiseit=True)
return a.next_minus(context=self)
def next_plus(self, a):
@@ -4497,7 +4802,10 @@
Decimal('-1.00000002')
>>> c.next_plus(Decimal('-Infinity'))
Decimal('-9.99999999E+999')
+ >>> c.next_plus(1)
+ Decimal('1.00000001')
"""
+ a = _convert_other(a, raiseit=True)
return a.next_plus(context=self)
def next_toward(self, a, b):
@@ -4525,7 +4833,14 @@
Decimal('-1.00000004')
>>> c.next_toward(Decimal('0.00'), Decimal('-0.0000'))
Decimal('-0.00')
+ >>> c.next_toward(0, 1)
+ Decimal('1E-1007')
+ >>> c.next_toward(Decimal(0), 1)
+ Decimal('1E-1007')
+ >>> c.next_toward(0, Decimal(1))
+ Decimal('1E-1007')
"""
+ a = _convert_other(a, raiseit=True)
return a.next_toward(b, context=self)
def normalize(self, a):
@@ -4546,7 +4861,10 @@
Decimal('1.2E+2')
>>> ExtendedContext.normalize(Decimal('0.00'))
Decimal('0')
+ >>> ExtendedContext.normalize(6)
+ Decimal('6')
"""
+ a = _convert_other(a, raiseit=True)
return a.normalize(context=self)
def number_class(self, a):
@@ -4593,7 +4911,10 @@
'NaN'
>>> c.number_class(Decimal('sNaN'))
'sNaN'
+ >>> c.number_class(123)
+ '+Normal'
"""
+ a = _convert_other(a, raiseit=True)
return a.number_class(context=self)
def plus(self, a):
@@ -4607,7 +4928,10 @@
Decimal('1.3')
>>> ExtendedContext.plus(Decimal('-1.3'))
Decimal('-1.3')
+ >>> ExtendedContext.plus(-1)
+ Decimal('-1')
"""
+ a = _convert_other(a, raiseit=True)
return a.__pos__(context=self)
def power(self, a, b, modulo=None):
@@ -4676,8 +5000,19 @@
Decimal('-0')
>>> c.power(Decimal('-23'), Decimal('0'), Decimal('65537'))
Decimal('1')
+ >>> ExtendedContext.power(7, 7)
+ Decimal('823543')
+ >>> ExtendedContext.power(Decimal(7), 7)
+ Decimal('823543')
+ >>> ExtendedContext.power(7, Decimal(7), 2)
+ Decimal('1')
"""
- return a.__pow__(b, modulo, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__pow__(b, modulo, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def quantize(self, a, b):
"""Returns a value equal to 'a' (rounded), having the exponent of 'b'.
@@ -4727,7 +5062,14 @@
Decimal('2.2E+2')
>>> ExtendedContext.quantize(Decimal('217'), Decimal('1e+2'))
Decimal('2E+2')
+ >>> ExtendedContext.quantize(1, 2)
+ Decimal('1')
+ >>> ExtendedContext.quantize(Decimal(1), 2)
+ Decimal('1')
+ >>> ExtendedContext.quantize(1, Decimal(2))
+ Decimal('1')
"""
+ a = _convert_other(a, raiseit=True)
return a.quantize(b, context=self)
def radix(self):
@@ -4762,8 +5104,19 @@
Decimal('0.1')
>>> ExtendedContext.remainder(Decimal('3.6'), Decimal('1.3'))
Decimal('1.0')
+ >>> ExtendedContext.remainder(22, 6)
+ Decimal('4')
+ >>> ExtendedContext.remainder(Decimal(22), 6)
+ Decimal('4')
+ >>> ExtendedContext.remainder(22, Decimal(6))
+ Decimal('4')
"""
- return a.__mod__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__mod__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def remainder_near(self, a, b):
"""Returns to be "a - b * n", where n is the integer nearest the exact
@@ -4789,7 +5142,14 @@
Decimal('0.1')
>>> ExtendedContext.remainder_near(Decimal('3.6'), Decimal('1.3'))
Decimal('-0.3')
+ >>> ExtendedContext.remainder_near(3, 11)
+ Decimal('3')
+ >>> ExtendedContext.remainder_near(Decimal(3), 11)
+ Decimal('3')
+ >>> ExtendedContext.remainder_near(3, Decimal(11))
+ Decimal('3')
"""
+ a = _convert_other(a, raiseit=True)
return a.remainder_near(b, context=self)
def rotate(self, a, b):
@@ -4811,7 +5171,14 @@
Decimal('123456789')
>>> ExtendedContext.rotate(Decimal('123456789'), Decimal('+2'))
Decimal('345678912')
+ >>> ExtendedContext.rotate(1333333, 1)
+ Decimal('13333330')
+ >>> ExtendedContext.rotate(Decimal(1333333), 1)
+ Decimal('13333330')
+ >>> ExtendedContext.rotate(1333333, Decimal(1))
+ Decimal('13333330')
"""
+ a = _convert_other(a, raiseit=True)
return a.rotate(b, context=self)
def same_quantum(self, a, b):
@@ -4828,7 +5195,14 @@
False
>>> ExtendedContext.same_quantum(Decimal('Inf'), Decimal('-Inf'))
True
+ >>> ExtendedContext.same_quantum(10000, -1)
+ True
+ >>> ExtendedContext.same_quantum(Decimal(10000), -1)
+ True
+ >>> ExtendedContext.same_quantum(10000, Decimal(-1))
+ True
"""
+ a = _convert_other(a, raiseit=True)
return a.same_quantum(b)
def scaleb (self, a, b):
@@ -4840,8 +5214,15 @@
Decimal('7.50')
>>> ExtendedContext.scaleb(Decimal('7.50'), Decimal('3'))
Decimal('7.50E+3')
+ >>> ExtendedContext.scaleb(1, 4)
+ Decimal('1E+4')
+ >>> ExtendedContext.scaleb(Decimal(1), 4)
+ Decimal('1E+4')
+ >>> ExtendedContext.scaleb(1, Decimal(4))
+ Decimal('1E+4')
"""
- return a.scaleb (b, context=self)
+ a = _convert_other(a, raiseit=True)
+ return a.scaleb(b, context=self)
def shift(self, a, b):
"""Returns a shifted copy of a, b times.
@@ -4863,7 +5244,14 @@
Decimal('123456789')
>>> ExtendedContext.shift(Decimal('123456789'), Decimal('+2'))
Decimal('345678900')
+ >>> ExtendedContext.shift(88888888, 2)
+ Decimal('888888800')
+ >>> ExtendedContext.shift(Decimal(88888888), 2)
+ Decimal('888888800')
+ >>> ExtendedContext.shift(88888888, Decimal(2))
+ Decimal('888888800')
"""
+ a = _convert_other(a, raiseit=True)
return a.shift(b, context=self)
def sqrt(self, a):
@@ -4890,9 +5278,12 @@
Decimal('2.64575131')
>>> ExtendedContext.sqrt(Decimal('10'))
Decimal('3.16227766')
+ >>> ExtendedContext.sqrt(2)
+ Decimal('1.41421356')
>>> ExtendedContext.prec
9
"""
+ a = _convert_other(a, raiseit=True)
return a.sqrt(context=self)
def subtract(self, a, b):
@@ -4904,14 +5295,26 @@
Decimal('0.00')
>>> ExtendedContext.subtract(Decimal('1.3'), Decimal('2.07'))
Decimal('-0.77')
+ >>> ExtendedContext.subtract(8, 5)
+ Decimal('3')
+ >>> ExtendedContext.subtract(Decimal(8), 5)
+ Decimal('3')
+ >>> ExtendedContext.subtract(8, Decimal(5))
+ Decimal('3')
"""
- return a.__sub__(b, context=self)
+ a = _convert_other(a, raiseit=True)
+ r = a.__sub__(b, context=self)
+ if r is NotImplemented:
+ raise TypeError("Unable to convert %s to Decimal" % b)
+ else:
+ return r
def to_eng_string(self, a):
"""Converts a number to a string, using scientific notation.
The operation is not affected by the context.
"""
+ a = _convert_other(a, raiseit=True)
return a.to_eng_string(context=self)
def to_sci_string(self, a):
@@ -4919,6 +5322,7 @@
The operation is not affected by the context.
"""
+ a = _convert_other(a, raiseit=True)
return a.__str__(context=self)
def to_integral_exact(self, a):
@@ -4948,6 +5352,7 @@
>>> ExtendedContext.to_integral_exact(Decimal('-Inf'))
Decimal('-Infinity')
"""
+ a = _convert_other(a, raiseit=True)
return a.to_integral_exact(context=self)
def to_integral_value(self, a):
@@ -4976,6 +5381,7 @@
>>> ExtendedContext.to_integral_value(Decimal('-Inf'))
Decimal('-Infinity')
"""
+ a = _convert_other(a, raiseit=True)
return a.to_integral_value(context=self)
# the method name changed, but we provide also the old one, for compatibility
@@ -5377,15 +5783,21 @@
##### Helper Functions ####################################################
-def _convert_other(other, raiseit=False):
+def _convert_other(other, raiseit=False, allow_float=False):
"""Convert other to Decimal.
Verifies that it's ok to use in an implicit construction.
+ If allow_float is true, allow conversion from float; this
+ is used in the comparison methods (__eq__ and friends).
+
"""
if isinstance(other, Decimal):
return other
if isinstance(other, (int, long)):
return Decimal(other)
+ if allow_float and isinstance(other, float):
+ return Decimal.from_float(other)
+
if raiseit:
raise TypeError("Unable to convert %s to Decimal" % other)
return NotImplemented
@@ -5460,14 +5872,13 @@
_exact_half = re.compile('50*$').match
##### PEP3101 support functions ##############################################
-# The functions parse_format_specifier and format_align have little to do
-# with the Decimal class, and could potentially be reused for other pure
+# The functions in this section have little to do with the Decimal
+# class, and could potentially be reused or adapted for other pure
# Python numeric classes that want to implement __format__
#
# A format specifier for Decimal looks like:
#
-# [[fill]align][sign][0][minimumwidth][.precision][type]
-#
+# [[fill]align][sign][0][minimumwidth][,][.precision][type]
_parse_format_specifier_regex = re.compile(r"""\A
(?:
@@ -5477,14 +5888,23 @@
(?P<sign>[-+ ])?
(?P<zeropad>0)?
(?P<minimumwidth>(?!0)\d+)?
+(?P<thousands_sep>,)?
(?:\.(?P<precision>0|(?!0)\d+))?
-(?P<type>[eEfFgG%])?
+(?P<type>[eEfFgGn%])?
\Z
""", re.VERBOSE)
del re
-def _parse_format_specifier(format_spec):
+# The locale module is only needed for the 'n' format specifier. The
+# rest of the PEP 3101 code functions quite happily without it, so we
+# don't care too much if locale isn't present.
+try:
+ import locale as _locale
+except ImportError:
+ pass
+
+def _parse_format_specifier(format_spec, _localeconv=None):
"""Parse and validate a format specifier.
Turns a standard numeric format specifier into a dict, with the
@@ -5494,9 +5914,14 @@
align: alignment type, either '<', '>', '=' or '^'
sign: either '+', '-' or ' '
minimumwidth: nonnegative integer giving minimum width
+ zeropad: boolean, indicating whether to pad with zeros
+ thousands_sep: string to use as thousands separator, or ''
+ grouping: grouping for thousands separators, in format
+ used by localeconv
+ decimal_point: string to use for decimal point
precision: nonnegative integer giving precision, or None
type: one of the characters 'eEfFgG%', or None
- unicode: either True or False (always True for Python 3.x)
+ unicode: boolean (always True for Python 3.x)
"""
m = _parse_format_specifier_regex.match(format_spec)
@@ -5506,26 +5931,28 @@
# get the dictionary
format_dict = m.groupdict()
- # defaults for fill and alignment
+ # zeropad; defaults for fill and alignment. If zero padding
+ # is requested, the fill and align fields should be absent.
fill = format_dict['fill']
align = format_dict['align']
- if format_dict.pop('zeropad') is not None:
- # in the face of conflict, refuse the temptation to guess
- if fill is not None and fill != '0':
+ format_dict['zeropad'] = (format_dict['zeropad'] is not None)
+ if format_dict['zeropad']:
+ if fill is not None:
raise ValueError("Fill character conflicts with '0'"
" in format specifier: " + format_spec)
- if align is not None and align != '=':
+ if align is not None:
raise ValueError("Alignment conflicts with '0' in "
"format specifier: " + format_spec)
- fill = '0'
- align = '='
format_dict['fill'] = fill or ' '
- format_dict['align'] = align or '<'
-
+ # PEP 3101 originally specified that the default alignment should
+ # be left; it was later agreed that right-aligned makes more sense
+ # for numeric types. See http://bugs.python.org/issue6857.
+ format_dict['align'] = align or '>'
+
+ # default sign handling: '-' for negative, '' for positive
if format_dict['sign'] is None:
format_dict['sign'] = '-'
- # turn minimumwidth and precision entries into integers.
# minimumwidth defaults to 0; precision remains None if not given
format_dict['minimumwidth'] = int(format_dict['minimumwidth'] or '0')
if format_dict['precision'] is not None:
@@ -5537,58 +5964,171 @@
if format_dict['type'] is None or format_dict['type'] in 'gG':
format_dict['precision'] = 1
+ # determine thousands separator, grouping, and decimal separator, and
+ # add appropriate entries to format_dict
+ if format_dict['type'] == 'n':
+ # apart from separators, 'n' behaves just like 'g'
+ format_dict['type'] = 'g'
+ if _localeconv is None:
+ _localeconv = _locale.localeconv()
+ if format_dict['thousands_sep'] is not None:
+ raise ValueError("Explicit thousands separator conflicts with "
+ "'n' type in format specifier: " + format_spec)
+ format_dict['thousands_sep'] = _localeconv['thousands_sep']
+ format_dict['grouping'] = _localeconv['grouping']
+ format_dict['decimal_point'] = _localeconv['decimal_point']
+ else:
+ if format_dict['thousands_sep'] is None:
+ format_dict['thousands_sep'] = ''
+ format_dict['grouping'] = [3, 0]
+ format_dict['decimal_point'] = '.'
+
# record whether return type should be str or unicode
format_dict['unicode'] = isinstance(format_spec, unicode)
return format_dict
-def _format_align(body, spec_dict):
- """Given an unpadded, non-aligned numeric string, add padding and
- aligment to conform with the given format specifier dictionary (as
- output from parse_format_specifier).
-
- It's assumed that if body is negative then it starts with '-'.
- Any leading sign ('-' or '+') is stripped from the body before
- applying the alignment and padding rules, and replaced in the
- appropriate position.
+def _format_align(sign, body, spec):
+ """Given an unpadded, non-aligned numeric string 'body' and sign
+ string 'sign', add padding and aligment conforming to the given
+ format specifier dictionary 'spec' (as produced by
+ parse_format_specifier).
+
+ Also converts result to unicode if necessary.
"""
- # figure out the sign; we only examine the first character, so if
- # body has leading whitespace the results may be surprising.
- if len(body) > 0 and body[0] in '-+':
- sign = body[0]
- body = body[1:]
- else:
- sign = ''
-
- if sign != '-':
- if spec_dict['sign'] in ' +':
- sign = spec_dict['sign']
- else:
- sign = ''
-
# how much extra space do we have to play with?
- minimumwidth = spec_dict['minimumwidth']
- fill = spec_dict['fill']
- padding = fill*(max(minimumwidth - (len(sign+body)), 0))
-
- align = spec_dict['align']
+ minimumwidth = spec['minimumwidth']
+ fill = spec['fill']
+ padding = fill*(minimumwidth - len(sign) - len(body))
+
+ align = spec['align']
if align == '<':
result = sign + body + padding
elif align == '>':
result = padding + sign + body
elif align == '=':
result = sign + padding + body
- else: #align == '^'
+ elif align == '^':
half = len(padding)//2
result = padding[:half] + sign + body + padding[half:]
+ else:
+ raise ValueError('Unrecognised alignment field')
# make sure that result is unicode if necessary
- if spec_dict['unicode']:
+ if spec['unicode']:
result = unicode(result)
return result
+def _group_lengths(grouping):
+ """Convert a localeconv-style grouping into a (possibly infinite)
+ iterable of integers representing group lengths.
+
+ """
+ # The result from localeconv()['grouping'], and the input to this
+ # function, should be a list of integers in one of the
+ # following three forms:
+ #
+ # (1) an empty list, or
+ # (2) nonempty list of positive integers + [0]
+ # (3) list of positive integers + [locale.CHAR_MAX], or
+
+ from itertools import chain, repeat
+ if not grouping:
+ return []
+ elif grouping[-1] == 0 and len(grouping) >= 2:
+ return chain(grouping[:-1], repeat(grouping[-2]))
+ elif grouping[-1] == _locale.CHAR_MAX:
+ return grouping[:-1]
+ else:
+ raise ValueError('unrecognised format for grouping')
+
+def _insert_thousands_sep(digits, spec, min_width=1):
+ """Insert thousands separators into a digit string.
+
+ spec is a dictionary whose keys should include 'thousands_sep' and
+ 'grouping'; typically it's the result of parsing the format
+ specifier using _parse_format_specifier.
+
+ The min_width keyword argument gives the minimum length of the
+ result, which will be padded on the left with zeros if necessary.
+
+ If necessary, the zero padding adds an extra '0' on the left to
+ avoid a leading thousands separator. For example, inserting
+ commas every three digits in '123456', with min_width=8, gives
+ '0,123,456', even though that has length 9.
+
+ """
+
+ sep = spec['thousands_sep']
+ grouping = spec['grouping']
+
+ groups = []
+ for l in _group_lengths(grouping):
+ if l <= 0:
+ raise ValueError("group length should be positive")
+ # max(..., 1) forces at least 1 digit to the left of a separator
+ l = min(max(len(digits), min_width, 1), l)
+ groups.append('0'*(l - len(digits)) + digits[-l:])
+ digits = digits[:-l]
+ min_width -= l
+ if not digits and min_width <= 0:
+ break
+ min_width -= len(sep)
+ else:
+ l = max(len(digits), min_width, 1)
+ groups.append('0'*(l - len(digits)) + digits[-l:])
+ return sep.join(reversed(groups))
+
+def _format_sign(is_negative, spec):
+ """Determine sign character."""
+
+ if is_negative:
+ return '-'
+ elif spec['sign'] in ' +':
+ return spec['sign']
+ else:
+ return ''
+
+def _format_number(is_negative, intpart, fracpart, exp, spec):
+ """Format a number, given the following data:
+
+ is_negative: true if the number is negative, else false
+ intpart: string of digits that must appear before the decimal point
+ fracpart: string of digits that must come after the point
+ exp: exponent, as an integer
+ spec: dictionary resulting from parsing the format specifier
+
+ This function uses the information in spec to:
+ insert separators (decimal separator and thousands separators)
+ format the sign
+ format the exponent
+ add trailing '%' for the '%' type
+ zero-pad if necessary
+ fill and align if necessary
+ """
+
+ sign = _format_sign(is_negative, spec)
+
+ if fracpart:
+ fracpart = spec['decimal_point'] + fracpart
+
+ if exp != 0 or spec['type'] in 'eE':
+ echar = {'E': 'E', 'e': 'e', 'G': 'E', 'g': 'e'}[spec['type']]
+ fracpart += "{0}{1:+}".format(echar, exp)
+ if spec['type'] == '%':
+ fracpart += '%'
+
+ if spec['zeropad']:
+ min_width = spec['minimumwidth'] - len(fracpart) - len(sign)
+ else:
+ min_width = 0
+ intpart = _insert_thousands_sep(intpart, spec, min_width)
+
+ return _format_align(sign, intpart+fracpart, spec)
+
+
##### Useful Constants (internal use only) ################################
# Reusable defaults
diff --git a/Lib/distutils/ccompiler.py b/Lib/distutils/ccompiler.py
--- a/Lib/distutils/ccompiler.py
+++ b/Lib/distutils/ccompiler.py
@@ -3,21 +3,73 @@
Contains CCompiler, an abstract base class that defines the interface
for the Distutils compiler abstraction model."""
-# This module should be kept compatible with Python 2.1.
+__revision__ = "$Id: ccompiler.py 86238 2010-11-06 04:06:18Z eric.araujo $"
-__revision__ = "$Id: ccompiler.py 77425 2010-01-11 22:54:57Z tarek.ziade $"
+import sys
+import os
+import re
-import sys, os, re
-from types import *
-from copy import copy
-from distutils.errors import *
+from distutils.errors import (CompileError, LinkError, UnknownFileError,
+ DistutilsPlatformError, DistutilsModuleError)
from distutils.spawn import spawn
from distutils.file_util import move_file
from distutils.dir_util import mkpath
-from distutils.dep_util import newer_pairwise, newer_group
+from distutils.dep_util import newer_group
from distutils.util import split_quoted, execute
from distutils import log
+_sysconfig = __import__('sysconfig')
+
+def customize_compiler(compiler):
+ """Do any platform-specific customization of a CCompiler instance.
+
+ Mainly needed on Unix, so we can plug in the information that
+ varies across Unices and is stored in Python's Makefile.
+ """
+ if compiler.compiler_type == "unix":
+ (cc, cxx, opt, cflags, ccshared, ldshared, so_ext, ar, ar_flags) = \
+ _sysconfig.get_config_vars('CC', 'CXX', 'OPT', 'CFLAGS',
+ 'CCSHARED', 'LDSHARED', 'SO', 'AR',
+ 'ARFLAGS')
+
+ if 'CC' in os.environ:
+ cc = os.environ['CC']
+ if 'CXX' in os.environ:
+ cxx = os.environ['CXX']
+ if 'LDSHARED' in os.environ:
+ ldshared = os.environ['LDSHARED']
+ if 'CPP' in os.environ:
+ cpp = os.environ['CPP']
+ else:
+ cpp = cc + " -E" # not always
+ if 'LDFLAGS' in os.environ:
+ ldshared = ldshared + ' ' + os.environ['LDFLAGS']
+ if 'CFLAGS' in os.environ:
+ cflags = opt + ' ' + os.environ['CFLAGS']
+ ldshared = ldshared + ' ' + os.environ['CFLAGS']
+ if 'CPPFLAGS' in os.environ:
+ cpp = cpp + ' ' + os.environ['CPPFLAGS']
+ cflags = cflags + ' ' + os.environ['CPPFLAGS']
+ ldshared = ldshared + ' ' + os.environ['CPPFLAGS']
+ if 'AR' in os.environ:
+ ar = os.environ['AR']
+ if 'ARFLAGS' in os.environ:
+ archiver = ar + ' ' + os.environ['ARFLAGS']
+ else:
+ archiver = ar + ' ' + ar_flags
+
+ cc_cmd = cc + ' ' + cflags
+ compiler.set_executables(
+ preprocessor=cpp,
+ compiler=cc_cmd,
+ compiler_so=cc_cmd + ' ' + ccshared,
+ compiler_cxx=cxx,
+ linker_so=ldshared,
+ linker_exe=cc,
+ archiver=archiver)
+
+ compiler.shared_lib_extension = so_ext
+
class CCompiler:
"""Abstract base class to define the interface that must be implemented
by real compiler classes. Also has some utility methods used by
@@ -88,11 +140,7 @@
}
language_order = ["c++", "objc", "c"]
- def __init__ (self,
- verbose=0,
- dry_run=0,
- force=0):
-
+ def __init__ (self, verbose=0, dry_run=0, force=0):
self.dry_run = dry_run
self.force = force
self.verbose = verbose
@@ -128,11 +176,7 @@
for key in self.executables.keys():
self.set_executable(key, self.executables[key])
- # __init__ ()
-
-
- def set_executables (self, **args):
-
+ def set_executables(self, **args):
"""Define the executables (and options for them) that will be run
to perform the various stages of compilation. The exact set of
executables that may be specified here depends on the compiler
@@ -165,36 +209,31 @@
(key, self.__class__.__name__)
self.set_executable(key, args[key])
- # set_executables ()
-
def set_executable(self, key, value):
- if type(value) is StringType:
+ if isinstance(value, str):
setattr(self, key, split_quoted(value))
else:
setattr(self, key, value)
-
- def _find_macro (self, name):
+ def _find_macro(self, name):
i = 0
for defn in self.macros:
if defn[0] == name:
return i
i = i + 1
-
return None
-
- def _check_macro_definitions (self, definitions):
+ def _check_macro_definitions(self, definitions):
"""Ensures that every element of 'definitions' is a valid macro
definition, ie. either (name,value) 2-tuple or a (name,) tuple. Do
nothing if all definitions are OK, raise TypeError otherwise.
"""
for defn in definitions:
- if not (type (defn) is TupleType and
+ if not (isinstance(defn, tuple) and
(len (defn) == 1 or
(len (defn) == 2 and
- (type (defn[1]) is StringType or defn[1] is None))) and
- type (defn[0]) is StringType):
+ (isinstance(defn[1], str) or defn[1] is None))) and
+ isinstance(defn[0], str)):
raise TypeError, \
("invalid macro definition '%s': " % defn) + \
"must be tuple (string,), (string, string), or " + \
@@ -203,7 +242,7 @@
# -- Bookkeeping methods -------------------------------------------
- def define_macro (self, name, value=None):
+ def define_macro(self, name, value=None):
"""Define a preprocessor macro for all compilations driven by this
compiler object. The optional parameter 'value' should be a
string; if it is not supplied, then the macro will be defined
@@ -219,8 +258,7 @@
defn = (name, value)
self.macros.append (defn)
-
- def undefine_macro (self, name):
+ def undefine_macro(self, name):
"""Undefine a preprocessor macro for all compilations driven by
this compiler object. If the same macro is defined by
'define_macro()' and undefined by 'undefine_macro()' the last call
@@ -238,8 +276,7 @@
undefn = (name,)
self.macros.append (undefn)
-
- def add_include_dir (self, dir):
+ def add_include_dir(self, dir):
"""Add 'dir' to the list of directories that will be searched for
header files. The compiler is instructed to search directories in
the order in which they are supplied by successive calls to
@@ -247,7 +284,7 @@
"""
self.include_dirs.append (dir)
- def set_include_dirs (self, dirs):
+ def set_include_dirs(self, dirs):
"""Set the list of directories that will be searched to 'dirs' (a
list of strings). Overrides any preceding calls to
'add_include_dir()'; subsequence calls to 'add_include_dir()' add
@@ -255,10 +292,9 @@
any list of standard include directories that the compiler may
search by default.
"""
- self.include_dirs = copy (dirs)
+ self.include_dirs = dirs[:]
-
- def add_library (self, libname):
+ def add_library(self, libname):
"""Add 'libname' to the list of libraries that will be included in
all links driven by this compiler object. Note that 'libname'
should *not* be the name of a file containing a library, but the
@@ -274,61 +310,59 @@
"""
self.libraries.append (libname)
- def set_libraries (self, libnames):
+ def set_libraries(self, libnames):
"""Set the list of libraries to be included in all links driven by
this compiler object to 'libnames' (a list of strings). This does
not affect any standard system libraries that the linker may
include by default.
"""
- self.libraries = copy (libnames)
+ self.libraries = libnames[:]
- def add_library_dir (self, dir):
+ def add_library_dir(self, dir):
"""Add 'dir' to the list of directories that will be searched for
libraries specified to 'add_library()' and 'set_libraries()'. The
linker will be instructed to search for libraries in the order they
are supplied to 'add_library_dir()' and/or 'set_library_dirs()'.
"""
- self.library_dirs.append (dir)
+ self.library_dirs.append(dir)
- def set_library_dirs (self, dirs):
+ def set_library_dirs(self, dirs):
"""Set the list of library search directories to 'dirs' (a list of
strings). This does not affect any standard library search path
that the linker may search by default.
"""
- self.library_dirs = copy (dirs)
+ self.library_dirs = dirs[:]
-
- def add_runtime_library_dir (self, dir):
+ def add_runtime_library_dir(self, dir):
"""Add 'dir' to the list of directories that will be searched for
shared libraries at runtime.
"""
- self.runtime_library_dirs.append (dir)
+ self.runtime_library_dirs.append(dir)
- def set_runtime_library_dirs (self, dirs):
+ def set_runtime_library_dirs(self, dirs):
"""Set the list of directories to search for shared libraries at
runtime to 'dirs' (a list of strings). This does not affect any
standard search path that the runtime linker may search by
default.
"""
- self.runtime_library_dirs = copy (dirs)
+ self.runtime_library_dirs = dirs[:]
-
- def add_link_object (self, object):
+ def add_link_object(self, object):
"""Add 'object' to the list of object files (or analogues, such as
explicitly named library files or the output of "resource
compilers") to be included in every link driven by this compiler
object.
"""
- self.objects.append (object)
+ self.objects.append(object)
- def set_link_objects (self, objects):
+ def set_link_objects(self, objects):
"""Set the list of object files (or analogues) to be included in
every link to 'objects'. This does not affect any standard object
files that the linker may include by default (such as system
libraries).
"""
- self.objects = copy (objects)
+ self.objects = objects[:]
# -- Private utility methods --------------------------------------
@@ -341,19 +375,19 @@
"""Process arguments and decide which source files to compile."""
if outdir is None:
outdir = self.output_dir
- elif type(outdir) is not StringType:
+ elif not isinstance(outdir, str):
raise TypeError, "'output_dir' must be a string or None"
if macros is None:
macros = self.macros
- elif type(macros) is ListType:
+ elif isinstance(macros, list):
macros = macros + (self.macros or [])
else:
raise TypeError, "'macros' (if supplied) must be a list of tuples"
if incdirs is None:
incdirs = self.include_dirs
- elif type(incdirs) in (ListType, TupleType):
+ elif isinstance(incdirs, (list, tuple)):
incdirs = list(incdirs) + (self.include_dirs or [])
else:
raise TypeError, \
@@ -389,7 +423,7 @@
cc_args[:0] = before
return cc_args
- def _fix_compile_args (self, output_dir, macros, include_dirs):
+ def _fix_compile_args(self, output_dir, macros, include_dirs):
"""Typecheck and fix-up some of the arguments to the 'compile()'
method, and return fixed-up values. Specifically: if 'output_dir'
is None, replaces it with 'self.output_dir'; ensures that 'macros'
@@ -401,19 +435,19 @@
"""
if output_dir is None:
output_dir = self.output_dir
- elif type (output_dir) is not StringType:
+ elif not isinstance(output_dir, str):
raise TypeError, "'output_dir' must be a string or None"
if macros is None:
macros = self.macros
- elif type (macros) is ListType:
+ elif isinstance(macros, list):
macros = macros + (self.macros or [])
else:
raise TypeError, "'macros' (if supplied) must be a list of tuples"
if include_dirs is None:
include_dirs = self.include_dirs
- elif type (include_dirs) in (ListType, TupleType):
+ elif isinstance(include_dirs, (list, tuple)):
include_dirs = list (include_dirs) + (self.include_dirs or [])
else:
raise TypeError, \
@@ -421,44 +455,25 @@
return output_dir, macros, include_dirs
- # _fix_compile_args ()
-
- def _prep_compile(self, sources, output_dir, depends=None):
- """Decide which souce files must be recompiled.
-
- Determine the list of object files corresponding to 'sources',
- and figure out which ones really need to be recompiled.
- Return a list of all object files and a dictionary telling
- which source files can be skipped.
- """
- # Get the list of expected output (object) files
- objects = self.object_filenames(sources, output_dir=output_dir)
- assert len(objects) == len(sources)
-
- # Return an empty dict for the "which source files can be skipped"
- # return value to preserve API compatibility.
- return objects, {}
-
- def _fix_object_args (self, objects, output_dir):
+ def _fix_object_args(self, objects, output_dir):
"""Typecheck and fix up some arguments supplied to various methods.
Specifically: ensure that 'objects' is a list; if output_dir is
None, replace with self.output_dir. Return fixed versions of
'objects' and 'output_dir'.
"""
- if type (objects) not in (ListType, TupleType):
+ if not isinstance(objects, (list, tuple)):
raise TypeError, \
"'objects' must be a list or tuple of strings"
objects = list (objects)
if output_dir is None:
output_dir = self.output_dir
- elif type (output_dir) is not StringType:
+ elif not isinstance(output_dir, str):
raise TypeError, "'output_dir' must be a string or None"
return (objects, output_dir)
-
- def _fix_lib_args (self, libraries, library_dirs, runtime_library_dirs):
+ def _fix_lib_args(self, libraries, library_dirs, runtime_library_dirs):
"""Typecheck and fix up some of the arguments supplied to the
'link_*' methods. Specifically: ensure that all arguments are
lists, and augment them with their permanent versions
@@ -467,7 +482,7 @@
"""
if libraries is None:
libraries = self.libraries
- elif type (libraries) in (ListType, TupleType):
+ elif isinstance(libraries, (list, tuple)):
libraries = list (libraries) + (self.libraries or [])
else:
raise TypeError, \
@@ -475,7 +490,7 @@
if library_dirs is None:
library_dirs = self.library_dirs
- elif type (library_dirs) in (ListType, TupleType):
+ elif isinstance(library_dirs, (list, tuple)):
library_dirs = list (library_dirs) + (self.library_dirs or [])
else:
raise TypeError, \
@@ -483,7 +498,7 @@
if runtime_library_dirs is None:
runtime_library_dirs = self.runtime_library_dirs
- elif type (runtime_library_dirs) in (ListType, TupleType):
+ elif isinstance(runtime_library_dirs, (list, tuple)):
runtime_library_dirs = (list (runtime_library_dirs) +
(self.runtime_library_dirs or []))
else:
@@ -493,10 +508,7 @@
return (libraries, library_dirs, runtime_library_dirs)
- # _fix_lib_args ()
-
-
- def _need_link (self, objects, output_file):
+ def _need_link(self, objects, output_file):
"""Return true if we need to relink the files listed in 'objects'
to recreate 'output_file'.
"""
@@ -509,13 +521,11 @@
newer = newer_group (objects, output_file)
return newer
- # _need_link ()
-
- def detect_language (self, sources):
+ def detect_language(self, sources):
"""Detect the language of a given file, or list of files. Uses
language_map, and language_order to do the job.
"""
- if type(sources) is not ListType:
+ if not isinstance(sources, list):
sources = [sources]
lang = None
index = len(self.language_order)
@@ -531,18 +541,11 @@
pass
return lang
- # detect_language ()
-
# -- Worker methods ------------------------------------------------
# (must be implemented by subclasses)
- def preprocess (self,
- source,
- output_file=None,
- macros=None,
- include_dirs=None,
- extra_preargs=None,
- extra_postargs=None):
+ def preprocess(self, source, output_file=None, macros=None,
+ include_dirs=None, extra_preargs=None, extra_postargs=None):
"""Preprocess a single C/C++ source file, named in 'source'.
Output will be written to file named 'output_file', or stdout if
'output_file' not supplied. 'macros' is a list of macro
@@ -630,12 +633,8 @@
# should implement _compile().
pass
- def create_static_lib (self,
- objects,
- output_libname,
- output_dir=None,
- debug=0,
- target_lang=None):
+ def create_static_lib(self, objects, output_libname, output_dir=None,
+ debug=0, target_lang=None):
"""Link a bunch of stuff together to create a static library file.
The "bunch of stuff" consists of the list of object files supplied
as 'objects', the extra object files supplied to
@@ -660,26 +659,15 @@
"""
pass
-
# values for target_desc parameter in link()
SHARED_OBJECT = "shared_object"
SHARED_LIBRARY = "shared_library"
EXECUTABLE = "executable"
- def link (self,
- target_desc,
- objects,
- output_filename,
- output_dir=None,
- libraries=None,
- library_dirs=None,
- runtime_library_dirs=None,
- export_symbols=None,
- debug=0,
- extra_preargs=None,
- extra_postargs=None,
- build_temp=None,
- target_lang=None):
+ def link(self, target_desc, objects, output_filename, output_dir=None,
+ libraries=None, library_dirs=None, runtime_library_dirs=None,
+ export_symbols=None, debug=0, extra_preargs=None,
+ extra_postargs=None, build_temp=None, target_lang=None):
"""Link a bunch of stuff together to create an executable or
shared library file.
@@ -728,19 +716,11 @@
# Old 'link_*()' methods, rewritten to use the new 'link()' method.
- def link_shared_lib (self,
- objects,
- output_libname,
- output_dir=None,
- libraries=None,
- library_dirs=None,
- runtime_library_dirs=None,
- export_symbols=None,
- debug=0,
- extra_preargs=None,
- extra_postargs=None,
- build_temp=None,
- target_lang=None):
+ def link_shared_lib(self, objects, output_libname, output_dir=None,
+ libraries=None, library_dirs=None,
+ runtime_library_dirs=None, export_symbols=None,
+ debug=0, extra_preargs=None, extra_postargs=None,
+ build_temp=None, target_lang=None):
self.link(CCompiler.SHARED_LIBRARY, objects,
self.library_filename(output_libname, lib_type='shared'),
output_dir,
@@ -749,37 +729,21 @@
extra_preargs, extra_postargs, build_temp, target_lang)
- def link_shared_object (self,
- objects,
- output_filename,
- output_dir=None,
- libraries=None,
- library_dirs=None,
- runtime_library_dirs=None,
- export_symbols=None,
- debug=0,
- extra_preargs=None,
- extra_postargs=None,
- build_temp=None,
- target_lang=None):
+ def link_shared_object(self, objects, output_filename, output_dir=None,
+ libraries=None, library_dirs=None,
+ runtime_library_dirs=None, export_symbols=None,
+ debug=0, extra_preargs=None, extra_postargs=None,
+ build_temp=None, target_lang=None):
self.link(CCompiler.SHARED_OBJECT, objects,
output_filename, output_dir,
libraries, library_dirs, runtime_library_dirs,
export_symbols, debug,
extra_preargs, extra_postargs, build_temp, target_lang)
-
- def link_executable (self,
- objects,
- output_progname,
- output_dir=None,
- libraries=None,
- library_dirs=None,
- runtime_library_dirs=None,
- debug=0,
- extra_preargs=None,
- extra_postargs=None,
- target_lang=None):
+ def link_executable(self, objects, output_progname, output_dir=None,
+ libraries=None, library_dirs=None,
+ runtime_library_dirs=None, debug=0, extra_preargs=None,
+ extra_postargs=None, target_lang=None):
self.link(CCompiler.EXECUTABLE, objects,
self.executable_filename(output_progname), output_dir,
libraries, library_dirs, runtime_library_dirs, None,
@@ -791,29 +755,26 @@
# no appropriate default implementation so subclasses should
# implement all of these.
- def library_dir_option (self, dir):
+ def library_dir_option(self, dir):
"""Return the compiler option to add 'dir' to the list of
directories searched for libraries.
"""
raise NotImplementedError
- def runtime_library_dir_option (self, dir):
+ def runtime_library_dir_option(self, dir):
"""Return the compiler option to add 'dir' to the list of
directories searched for runtime libraries.
"""
raise NotImplementedError
- def library_option (self, lib):
+ def library_option(self, lib):
"""Return the compiler option to add 'dir' to the list of libraries
linked into the shared library or executable.
"""
raise NotImplementedError
- def has_function(self, funcname,
- includes=None,
- include_dirs=None,
- libraries=None,
- library_dirs=None):
+ def has_function(self, funcname, includes=None, include_dirs=None,
+ libraries=None, library_dirs=None):
"""Return a boolean indicating whether funcname is supported on
the current platform. The optional arguments can be used to
augment the compilation environment.
@@ -833,14 +794,16 @@
library_dirs = []
fd, fname = tempfile.mkstemp(".c", funcname, text=True)
f = os.fdopen(fd, "w")
- for incl in includes:
- f.write("""#include "%s"\n""" % incl)
- f.write("""\
+ try:
+ for incl in includes:
+ f.write("""#include "%s"\n""" % incl)
+ f.write("""\
main (int argc, char **argv) {
%s();
}
""" % funcname)
- f.close()
+ finally:
+ f.close()
try:
objects = self.compile([fname], include_dirs=include_dirs)
except CompileError:
@@ -944,28 +907,28 @@
# -- Utility methods -----------------------------------------------
- def announce (self, msg, level=1):
+ def announce(self, msg, level=1):
log.debug(msg)
- def debug_print (self, msg):
+ def debug_print(self, msg):
from distutils.debug import DEBUG
if DEBUG:
print msg
- def warn (self, msg):
- sys.stderr.write ("warning: %s\n" % msg)
+ def warn(self, msg):
+ sys.stderr.write("warning: %s\n" % msg)
- def execute (self, func, args, msg=None, level=1):
+ def execute(self, func, args, msg=None, level=1):
execute(func, args, msg, self.dry_run)
- def spawn (self, cmd):
- spawn (cmd, dry_run=self.dry_run)
+ def spawn(self, cmd):
+ spawn(cmd, dry_run=self.dry_run)
- def move_file (self, src, dst):
- return move_file (src, dst, dry_run=self.dry_run)
+ def move_file(self, src, dst):
+ return move_file(src, dst, dry_run=self.dry_run)
- def mkpath (self, name, mode=0777):
- mkpath (name, mode, dry_run=self.dry_run)
+ def mkpath(self, name, mode=0777):
+ mkpath(name, mode, dry_run=self.dry_run)
# class CCompiler
@@ -988,12 +951,10 @@
# OS name mappings
('posix', 'unix'),
('nt', 'msvc'),
- ('mac', 'mwerks'),
)
def get_default_compiler(osname=None, platform=None):
-
""" Determine the default compiler to use for the given platform.
osname should be one of the standard Python OS names (i.e. the
@@ -1028,8 +989,6 @@
"Mingw32 port of GNU C Compiler for Win32"),
'bcpp': ('bcppcompiler', 'BCPPCompiler',
"Borland C++ Compiler"),
- 'mwerks': ('mwerkscompiler', 'MWerksCompiler',
- "MetroWerks CodeWarrior"),
'emx': ('emxccompiler', 'EMXCCompiler',
"EMX port of GNU C Compiler for OS/2"),
'jython': ('jythoncompiler', 'JythonCompiler',
@@ -1053,11 +1012,7 @@
pretty_printer.print_help("List of available compilers:")
-def new_compiler (plat=None,
- compiler=None,
- verbose=0,
- dry_run=0,
- force=0):
+def new_compiler(plat=None, compiler=None, verbose=0, dry_run=0, force=0):
"""Generate an instance of some CCompiler subclass for the supplied
platform/compiler combination. 'plat' defaults to 'os.name'
(eg. 'posix', 'nt'), and 'compiler' defaults to the default compiler
@@ -1099,10 +1054,10 @@
# XXX The None is necessary to preserve backwards compatibility
# with classes that expect verbose to be the first positional
# argument.
- return klass (None, dry_run, force)
+ return klass(None, dry_run, force)
-def gen_preprocess_options (macros, include_dirs):
+def gen_preprocess_options(macros, include_dirs):
"""Generate C pre-processor options (-D, -U, -I) as used by at least
two types of compilers: the typical Unix compiler and Visual C++.
'macros' is the usual thing, a list of 1- or 2-tuples, where (name,)
@@ -1127,7 +1082,7 @@
pp_opts = []
for macro in macros:
- if not (type (macro) is TupleType and
+ if not (isinstance(macro, tuple) and
1 <= len (macro) <= 2):
raise TypeError, \
("bad macro definition '%s': " +
@@ -1150,27 +1105,27 @@
return pp_opts
-# gen_preprocess_options ()
+def gen_lib_options(compiler, library_dirs, runtime_library_dirs, libraries):
+ """Generate linker options for searching library directories and
+ linking with specific libraries.
-def gen_lib_options (compiler, library_dirs, runtime_library_dirs, libraries):
- """Generate linker options for searching library directories and
- linking with specific libraries. 'libraries' and 'library_dirs' are,
- respectively, lists of library names (not filenames!) and search
- directories. Returns a list of command-line options suitable for use
- with some compiler (depending on the two format strings passed in).
+ 'libraries' and 'library_dirs' are, respectively, lists of library names
+ (not filenames!) and search directories. Returns a list of command-line
+ options suitable for use with some compiler (depending on the two format
+ strings passed in).
"""
lib_opts = []
for dir in library_dirs:
- lib_opts.append (compiler.library_dir_option (dir))
+ lib_opts.append(compiler.library_dir_option(dir))
for dir in runtime_library_dirs:
- opt = compiler.runtime_library_dir_option (dir)
- if type(opt) is ListType:
- lib_opts = lib_opts + opt
+ opt = compiler.runtime_library_dir_option(dir)
+ if isinstance(opt, list):
+ lib_opts.extend(opt)
else:
- lib_opts.append (opt)
+ lib_opts.append(opt)
# XXX it's important that we *not* remove redundant library mentions!
# sometimes you really do have to say "-lfoo -lbar -lfoo" in order to
@@ -1179,17 +1134,15 @@
# pretty nasty way to arrange your C code.
for lib in libraries:
- (lib_dir, lib_name) = os.path.split (lib)
- if lib_dir:
- lib_file = compiler.find_library_file ([lib_dir], lib_name)
- if lib_file:
- lib_opts.append (lib_file)
+ lib_dir, lib_name = os.path.split(lib)
+ if lib_dir != '':
+ lib_file = compiler.find_library_file([lib_dir], lib_name)
+ if lib_file is not None:
+ lib_opts.append(lib_file)
else:
- compiler.warn ("no library file corresponding to "
- "'%s' found (skipping)" % lib)
+ compiler.warn("no library file corresponding to "
+ "'%s' found (skipping)" % lib)
else:
- lib_opts.append (compiler.library_option (lib))
+ lib_opts.append(compiler.library_option(lib))
return lib_opts
-
-# gen_lib_options ()
diff --git a/Lib/distutils/command/bdist_dumb.py b/Lib/distutils/command/bdist_dumb.py
--- a/Lib/distutils/command/bdist_dumb.py
+++ b/Lib/distutils/command/bdist_dumb.py
@@ -4,21 +4,21 @@
distribution -- i.e., just an archive to be unpacked under $prefix or
$exec_prefix)."""
-# This module should be kept compatible with Python 2.1.
-
-__revision__ = "$Id: bdist_dumb.py 61000 2008-02-23 17:40:11Z christian.heimes $"
+__revision__ = "$Id: bdist_dumb.py 77761 2010-01-26 22:46:15Z tarek.ziade $"
import os
+
+from sysconfig import get_python_version
+
+from distutils.util import get_platform
from distutils.core import Command
-from distutils.util import get_platform
from distutils.dir_util import remove_tree, ensure_relative
-from distutils.errors import *
-from distutils.sysconfig import get_python_version
+from distutils.errors import DistutilsPlatformError
from distutils import log
class bdist_dumb (Command):
- description = "create a \"dumb\" built distribution"
+ description = 'create a "dumb" built distribution'
user_options = [('bdist-dir=', 'd',
"temporary directory for creating the distribution"),
@@ -37,6 +37,12 @@
('relative', None,
"build the archive using relative paths"
"(default: false)"),
+ ('owner=', 'u',
+ "Owner name used when creating a tar file"
+ " [default: current user]"),
+ ('group=', 'g',
+ "Group name used when creating a tar file"
+ " [default: current group]"),
]
boolean_options = ['keep-temp', 'skip-build', 'relative']
@@ -55,12 +61,10 @@
self.dist_dir = None
self.skip_build = 0
self.relative = 0
+ self.owner = None
+ self.group = None
- # initialize_options()
-
-
- def finalize_options (self):
-
+ def finalize_options(self):
if self.bdist_dir is None:
bdist_base = self.get_finalized_command('bdist').bdist_base
self.bdist_dir = os.path.join(bdist_base, 'dumb')
@@ -77,11 +81,7 @@
('dist_dir', 'dist_dir'),
('plat_name', 'plat_name'))
- # finalize_options()
-
-
- def run (self):
-
+ def run(self):
if not self.skip_build:
self.run_command('build')
@@ -120,7 +120,8 @@
# Make the archive
filename = self.make_archive(pseudoinstall_root,
- self.format, root_dir=archive_root)
+ self.format, root_dir=archive_root,
+ owner=self.owner, group=self.group)
if self.distribution.has_ext_modules():
pyversion = get_python_version()
else:
@@ -130,7 +131,3 @@
if not self.keep_temp:
remove_tree(self.bdist_dir, dry_run=self.dry_run)
-
- # run()
-
-# class bdist_dumb
diff --git a/Lib/distutils/command/install_scripts.py b/Lib/distutils/command/install_scripts.py
--- a/Lib/distutils/command/install_scripts.py
+++ b/Lib/distutils/command/install_scripts.py
@@ -5,9 +5,7 @@
# contributed by Bastian Kleineidam
-# This module should be kept compatible with Python 2.1.
-
-__revision__ = "$Id: install_scripts.py 37828 2004-11-10 22:23:15Z loewis $"
+__revision__ = "$Id: install_scripts.py 68943 2009-01-25 22:09:10Z tarek.ziade $"
import os
from distutils.core import Command
diff --git a/Lib/distutils/file_util.py b/Lib/distutils/file_util.py
--- a/Lib/distutils/file_util.py
+++ b/Lib/distutils/file_util.py
@@ -3,58 +3,55 @@
Utility functions for operating on single files.
"""
-# This module should be kept compatible with Python 2.1.
-
-__revision__ = "$Id: file_util.py 37828 2004-11-10 22:23:15Z loewis $"
+__revision__ = "$Id: file_util.py 86238 2010-11-06 04:06:18Z eric.araujo $"
import os
from distutils.errors import DistutilsFileError
from distutils import log
# for generating verbose output in 'copy_file()'
-_copy_action = { None: 'copying',
- 'hard': 'hard linking',
- 'sym': 'symbolically linking' }
+_copy_action = {None: 'copying',
+ 'hard': 'hard linking',
+ 'sym': 'symbolically linking'}
-def _copy_file_contents (src, dst, buffer_size=16*1024):
- """Copy the file 'src' to 'dst'; both must be filenames. Any error
- opening either file, reading from 'src', or writing to 'dst', raises
- DistutilsFileError. Data is read/written in chunks of 'buffer_size'
- bytes (default 16k). No attempt is made to handle anything apart from
- regular files.
+def _copy_file_contents(src, dst, buffer_size=16*1024):
+ """Copy the file 'src' to 'dst'.
+
+ Both must be filenames. Any error opening either file, reading from
+ 'src', or writing to 'dst', raises DistutilsFileError. Data is
+ read/written in chunks of 'buffer_size' bytes (default 16k). No attempt
+ is made to handle anything apart from regular files.
"""
# Stolen from shutil module in the standard library, but with
# custom error-handling added.
-
fsrc = None
fdst = None
try:
try:
fsrc = open(src, 'rb')
except os.error, (errno, errstr):
- raise DistutilsFileError, \
- "could not open '%s': %s" % (src, errstr)
+ raise DistutilsFileError("could not open '%s': %s" % (src, errstr))
if os.path.exists(dst):
try:
os.unlink(dst)
except os.error, (errno, errstr):
- raise DistutilsFileError, \
- "could not delete '%s': %s" % (dst, errstr)
+ raise DistutilsFileError(
+ "could not delete '%s': %s" % (dst, errstr))
try:
fdst = open(dst, 'wb')
except os.error, (errno, errstr):
- raise DistutilsFileError, \
- "could not create '%s': %s" % (dst, errstr)
+ raise DistutilsFileError(
+ "could not create '%s': %s" % (dst, errstr))
while 1:
try:
buf = fsrc.read(buffer_size)
except os.error, (errno, errstr):
- raise DistutilsFileError, \
- "could not read from '%s': %s" % (src, errstr)
+ raise DistutilsFileError(
+ "could not read from '%s': %s" % (src, errstr))
if not buf:
break
@@ -62,8 +59,8 @@
try:
fdst.write(buf)
except os.error, (errno, errstr):
- raise DistutilsFileError, \
- "could not write to '%s': %s" % (dst, errstr)
+ raise DistutilsFileError(
+ "could not write to '%s': %s" % (dst, errstr))
finally:
if fdst:
@@ -71,25 +68,18 @@
if fsrc:
fsrc.close()
-# _copy_file_contents()
+def copy_file(src, dst, preserve_mode=1, preserve_times=1, update=0,
+ link=None, verbose=1, dry_run=0):
+ """Copy a file 'src' to 'dst'.
-def copy_file (src, dst,
- preserve_mode=1,
- preserve_times=1,
- update=0,
- link=None,
- verbose=0,
- dry_run=0):
-
- """Copy a file 'src' to 'dst'. If 'dst' is a directory, then 'src' is
- copied there with the same name; otherwise, it must be a filename. (If
- the file exists, it will be ruthlessly clobbered.) If 'preserve_mode'
- is true (the default), the file's mode (type and permission bits, or
- whatever is analogous on the current platform) is copied. If
- 'preserve_times' is true (the default), the last-modified and
- last-access times are copied as well. If 'update' is true, 'src' will
- only be copied if 'dst' does not exist, or if 'dst' does exist but is
- older than 'src'.
+ If 'dst' is a directory, then 'src' is copied there with the same name;
+ otherwise, it must be a filename. (If the file exists, it will be
+ ruthlessly clobbered.) If 'preserve_mode' is true (the default),
+ the file's mode (type and permission bits, or whatever is analogous on
+ the current platform) is copied. If 'preserve_times' is true (the
+ default), the last-modified and last-access times are copied as well.
+ If 'update' is true, 'src' will only be copied if 'dst' does not exist,
+ or if 'dst' does exist but is older than 'src'.
'link' allows you to make hard links (os.link) or symbolic links
(os.symlink) instead of copying: set it to "hard" or "sym"; if it is
@@ -115,8 +105,8 @@
from stat import ST_ATIME, ST_MTIME, ST_MODE, S_IMODE
if not os.path.isfile(src):
- raise DistutilsFileError, \
- "can't copy '%s': doesn't exist or not a regular file" % src
+ raise DistutilsFileError(
+ "can't copy '%s': doesn't exist or not a regular file" % src)
if os.path.isdir(dst):
dir = dst
@@ -125,34 +115,27 @@
dir = os.path.dirname(dst)
if update and not newer(src, dst):
- log.debug("not copying %s (output up-to-date)", src)
+ if verbose >= 1:
+ log.debug("not copying %s (output up-to-date)", src)
return dst, 0
try:
action = _copy_action[link]
except KeyError:
- raise ValueError, \
- "invalid value '%s' for 'link' argument" % link
- if os.path.basename(dst) == os.path.basename(src):
- log.info("%s %s -> %s", action, src, dir)
- else:
- log.info("%s %s -> %s", action, src, dst)
+ raise ValueError("invalid value '%s' for 'link' argument" % link)
+
+ if verbose >= 1:
+ if os.path.basename(dst) == os.path.basename(src):
+ log.info("%s %s -> %s", action, src, dir)
+ else:
+ log.info("%s %s -> %s", action, src, dst)
if dry_run:
return (dst, 1)
- # On Mac OS, use the native file copy routine
- if os.name == 'mac':
- import macostools
- try:
- macostools.copy(src, dst, 0, preserve_times)
- except os.error, exc:
- raise DistutilsFileError, \
- "could not copy '%s' to '%s': %s" % (src, dst, exc[-1])
-
# If linking (hard or symbolic), use the appropriate system call
# (Unix only, of course, but that's the caller's responsibility)
- elif link == 'hard':
+ if link == 'hard':
if not (os.path.exists(dst) and os.path.samefile(src, dst)):
os.link(src, dst)
elif link == 'sym':
@@ -175,17 +158,13 @@
return (dst, 1)
-# copy_file ()
+# XXX I suspect this is Unix-specific -- need porting help!
+def move_file (src, dst, verbose=1, dry_run=0):
+ """Move a file 'src' to 'dst'.
-
-# XXX I suspect this is Unix-specific -- need porting help!
-def move_file (src, dst,
- verbose=0,
- dry_run=0):
-
- """Move a file 'src' to 'dst'. If 'dst' is a directory, the file will
- be moved into it with the same name; otherwise, 'src' is just renamed
- to 'dst'. Return the new full name of the file.
+ If 'dst' is a directory, the file will be moved into it with the same
+ name; otherwise, 'src' is just renamed to 'dst'. Return the new
+ full name of the file.
Handles cross-device moves on Unix using 'copy_file()'. What about
other systems???
@@ -193,26 +172,26 @@
from os.path import exists, isfile, isdir, basename, dirname
import errno
- log.info("moving %s -> %s", src, dst)
+ if verbose >= 1:
+ log.info("moving %s -> %s", src, dst)
if dry_run:
return dst
if not isfile(src):
- raise DistutilsFileError, \
- "can't move '%s': not a regular file" % src
+ raise DistutilsFileError("can't move '%s': not a regular file" % src)
if isdir(dst):
dst = os.path.join(dst, basename(src))
elif exists(dst):
- raise DistutilsFileError, \
- "can't move '%s': destination '%s' already exists" % \
- (src, dst)
+ raise DistutilsFileError(
+ "can't move '%s': destination '%s' already exists" %
+ (src, dst))
if not isdir(dirname(dst)):
- raise DistutilsFileError, \
+ raise DistutilsFileError(
"can't move '%s': destination '%s' not a valid path" % \
- (src, dst)
+ (src, dst))
copy_it = 0
try:
@@ -221,11 +200,11 @@
if num == errno.EXDEV:
copy_it = 1
else:
- raise DistutilsFileError, \
- "couldn't move '%s' to '%s': %s" % (src, dst, msg)
+ raise DistutilsFileError(
+ "couldn't move '%s' to '%s': %s" % (src, dst, msg))
if copy_it:
- copy_file(src, dst)
+ copy_file(src, dst, verbose=verbose)
try:
os.unlink(src)
except os.error, (num, msg):
@@ -233,21 +212,20 @@
os.unlink(dst)
except os.error:
pass
- raise DistutilsFileError, \
+ raise DistutilsFileError(
("couldn't move '%s' to '%s' by copy/delete: " +
- "delete '%s' failed: %s") % \
- (src, dst, src, msg)
-
+ "delete '%s' failed: %s") %
+ (src, dst, src, msg))
return dst
-# move_file ()
-
def write_file (filename, contents):
"""Create a file with the specified name and write 'contents' (a
sequence of strings without line terminators) to it.
"""
f = open(filename, "w")
- for line in contents:
- f.write(line + "\n")
- f.close()
+ try:
+ for line in contents:
+ f.write(line + "\n")
+ finally:
+ f.close()
diff --git a/Lib/distutils/spawn.py b/Lib/distutils/spawn.py
--- a/Lib/distutils/spawn.py
+++ b/Lib/distutils/spawn.py
@@ -6,21 +6,18 @@
executable name.
"""
-# This module should be kept compatible with Python 2.1.
+__revision__ = "$Id: spawn.py 73147 2009-06-02 15:58:43Z tarek.ziade $"
-__revision__ = "$Id: spawn.py 37828 2004-11-10 22:23:15Z loewis $"
+import sys
+import os
-import sys, os, string
-from distutils.errors import *
+from distutils.errors import DistutilsPlatformError, DistutilsExecError
from distutils import log
-def spawn (cmd,
- search_path=1,
- verbose=0,
- dry_run=0):
+def spawn(cmd, search_path=1, verbose=0, dry_run=0):
+ """Run another program, specified as a command list 'cmd', in a new process.
- """Run another program, specified as a command list 'cmd', in a new
- process. 'cmd' is just the argument list for the new process, ie.
+ 'cmd' is just the argument list for the new process, ie.
cmd[0] is the program to run and cmd[1:] are the rest of its arguments.
There is no way to run a program with a name different from that of its
executable.
@@ -45,37 +42,29 @@
raise DistutilsPlatformError, \
"don't know how to spawn programs on platform '%s'" % os.name
-# spawn ()
+def _nt_quote_args(args):
+ """Quote command-line arguments for DOS/Windows conventions.
-
-def _nt_quote_args (args):
- """Quote command-line arguments for DOS/Windows conventions: just
- wraps every argument which contains blanks in double quotes, and
+ Just wraps every argument which contains blanks in double quotes, and
returns a new argument list.
"""
-
# XXX this doesn't seem very robust to me -- but if the Windows guys
# say it'll work, I guess I'll have to accept it. (What if an arg
# contains quotes? What other magic characters, other than spaces,
# have to be escaped? Is there an escaping mechanism other than
# quoting?)
-
- for i in range(len(args)):
- if string.find(args[i], ' ') != -1:
- args[i] = '"%s"' % args[i]
+ for i, arg in enumerate(args):
+ if ' ' in arg:
+ args[i] = '"%s"' % arg
return args
-def _spawn_nt (cmd,
- search_path=1,
- verbose=0,
- dry_run=0):
-
+def _spawn_nt(cmd, search_path=1, verbose=0, dry_run=0):
executable = cmd[0]
cmd = _nt_quote_args(cmd)
if search_path:
# either we find one or it stays the same
executable = find_executable(executable) or executable
- log.info(string.join([executable] + cmd[1:], ' '))
+ log.info(' '.join([executable] + cmd[1:]))
if not dry_run:
# spawn for NT requires a full path to the .exe
try:
@@ -89,18 +78,12 @@
raise DistutilsExecError, \
"command '%s' failed with exit status %d" % (cmd[0], rc)
-
-def _spawn_os2 (cmd,
- search_path=1,
- verbose=0,
- dry_run=0):
-
+def _spawn_os2(cmd, search_path=1, verbose=0, dry_run=0):
executable = cmd[0]
- #cmd = _nt_quote_args(cmd)
if search_path:
# either we find one or it stays the same
executable = find_executable(executable) or executable
- log.info(string.join([executable] + cmd[1:], ' '))
+ log.info(' '.join([executable] + cmd[1:]))
if not dry_run:
# spawnv for OS/2 EMX requires a full path to the .exe
try:
@@ -111,27 +94,20 @@
"command '%s' failed: %s" % (cmd[0], exc[-1])
if rc != 0:
# and this reflects the command running but failing
- print "command '%s' failed with exit status %d" % (cmd[0], rc)
+ log.debug("command '%s' failed with exit status %d" % (cmd[0], rc))
raise DistutilsExecError, \
"command '%s' failed with exit status %d" % (cmd[0], rc)
-def _spawn_posix (cmd,
- search_path=1,
- verbose=0,
- dry_run=0):
-
- log.info(string.join(cmd, ' '))
+def _spawn_posix(cmd, search_path=1, verbose=0, dry_run=0):
+ log.info(' '.join(cmd))
if dry_run:
return
exec_fn = search_path and os.execvp or os.execv
-
pid = os.fork()
- if pid == 0: # in the child
+ if pid == 0: # in the child
try:
- #print "cmd[0] =", cmd[0]
- #print "cmd =", cmd
exec_fn(cmd[0], cmd)
except OSError, e:
sys.stderr.write("unable to execute %s: %s\n" %
@@ -140,14 +116,12 @@
sys.stderr.write("unable to execute %s for unknown reasons" % cmd[0])
os._exit(1)
-
-
- else: # in the parent
+ else: # in the parent
# Loop until the child either exits or is terminated by a signal
# (ie. keep waiting if it's merely stopped)
while 1:
try:
- (pid, status) = os.waitpid(pid, 0)
+ pid, status = os.waitpid(pid, 0)
except OSError, exc:
import errno
if exc.errno == errno.EINTR:
@@ -162,7 +136,7 @@
elif os.WIFEXITED(status):
exit_status = os.WEXITSTATUS(status)
if exit_status == 0:
- return # hey, it succeeded!
+ return # hey, it succeeded!
else:
raise DistutilsExecError, \
"command '%s' failed with exit status %d" % \
@@ -175,8 +149,6 @@
raise DistutilsExecError, \
"unknown error executing '%s': termination status %d" % \
(cmd[0], status)
-# _spawn_posix ()
-
def _spawn_java(cmd,
search_path=1,
@@ -200,17 +172,19 @@
def find_executable(executable, path=None):
- """Try to find 'executable' in the directories listed in 'path' (a
- string listing directories separated by 'os.pathsep'; defaults to
- os.environ['PATH']). Returns the complete filename or None if not
- found.
+ """Tries to find 'executable' in the directories listed in 'path'.
+
+ A string listing directories separated by 'os.pathsep'; defaults to
+ os.environ['PATH']. Returns the complete filename or None if not found.
"""
if path is None:
path = os.environ['PATH']
- paths = string.split(path, os.pathsep)
- (base, ext) = os.path.splitext(executable)
+ paths = path.split(os.pathsep)
+ base, ext = os.path.splitext(executable)
+
if (sys.platform == 'win32' or os.name == 'os2') and (ext != '.exe'):
executable = executable + '.exe'
+
if not os.path.isfile(executable):
for p in paths:
f = os.path.join(p, executable)
@@ -220,5 +194,3 @@
return None
else:
return executable
-
-# find_executable()
diff --git a/Lib/gettext.py b/Lib/gettext.py
--- a/Lib/gettext.py
+++ b/Lib/gettext.py
@@ -468,15 +468,15 @@
if fallback:
return NullTranslations()
raise IOError(ENOENT, 'No translation file found for domain', domain)
- # TBD: do we need to worry about the file pointer getting collected?
# Avoid opening, reading, and parsing the .mo file after it's been done
# once.
result = None
for mofile in mofiles:
- key = os.path.abspath(mofile)
+ key = (class_, os.path.abspath(mofile))
t = _translations.get(key)
if t is None:
- t = _translations.setdefault(key, class_(open(mofile, 'rb')))
+ with open(mofile, 'rb') as fp:
+ t = _translations.setdefault(key, class_(fp))
# Copy the translation object to allow setting fallbacks and
# output charset. All other instance data is shared with the
# cached object.
diff --git a/Lib/mailbox.py b/Lib/mailbox.py
--- a/Lib/mailbox.py
+++ b/Lib/mailbox.py
@@ -243,14 +243,21 @@
else:
raise NoSuchMailboxError(self._path)
self._toc = {}
+ self._last_read = None # Records last time we read cur/new
+ # NOTE: we manually invalidate _last_read each time we do any
+ # modifications ourselves, otherwise we might get tripped up by
+ # bogus mtime behaviour on some systems (see issue #6896).
def add(self, message):
"""Add message and return assigned key."""
tmp_file = self._create_tmp()
try:
self._dump_message(message, tmp_file)
- finally:
- _sync_close(tmp_file)
+ except BaseException:
+ tmp_file.close()
+ os.remove(tmp_file.name)
+ raise
+ _sync_close(tmp_file)
if isinstance(message, MaildirMessage):
subdir = message.get_subdir()
suffix = self.colon + message.get_info()
@@ -276,11 +283,15 @@
raise
if isinstance(message, MaildirMessage):
os.utime(dest, (os.path.getatime(dest), message.get_date()))
+ # Invalidate cached toc
+ self._last_read = None
return uniq
def remove(self, key):
"""Remove the keyed message; raise KeyError if it doesn't exist."""
os.remove(os.path.join(self._path, self._lookup(key)))
+ # Invalidate cached toc (only on success)
+ self._last_read = None
def discard(self, key):
"""If the keyed message exists, remove it."""
@@ -315,6 +326,8 @@
if isinstance(message, MaildirMessage):
os.utime(new_path, (os.path.getatime(new_path),
message.get_date()))
+ # Invalidate cached toc
+ self._last_read = None
def get_message(self, key):
"""Return a Message representation or raise a KeyError."""
@@ -369,7 +382,9 @@
def flush(self):
"""Write any pending changes to disk."""
- return # Maildir changes are always written immediately.
+ # Maildir changes are always written immediately, so there's nothing
+ # to do except invalidate our cached toc.
+ self._last_read = None
def lock(self):
"""Lock the mailbox."""
@@ -467,16 +482,37 @@
def _refresh(self):
"""Update table of contents mapping."""
+ if self._last_read is not None:
+ for subdir in ('new', 'cur'):
+ mtime = os.path.getmtime(os.path.join(self._path, subdir))
+ if mtime > self._last_read:
+ break
+ else:
+ return
+
+ # We record the current time - 1sec so that, if _refresh() is called
+ # again in the same second, we will always re-read the mailbox
+ # just in case it's been modified. (os.path.mtime() only has
+ # 1sec resolution.) This results in a few unnecessary re-reads
+ # when _refresh() is called multiple times in the same second,
+ # but once the clock ticks over, we will only re-read as needed.
+ now = time.time() - 1
+
self._toc = {}
- for subdir in ('new', 'cur'):
- subdir_path = os.path.join(self._path, subdir)
- for entry in os.listdir(subdir_path):
- p = os.path.join(subdir_path, entry)
+ def update_dir (subdir):
+ path = os.path.join(self._path, subdir)
+ for entry in os.listdir(path):
+ p = os.path.join(path, entry)
if os.path.isdir(p):
continue
uniq = entry.split(self.colon)[0]
self._toc[uniq] = os.path.join(subdir, entry)
+ update_dir('new')
+ update_dir('cur')
+
+ self._last_read = now
+
def _lookup(self, key):
"""Use TOC to return subpath for given key, or raise a KeyError."""
try:
@@ -518,7 +554,7 @@
f = open(self._path, 'wb+')
else:
raise NoSuchMailboxError(self._path)
- elif e.errno == errno.EACCES:
+ elif e.errno in (errno.EACCES, errno.EROFS):
f = open(self._path, 'rb')
else:
raise
@@ -667,9 +703,14 @@
def _append_message(self, message):
"""Append message to mailbox and return (start, stop) offsets."""
self._file.seek(0, 2)
- self._pre_message_hook(self._file)
- offsets = self._install_message(message)
- self._post_message_hook(self._file)
+ before = self._file.tell()
+ try:
+ self._pre_message_hook(self._file)
+ offsets = self._install_message(message)
+ self._post_message_hook(self._file)
+ except BaseException:
+ self._file.truncate(before)
+ raise
self._file.flush()
self._file_length = self._file.tell() # Record current length of mailbox
return offsets
@@ -835,18 +876,29 @@
new_key = max(keys) + 1
new_path = os.path.join(self._path, str(new_key))
f = _create_carefully(new_path)
+ closed = False
try:
if self._locked:
_lock_file(f)
try:
- self._dump_message(message, f)
+ try:
+ self._dump_message(message, f)
+ except BaseException:
+ # Unlock and close so it can be deleted on Windows
+ if self._locked:
+ _unlock_file(f)
+ _sync_close(f)
+ closed = True
+ os.remove(new_path)
+ raise
if isinstance(message, MHMessage):
self._dump_sequences(message, new_key)
finally:
if self._locked:
_unlock_file(f)
finally:
- _sync_close(f)
+ if not closed:
+ _sync_close(f)
return new_key
def remove(self, key):
@@ -859,17 +911,9 @@
raise KeyError('No message with key: %s' % key)
else:
raise
- try:
- if self._locked:
- _lock_file(f)
- try:
- f.close()
- os.remove(os.path.join(self._path, str(key)))
- finally:
- if self._locked:
- _unlock_file(f)
- finally:
+ else:
f.close()
+ os.remove(path)
def __setitem__(self, key, message):
"""Replace the keyed message; raise KeyError if it doesn't exist."""
@@ -1863,7 +1907,7 @@
try:
fcntl.lockf(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError, e:
- if e.errno in (errno.EAGAIN, errno.EACCES):
+ if e.errno in (errno.EAGAIN, errno.EACCES, errno.EROFS):
raise ExternalClashError('lockf: lock unavailable: %s' %
f.name)
else:
@@ -1873,7 +1917,7 @@
pre_lock = _create_temporary(f.name + '.lock')
pre_lock.close()
except IOError, e:
- if e.errno == errno.EACCES:
+ if e.errno in (errno.EACCES, errno.EROFS):
return # Without write access, just skip dotlocking.
else:
raise
diff --git a/Lib/netrc.py b/Lib/netrc.py
--- a/Lib/netrc.py
+++ b/Lib/netrc.py
@@ -27,16 +27,23 @@
file = os.path.join(os.environ['HOME'], ".netrc")
except KeyError:
raise IOError("Could not find .netrc: $HOME is not set")
- fp = open(file)
self.hosts = {}
self.macros = {}
+ with open(file) as fp:
+ self._parse(file, fp)
+
+ def _parse(self, file, fp):
lexer = shlex.shlex(fp)
lexer.wordchars += r"""!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~"""
+ lexer.commenters = lexer.commenters.replace('#', '')
while 1:
# Look for a machine, default, or macdef top-level keyword
toplevel = tt = lexer.get_token()
if not tt:
break
+ elif tt[0] == '#':
+ fp.readline();
+ continue;
elif tt == 'machine':
entryname = lexer.get_token()
elif tt == 'default':
diff --git a/Lib/robotparser.py b/Lib/robotparser.py
--- a/Lib/robotparser.py
+++ b/Lib/robotparser.py
@@ -133,7 +133,12 @@
return True
# search for given user agent matches
# the first match counts
- url = urllib.quote(urlparse.urlparse(urllib.unquote(url))[2]) or "/"
+ parsed_url = urlparse.urlparse(urllib.unquote(url))
+ url = urlparse.urlunparse(('', '', parsed_url.path,
+ parsed_url.params, parsed_url.query, parsed_url.fragment))
+ url = urllib.quote(url)
+ if not url:
+ url = "/"
for entry in self.entries:
if entry.applies_to(useragent):
return entry.allowance(url)
diff --git a/Lib/test/list_tests.py b/Lib/test/list_tests.py
--- a/Lib/test/list_tests.py
+++ b/Lib/test/list_tests.py
@@ -330,7 +330,7 @@
self.assertRaises(BadExc, d.remove, 'c')
for x, y in zip(d, e):
# verify that original order and values are retained.
- self.assert_(x is y)
+ self.assertIs(x, y)
def test_count(self):
a = self.type2test([0, 1, 2])*3
@@ -419,7 +419,7 @@
self.assertRaises(TypeError, u.reverse, 42)
def test_sort(self):
- with test_support._check_py3k_warnings(
+ with test_support.check_py3k_warnings(
("the cmp argument is not supported", DeprecationWarning)):
self._test_sort()
@@ -466,7 +466,7 @@
u = self.type2test([0, 1])
u2 = u
u += [2, 3]
- self.assert_(u is u2)
+ self.assertIs(u, u2)
u = self.type2test("spam")
u += "eggs"
diff --git a/Lib/test/pickletester.py b/Lib/test/pickletester.py
--- a/Lib/test/pickletester.py
+++ b/Lib/test/pickletester.py
@@ -1,6 +1,8 @@
import unittest
import pickle
import cPickle
+import StringIO
+import cStringIO
import pickletools
import copy_reg
@@ -12,6 +14,42 @@
assert pickle.HIGHEST_PROTOCOL == cPickle.HIGHEST_PROTOCOL == 2
protocols = range(pickle.HIGHEST_PROTOCOL + 1)
+# Copy of test.test_support.run_with_locale. This is needed to support Python
+# 2.4, which didn't include it. This is all to support test_xpickle, which
+# bounces pickled objects through older Python versions to test backwards
+# compatibility.
+def run_with_locale(catstr, *locales):
+ def decorator(func):
+ def inner(*args, **kwds):
+ try:
+ import locale
+ category = getattr(locale, catstr)
+ orig_locale = locale.setlocale(category)
+ except AttributeError:
+ # if the test author gives us an invalid category string
+ raise
+ except:
+ # cannot retrieve original locale, so do nothing
+ locale = orig_locale = None
+ else:
+ for loc in locales:
+ try:
+ locale.setlocale(category, loc)
+ break
+ except:
+ pass
+
+ # now run the function, resetting the locale on exceptions
+ try:
+ return func(*args, **kwds)
+ finally:
+ if locale and orig_locale:
+ locale.setlocale(category, orig_locale)
+ inner.func_name = func.func_name
+ inner.__doc__ = func.__doc__
+ return inner
+ return decorator
+
# Return True if opcode code appears in the pickle, else False.
def opcode_in_pickle(code, pickle):
@@ -408,12 +446,11 @@
# is a mystery. cPickle also suppresses PUT for objects with a refcount
# of 1.
def dont_test_disassembly(self):
- from cStringIO import StringIO
from pickletools import dis
for proto, expected in (0, DATA0_DIS), (1, DATA1_DIS):
s = self.dumps(self._testdata, proto)
- filelike = StringIO()
+ filelike = cStringIO.StringIO()
dis(s, out=filelike)
got = filelike.getvalue()
self.assertEqual(expected, got)
@@ -424,9 +461,18 @@
for proto in protocols:
s = self.dumps(l, proto)
x = self.loads(s)
- self.assertEqual(x, l)
- self.assertEqual(x, x[0])
- self.assertEqual(id(x), id(x[0]))
+ self.assertEqual(len(x), 1)
+ self.assertTrue(x is x[0])
+
+ def test_recursive_tuple(self):
+ t = ([],)
+ t[0].append(t)
+ for proto in protocols:
+ s = self.dumps(t, proto)
+ x = self.loads(s)
+ self.assertEqual(len(x), 1)
+ self.assertEqual(len(x[0]), 1)
+ self.assertTrue(x is x[0][0])
def test_recursive_dict(self):
d = {}
@@ -434,9 +480,8 @@
for proto in protocols:
s = self.dumps(d, proto)
x = self.loads(s)
- self.assertEqual(x, d)
- self.assertEqual(x[1], x)
- self.assertEqual(id(x[1]), id(x))
+ self.assertEqual(x.keys(), [1])
+ self.assertTrue(x[1] is x)
def test_recursive_inst(self):
i = C()
@@ -444,9 +489,8 @@
for proto in protocols:
s = self.dumps(i, 2)
x = self.loads(s)
- self.assertEqual(x, i)
- self.assertEqual(x.attr, x)
- self.assertEqual(id(x.attr), id(x))
+ self.assertEqual(dir(x), dir(i))
+ self.assertTrue(x.attr is x)
def test_recursive_multi(self):
l = []
@@ -457,12 +501,10 @@
for proto in protocols:
s = self.dumps(l, proto)
x = self.loads(s)
- self.assertEqual(x, l)
- self.assertEqual(x[0], i)
- self.assertEqual(x[0].attr, d)
- self.assertEqual(x[0].attr[1], x)
- self.assertEqual(x[0].attr[1][0], i)
- self.assertEqual(x[0].attr[1][0].attr, d)
+ self.assertEqual(len(x), 1)
+ self.assertEqual(dir(x[0]), dir(i))
+ self.assertEqual(x[0].attr.keys(), [1])
+ self.assertTrue(x[0].attr[1] is x)
def test_garyp(self):
self.assertRaises(self.error, self.loads, 'garyp')
@@ -484,14 +526,21 @@
if have_unicode:
def test_unicode(self):
- endcases = [unicode(''), unicode('<\\u>'), unicode('<\\\u1234>'),
- unicode('<\n>'), unicode('<\\>')]
+ endcases = [u'', u'<\\u>', u'<\\\u1234>', u'<\n>',
+ u'<\\>', u'<\\\U00012345>']
for proto in protocols:
for u in endcases:
p = self.dumps(u, proto)
u2 = self.loads(p)
self.assertEqual(u2, u)
+ def test_unicode_high_plane(self):
+ t = u'\U00012345'
+ for proto in protocols:
+ p = self.dumps(t, proto)
+ t2 = self.loads(p)
+ self.assertEqual(t2, t)
+
def test_ints(self):
import sys
for proto in protocols:
@@ -534,6 +583,21 @@
got = self.loads(p)
self.assertEqual(n, got)
+ def test_float(self):
+ test_values = [0.0, 4.94e-324, 1e-310, 7e-308, 6.626e-34, 0.1, 0.5,
+ 3.14, 263.44582062374053, 6.022e23, 1e30]
+ test_values = test_values + [-x for x in test_values]
+ for proto in protocols:
+ for value in test_values:
+ pickle = self.dumps(value, proto)
+ got = self.loads(pickle)
+ self.assertEqual(value, got)
+
+ @run_with_locale('LC_ALL', 'de_DE', 'fr_FR')
+ def test_float_format(self):
+ # make sure that floats are formatted locale independent
+ self.assertEqual(self.dumps(1.2)[0:3], 'F1.')
+
def test_reduce(self):
pass
@@ -583,7 +647,7 @@
try:
self.loads(badpickle)
except ValueError, detail:
- self.failUnless(str(detail).startswith(
+ self.assertTrue(str(detail).startswith(
"unsupported pickle protocol"))
else:
self.fail("expected bad protocol number to raise ValueError")
@@ -655,7 +719,7 @@
for x in None, False, True:
s = self.dumps(x, proto)
y = self.loads(s)
- self.assert_(x is y, (proto, x, s, y))
+ self.assertTrue(x is y, (proto, x, s, y))
expected = expected_opcode[proto, x]
self.assertEqual(opcode_in_pickle(expected, s), True)
@@ -705,8 +769,8 @@
# Dump using protocol 1 for comparison.
s1 = self.dumps(x, 1)
- self.assert_(__name__ in s1)
- self.assert_("MyList" in s1)
+ self.assertIn(__name__, s1)
+ self.assertIn("MyList", s1)
self.assertEqual(opcode_in_pickle(opcode, s1), False)
y = self.loads(s1)
@@ -715,8 +779,8 @@
# Dump using protocol 2 for test.
s2 = self.dumps(x, 2)
- self.assert_(__name__ not in s2)
- self.assert_("MyList" not in s2)
+ self.assertNotIn(__name__, s2)
+ self.assertNotIn("MyList", s2)
self.assertEqual(opcode_in_pickle(opcode, s2), True)
y = self.loads(s2)
@@ -760,7 +824,7 @@
if proto == 0:
self.assertEqual(num_appends, 0)
else:
- self.failUnless(num_appends >= 2)
+ self.assertTrue(num_appends >= 2)
def test_dict_chunking(self):
n = 10 # too small to chunk
@@ -782,7 +846,7 @@
if proto == 0:
self.assertEqual(num_setitems, 0)
else:
- self.failUnless(num_setitems >= 2)
+ self.assertTrue(num_setitems >= 2)
def test_simple_newobj(self):
x = object.__new__(SimpleNewObj) # avoid __init__
@@ -806,7 +870,7 @@
self.assertEqual(x.bar, y.bar)
def test_reduce_overrides_default_reduce_ex(self):
- for proto in 0, 1, 2:
+ for proto in protocols:
x = REX_one()
self.assertEqual(x._reduce_called, 0)
s = self.dumps(x, proto)
@@ -815,7 +879,7 @@
self.assertEqual(y._reduce_called, 0)
def test_reduce_ex_called(self):
- for proto in 0, 1, 2:
+ for proto in protocols:
x = REX_two()
self.assertEqual(x._proto, None)
s = self.dumps(x, proto)
@@ -824,7 +888,7 @@
self.assertEqual(y._proto, None)
def test_reduce_ex_overrides_reduce(self):
- for proto in 0, 1, 2:
+ for proto in protocols:
x = REX_three()
self.assertEqual(x._proto, None)
s = self.dumps(x, proto)
@@ -832,6 +896,76 @@
y = self.loads(s)
self.assertEqual(y._proto, None)
+ def test_reduce_ex_calls_base(self):
+ for proto in protocols:
+ x = REX_four()
+ self.assertEqual(x._proto, None)
+ s = self.dumps(x, proto)
+ self.assertEqual(x._proto, proto)
+ y = self.loads(s)
+ self.assertEqual(y._proto, proto)
+
+ def test_reduce_calls_base(self):
+ for proto in protocols:
+ x = REX_five()
+ self.assertEqual(x._reduce_called, 0)
+ s = self.dumps(x, proto)
+ self.assertEqual(x._reduce_called, 1)
+ y = self.loads(s)
+ self.assertEqual(y._reduce_called, 1)
+
+ def test_reduce_bad_iterator(self):
+ # Issue4176: crash when 4th and 5th items of __reduce__()
+ # are not iterators
+ class C(object):
+ def __reduce__(self):
+ # 4th item is not an iterator
+ return list, (), None, [], None
+ class D(object):
+ def __reduce__(self):
+ # 5th item is not an iterator
+ return dict, (), None, None, []
+
+ # Protocol 0 is less strict and also accept iterables.
+ for proto in protocols:
+ try:
+ self.dumps(C(), proto)
+ except (AttributeError, pickle.PickleError, cPickle.PickleError):
+ pass
+ try:
+ self.dumps(D(), proto)
+ except (AttributeError, pickle.PickleError, cPickle.PickleError):
+ pass
+
+ def test_many_puts_and_gets(self):
+ # Test that internal data structures correctly deal with lots of
+ # puts/gets.
+ keys = ("aaa" + str(i) for i in xrange(100))
+ large_dict = dict((k, [4, 5, 6]) for k in keys)
+ obj = [dict(large_dict), dict(large_dict), dict(large_dict)]
+
+ for proto in protocols:
+ dumped = self.dumps(obj, proto)
+ loaded = self.loads(dumped)
+ self.assertEqual(loaded, obj,
+ "Failed protocol %d: %r != %r"
+ % (proto, obj, loaded))
+
+ def test_attribute_name_interning(self):
+ # Test that attribute names of pickled objects are interned when
+ # unpickling.
+ for proto in protocols:
+ x = C()
+ x.foo = 42
+ x.bar = "hello"
+ s = self.dumps(x, proto)
+ y = self.loads(s)
+ x_keys = sorted(x.__dict__)
+ y_keys = sorted(y.__dict__)
+ for x_key, y_key in zip(x_keys, y_keys):
+ self.assertIs(x_key, y_key)
+
+
# Test classes for reduce_ex
class REX_one(object):
@@ -856,6 +990,20 @@
def __reduce__(self):
raise TestFailed, "This __reduce__ shouldn't be called"
+class REX_four(object):
+ _proto = None
+ def __reduce_ex__(self, proto):
+ self._proto = proto
+ return object.__reduce_ex__(self, proto)
+ # Calling base class method should succeed
+
+class REX_five(object):
+ _reduce_called = 0
+ def __reduce__(self):
+ self._reduce_called = 1
+ return object.__reduce__(self)
+ # This one used to fail with infinite recursion
+
# Test classes for newobj
class MyInt(int):
@@ -919,10 +1067,50 @@
finally:
os.remove(TESTFN)
+ def test_load_from_and_dump_to_file(self):
+ stream = cStringIO.StringIO()
+ data = [123, {}, 124]
+ self.module.dump(data, stream)
+ stream.seek(0)
+ unpickled = self.module.load(stream)
+ self.assertEqual(unpickled, data)
+
def test_highest_protocol(self):
# Of course this needs to be changed when HIGHEST_PROTOCOL changes.
self.assertEqual(self.module.HIGHEST_PROTOCOL, 2)
+ def test_callapi(self):
+ f = cStringIO.StringIO()
+ # With and without keyword arguments
+ self.module.dump(123, f, -1)
+ self.module.dump(123, file=f, protocol=-1)
+ self.module.dumps(123, -1)
+ self.module.dumps(123, protocol=-1)
+ self.module.Pickler(f, -1)
+ self.module.Pickler(f, protocol=-1)
+
+ def test_incomplete_input(self):
+ s = StringIO.StringIO("X''.")
+ self.assertRaises(EOFError, self.module.load, s)
+
+ def test_restricted(self):
+ # issue7128: cPickle failed in restricted mode
+ builtins = {self.module.__name__: self.module,
+ '__import__': __import__}
+ d = {}
+ teststr = "def f(): {0}.dumps(0)".format(self.module.__name__)
+ exec teststr in {'__builtins__': builtins}, d
+ d['f']()
+
+ def test_bad_input(self):
+ # Test issue4298
+ s = '\x58\0\0\0\x54'
+ self.assertRaises(EOFError, self.module.loads, s)
+ # Test issue7455
+ s = '0'
+ # XXX Why doesn't pickle raise UnpicklingError?
+ self.assertRaises((IndexError, cPickle.UnpicklingError),
+ self.module.loads, s)
class AbstractPersistentPicklerTests(unittest.TestCase):
@@ -958,3 +1146,116 @@
self.assertEqual(self.loads(self.dumps(L, 1)), L)
self.assertEqual(self.id_count, 5)
self.assertEqual(self.load_count, 5)
+
+class AbstractPicklerUnpicklerObjectTests(unittest.TestCase):
+
+ pickler_class = None
+ unpickler_class = None
+
+ def setUp(self):
+ assert self.pickler_class
+ assert self.unpickler_class
+
+ def test_clear_pickler_memo(self):
+ # To test whether clear_memo() has any effect, we pickle an object,
+ # then pickle it again without clearing the memo; the two serialized
+ # forms should be different. If we clear_memo() and then pickle the
+ # object again, the third serialized form should be identical to the
+ # first one we obtained.
+ data = ["abcdefg", "abcdefg", 44]
+ f = cStringIO.StringIO()
+ pickler = self.pickler_class(f)
+
+ pickler.dump(data)
+ first_pickled = f.getvalue()
+
+ # Reset StringIO object.
+ f.seek(0)
+ f.truncate()
+
+ pickler.dump(data)
+ second_pickled = f.getvalue()
+
+ # Reset the Pickler and StringIO objects.
+ pickler.clear_memo()
+ f.seek(0)
+ f.truncate()
+
+ pickler.dump(data)
+ third_pickled = f.getvalue()
+
+ self.assertNotEqual(first_pickled, second_pickled)
+ self.assertEqual(first_pickled, third_pickled)
+
+ def test_priming_pickler_memo(self):
+ # Verify that we can set the Pickler's memo attribute.
+ data = ["abcdefg", "abcdefg", 44]
+ f = cStringIO.StringIO()
+ pickler = self.pickler_class(f)
+
+ pickler.dump(data)
+ first_pickled = f.getvalue()
+
+ f = cStringIO.StringIO()
+ primed = self.pickler_class(f)
+ primed.memo = pickler.memo
+
+ primed.dump(data)
+ primed_pickled = f.getvalue()
+
+ self.assertNotEqual(first_pickled, primed_pickled)
+
+ def test_priming_unpickler_memo(self):
+ # Verify that we can set the Unpickler's memo attribute.
+ data = ["abcdefg", "abcdefg", 44]
+ f = cStringIO.StringIO()
+ pickler = self.pickler_class(f)
+
+ pickler.dump(data)
+ first_pickled = f.getvalue()
+
+ f = cStringIO.StringIO()
+ primed = self.pickler_class(f)
+ primed.memo = pickler.memo
+
+ primed.dump(data)
+ primed_pickled = f.getvalue()
+
+ unpickler = self.unpickler_class(cStringIO.StringIO(first_pickled))
+ unpickled_data1 = unpickler.load()
+
+ self.assertEqual(unpickled_data1, data)
+
+ primed = self.unpickler_class(cStringIO.StringIO(primed_pickled))
+ primed.memo = unpickler.memo
+ unpickled_data2 = primed.load()
+
+ primed.memo.clear()
+
+ self.assertEqual(unpickled_data2, data)
+ self.assertTrue(unpickled_data2 is unpickled_data1)
+
+ def test_reusing_unpickler_objects(self):
+ data1 = ["abcdefg", "abcdefg", 44]
+ f = cStringIO.StringIO()
+ pickler = self.pickler_class(f)
+ pickler.dump(data1)
+ pickled1 = f.getvalue()
+
+ data2 = ["abcdefg", 44, 44]
+ f = cStringIO.StringIO()
+ pickler = self.pickler_class(f)
+ pickler.dump(data2)
+ pickled2 = f.getvalue()
+
+ f = cStringIO.StringIO()
+ f.write(pickled1)
+ f.seek(0)
+ unpickler = self.unpickler_class(f)
+ self.assertEqual(unpickler.load(), data1)
+
+ f.seek(0)
+ f.truncate()
+ f.write(pickled2)
+ f.seek(0)
+ self.assertEqual(unpickler.load(), data2)
diff --git a/Lib/test/test_array.py b/Lib/test/test_array.py
--- a/Lib/test/test_array.py
+++ b/Lib/test/test_array.py
@@ -18,7 +18,7 @@
class ArraySubclassWithKwargs(array.array):
def __init__(self, typecode, newarg=None):
- array.array.__init__(typecode)
+ array.array.__init__(self, typecode)
tests = [] # list to accumulate all tests
typecodes = "cubBhHiIlLfd"
@@ -52,7 +52,7 @@
def test_constructor(self):
a = array.array(self.typecode)
self.assertEqual(a.typecode, self.typecode)
- self.assert_(a.itemsize>=self.minitemsize)
+ self.assertTrue(a.itemsize>=self.minitemsize)
self.assertRaises(TypeError, array.array, self.typecode, None)
def test_len(self):
@@ -67,10 +67,10 @@
a = array.array(self.typecode, self.example)
self.assertRaises(TypeError, a.buffer_info, 42)
bi = a.buffer_info()
- self.assert_(isinstance(bi, tuple))
+ self.assertIsInstance(bi, tuple)
self.assertEqual(len(bi), 2)
- self.assert_(isinstance(bi[0], (int, long)))
- self.assert_(isinstance(bi[1], int))
+ self.assertIsInstance(bi[0], (int, long))
+ self.assertIsInstance(bi[1], int)
self.assertEqual(bi[1], len(a))
def test_byteswap(self):
@@ -207,6 +207,25 @@
f.close()
test_support.unlink(test_support.TESTFN)
+ def test_filewrite(self):
+ a = array.array(self.typecode, 2*self.example)
+ f = open(test_support.TESTFN, 'wb')
+ try:
+ f.write(a)
+ f.close()
+ b = array.array(self.typecode)
+ f = open(test_support.TESTFN, 'rb')
+ b.fromfile(f, len(self.example))
+ self.assertEqual(b, array.array(self.typecode, self.example))
+ self.assertNotEqual(a, b)
+ b.fromfile(f, len(self.example))
+ self.assertEqual(a, b)
+ f.close()
+ finally:
+ if not f.closed:
+ f.close()
+ test_support.unlink(test_support.TESTFN)
+
def test_tofromlist(self):
a = array.array(self.typecode, 2*self.example)
b = array.array(self.typecode)
@@ -260,39 +279,39 @@
def test_cmp(self):
a = array.array(self.typecode, self.example)
- self.assert_((a == 42) is False)
- self.assert_((a != 42) is True)
+ self.assertTrue((a == 42) is False)
+ self.assertTrue((a != 42) is True)
- self.assert_((a == a) is True)
- self.assert_((a != a) is False)
- self.assert_((a < a) is False)
- self.assert_((a <= a) is True)
- self.assert_((a > a) is False)
- self.assert_((a >= a) is True)
+ self.assertTrue((a == a) is True)
+ self.assertTrue((a != a) is False)
+ self.assertTrue((a < a) is False)
+ self.assertTrue((a <= a) is True)
+ self.assertTrue((a > a) is False)
+ self.assertTrue((a >= a) is True)
al = array.array(self.typecode, self.smallerexample)
ab = array.array(self.typecode, self.biggerexample)
- self.assert_((a == 2*a) is False)
- self.assert_((a != 2*a) is True)
- self.assert_((a < 2*a) is True)
- self.assert_((a <= 2*a) is True)
- self.assert_((a > 2*a) is False)
- self.assert_((a >= 2*a) is False)
+ self.assertTrue((a == 2*a) is False)
+ self.assertTrue((a != 2*a) is True)
+ self.assertTrue((a < 2*a) is True)
+ self.assertTrue((a <= 2*a) is True)
+ self.assertTrue((a > 2*a) is False)
+ self.assertTrue((a >= 2*a) is False)
- self.assert_((a == al) is False)
- self.assert_((a != al) is True)
- self.assert_((a < al) is False)
- self.assert_((a <= al) is False)
- self.assert_((a > al) is True)
- self.assert_((a >= al) is True)
+ self.assertTrue((a == al) is False)
+ self.assertTrue((a != al) is True)
+ self.assertTrue((a < al) is False)
+ self.assertTrue((a <= al) is False)
+ self.assertTrue((a > al) is True)
+ self.assertTrue((a >= al) is True)
- self.assert_((a == ab) is False)
- self.assert_((a != ab) is True)
- self.assert_((a < ab) is True)
- self.assert_((a <= ab) is True)
- self.assert_((a > ab) is False)
- self.assert_((a >= ab) is False)
+ self.assertTrue((a == ab) is False)
+ self.assertTrue((a != ab) is True)
+ self.assertTrue((a < ab) is True)
+ self.assertTrue((a <= ab) is True)
+ self.assertTrue((a > ab) is False)
+ self.assertTrue((a >= ab) is False)
def test_add(self):
a = array.array(self.typecode, self.example) \
@@ -314,11 +333,17 @@
a = array.array(self.typecode, self.example[::-1])
b = a
a += array.array(self.typecode, 2*self.example)
- self.assert_(a is b)
+ self.assertTrue(a is b)
self.assertEqual(
a,
array.array(self.typecode, self.example[::-1]+2*self.example)
)
+ a = array.array(self.typecode, self.example)
+ a += a
+ self.assertEqual(
+ a,
+ array.array(self.typecode, self.example + self.example)
+ )
b = array.array(self.badtypecode())
if test_support.is_jython:
@@ -363,22 +388,22 @@
b = a
a *= 5
- self.assert_(a is b)
+ self.assertTrue(a is b)
self.assertEqual(
a,
array.array(self.typecode, 5*self.example)
)
a *= 0
- self.assert_(a is b)
+ self.assertTrue(a is b)
self.assertEqual(a, array.array(self.typecode))
a *= 1000
- self.assert_(a is b)
+ self.assertTrue(a is b)
self.assertEqual(a, array.array(self.typecode))
a *= -1
- self.assert_(a is b)
+ self.assertTrue(a is b)
self.assertEqual(a, array.array(self.typecode))
a = array.array(self.typecode, self.example)
@@ -642,11 +667,11 @@
data.reverse()
L[start:stop:step] = data
a[start:stop:step] = array.array(self.typecode, data)
- self.assertEquals(a, array.array(self.typecode, L))
+ self.assertEqual(a, array.array(self.typecode, L))
del L[start:stop:step]
del a[start:stop:step]
- self.assertEquals(a, array.array(self.typecode, L))
+ self.assertEqual(a, array.array(self.typecode, L))
def test_index(self):
example = 2*self.example
@@ -725,6 +750,13 @@
array.array(self.typecode, self.example+self.example[::-1])
)
+ a = array.array(self.typecode, self.example)
+ a.extend(a)
+ self.assertEqual(
+ a,
+ array.array(self.typecode, self.example+self.example)
+ )
+
b = array.array(self.badtypecode())
self.assertRaises(TypeError, a.extend, b)
@@ -767,7 +799,7 @@
def test_buffer(self):
a = array.array(self.typecode, self.example)
- with test_support._check_py3k_warnings():
+ with test_support.check_py3k_warnings():
b = buffer(a)
self.assertEqual(b[0], a.tostring()[0])
diff --git a/Lib/test/test_code.py b/Lib/test/test_code.py
--- a/Lib/test/test_code.py
+++ b/Lib/test/test_code.py
@@ -70,6 +70,11 @@
"""
+import unittest
+import weakref
+import _testcapi
+
+
def consts(t):
"""Yield a doctest-safe sequence of object reprs."""
for elt in t:
@@ -85,11 +90,47 @@
"freevars", "nlocals"]:
print "%s: %s" % (attr, getattr(co, "co_" + attr))
+
+class CodeTest(unittest.TestCase):
+
+ def test_newempty(self):
+ co = _testcapi.code_newempty("filename", "funcname", 15)
+ self.assertEqual(co.co_filename, "filename")
+ self.assertEqual(co.co_name, "funcname")
+ self.assertEqual(co.co_firstlineno, 15)
+
+
+class CodeWeakRefTest(unittest.TestCase):
+
+ def test_basic(self):
+ # Create a code object in a clean environment so that we know we have
+ # the only reference to it left.
+ namespace = {}
+ exec "def f(): pass" in globals(), namespace
+ f = namespace["f"]
+ del namespace
+
+ self.called = False
+ def callback(code):
+ self.called = True
+
+ # f is now the last reference to the function, and through it, the code
+ # object. While we hold it, check that we can create a weakref and
+ # deref it. Then delete it, and check that the callback gets called and
+ # the reference dies.
+ coderef = weakref.ref(f.__code__, callback)
+ self.assertTrue(bool(coderef()))
+ del f
+ self.assertFalse(bool(coderef()))
+ self.assertTrue(self.called)
+
+
def test_main(verbose=None):
- from test.test_support import run_doctest
+ from test.test_support import run_doctest, run_unittest
from test import test_code
run_doctest(test_code, verbose)
+ run_unittest(CodeTest, CodeWeakRefTest)
-if __name__ == '__main__':
+if __name__ == "__main__":
test_main()
diff --git a/Lib/test/test_codeccallbacks.py b/Lib/test/test_codeccallbacks.py
--- a/Lib/test/test_codeccallbacks.py
+++ b/Lib/test/test_codeccallbacks.py
@@ -186,7 +186,7 @@
charmap = dict([ (ord(c), 2*c.upper()) for c in "abcdefgh"])
sin = u"abc"
sout = "AABBCC"
- self.assertEquals(codecs.charmap_encode(sin, "strict", charmap)[0], sout)
+ self.assertEqual(codecs.charmap_encode(sin, "strict", charmap)[0], sout)
sin = u"abcA"
self.assertRaises(UnicodeError, codecs.charmap_encode, sin, "strict", charmap)
@@ -194,7 +194,7 @@
charmap[ord("?")] = "XYZ"
sin = u"abcDEF"
sout = "AABBCCXYZXYZXYZ"
- self.assertEquals(codecs.charmap_encode(sin, "replace", charmap)[0], sout)
+ self.assertEqual(codecs.charmap_encode(sin, "replace", charmap)[0], sout)
charmap[ord("?")] = u"XYZ"
self.assertRaises(TypeError, codecs.charmap_encode, sin, "replace", charmap)
@@ -327,7 +327,7 @@
# check with the correct number and type of arguments
exc = exctype(*args)
- self.assertEquals(str(exc), msg)
+ self.assertEqual(str(exc), msg)
def test_unicodeencodeerror(self):
self.check_exceptionobjectargs(
@@ -437,15 +437,15 @@
UnicodeError("ouch")
)
# If the correct exception is passed in, "ignore" returns an empty replacement
- self.assertEquals(
+ self.assertEqual(
codecs.ignore_errors(UnicodeEncodeError("ascii", u"\u3042", 0, 1, "ouch")),
(u"", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.ignore_errors(UnicodeDecodeError("ascii", "\xff", 0, 1, "ouch")),
(u"", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.ignore_errors(UnicodeTranslateError(u"\u3042", 0, 1, "ouch")),
(u"", 1)
)
@@ -474,15 +474,15 @@
BadObjectUnicodeDecodeError()
)
# With the correct exception, "replace" returns an "?" or u"\ufffd" replacement
- self.assertEquals(
+ self.assertEqual(
codecs.replace_errors(UnicodeEncodeError("ascii", u"\u3042", 0, 1, "ouch")),
(u"?", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.replace_errors(UnicodeDecodeError("ascii", "\xff", 0, 1, "ouch")),
(u"\ufffd", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.replace_errors(UnicodeTranslateError(u"\u3042", 0, 1, "ouch")),
(u"\ufffd", 1)
)
@@ -514,7 +514,7 @@
# Use the correct exception
cs = (0, 1, 9, 10, 99, 100, 999, 1000, 9999, 10000, 0x3042)
s = "".join(unichr(c) for c in cs)
- self.assertEquals(
+ self.assertEqual(
codecs.xmlcharrefreplace_errors(
UnicodeEncodeError("ascii", s, 0, len(s), "ouch")
),
@@ -546,32 +546,32 @@
UnicodeTranslateError(u"\u3042", 0, 1, "ouch")
)
# Use the correct exception
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\u3042", 0, 1, "ouch")),
(u"\\u3042", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\x00", 0, 1, "ouch")),
(u"\\x00", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\xff", 0, 1, "ouch")),
(u"\\xff", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\u0100", 0, 1, "ouch")),
(u"\\u0100", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\uffff", 0, 1, "ouch")),
(u"\\uffff", 1)
)
if sys.maxunicode>0xffff:
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\U00010000", 0, 1, "ouch")),
(u"\\U00010000", 1)
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors(UnicodeEncodeError("ascii", u"\U0010ffff", 0, 1, "ouch")),
(u"\\U0010ffff", 1)
)
@@ -581,7 +581,7 @@
encs = ("ascii", "latin-1", "iso-8859-1", "iso-8859-15")
for res in results:
- codecs.register_error("test.badhandler", lambda: res)
+ codecs.register_error("test.badhandler", lambda x: res)
for enc in encs:
self.assertRaises(
TypeError,
@@ -603,14 +603,14 @@
)
def test_lookup(self):
- self.assertEquals(codecs.strict_errors, codecs.lookup_error("strict"))
- self.assertEquals(codecs.ignore_errors, codecs.lookup_error("ignore"))
- self.assertEquals(codecs.strict_errors, codecs.lookup_error("strict"))
- self.assertEquals(
+ self.assertEqual(codecs.strict_errors, codecs.lookup_error("strict"))
+ self.assertEqual(codecs.ignore_errors, codecs.lookup_error("ignore"))
+ self.assertEqual(codecs.strict_errors, codecs.lookup_error("strict"))
+ self.assertEqual(
codecs.xmlcharrefreplace_errors,
codecs.lookup_error("xmlcharrefreplace")
)
- self.assertEquals(
+ self.assertEqual(
codecs.backslashreplace_errors,
codecs.lookup_error("backslashreplace")
)
@@ -686,11 +686,11 @@
# Valid negative position
handler.pos = -1
- self.assertEquals("\xff0".decode("ascii", "test.posreturn"), u"<?>0")
+ self.assertEqual("\xff0".decode("ascii", "test.posreturn"), u"<?>0")
# Valid negative position
handler.pos = -2
- self.assertEquals("\xff0".decode("ascii", "test.posreturn"), u"<?><?>")
+ self.assertEqual("\xff0".decode("ascii", "test.posreturn"), u"<?><?>")
# Negative position out of bounds
handler.pos = -3
@@ -698,11 +698,11 @@
# Valid positive position
handler.pos = 1
- self.assertEquals("\xff0".decode("ascii", "test.posreturn"), u"<?>0")
+ self.assertEqual("\xff0".decode("ascii", "test.posreturn"), u"<?>0")
# Largest valid positive position (one beyond end of input)
handler.pos = 2
- self.assertEquals("\xff0".decode("ascii", "test.posreturn"), u"<?>")
+ self.assertEqual("\xff0".decode("ascii", "test.posreturn"), u"<?>")
# Invalid positive position
handler.pos = 3
@@ -710,7 +710,7 @@
# Restart at the "0"
handler.pos = 6
- self.assertEquals("\\uyyyy0".decode("raw-unicode-escape", "test.posreturn"), u"<?>0")
+ self.assertEqual("\\uyyyy0".decode("raw-unicode-escape", "test.posreturn"), u"<?>0")
class D(dict):
def __getitem__(self, key):
@@ -740,11 +740,11 @@
# Valid negative position
handler.pos = -1
- self.assertEquals(u"\xff0".encode("ascii", "test.posreturn"), "<?>0")
+ self.assertEqual(u"\xff0".encode("ascii", "test.posreturn"), "<?>0")
# Valid negative position
handler.pos = -2
- self.assertEquals(u"\xff0".encode("ascii", "test.posreturn"), "<?><?>")
+ self.assertEqual(u"\xff0".encode("ascii", "test.posreturn"), "<?><?>")
# Negative position out of bounds
handler.pos = -3
@@ -752,11 +752,11 @@
# Valid positive position
handler.pos = 1
- self.assertEquals(u"\xff0".encode("ascii", "test.posreturn"), "<?>0")
+ self.assertEqual(u"\xff0".encode("ascii", "test.posreturn"), "<?>0")
# Largest valid positive position (one beyond end of input
handler.pos = 2
- self.assertEquals(u"\xff0".encode("ascii", "test.posreturn"), "<?>")
+ self.assertEqual(u"\xff0".encode("ascii", "test.posreturn"), "<?>")
# Invalid positive position
handler.pos = 3
diff --git a/Lib/test/test_hashlib.py b/Lib/test/test_hashlib.py
--- a/Lib/test/test_hashlib.py
+++ b/Lib/test/test_hashlib.py
@@ -1,16 +1,28 @@
# Test hashlib module
#
-# $Id: test_hashlib.py 79216 2010-03-21 19:16:28Z georg.brandl $
+# $Id: test_hashlib.py 80564 2010-04-27 22:59:35Z victor.stinner $
#
# Copyright (C) 2005-2010 Gregory P. Smith (greg at krypto.org)
# Licensed to PSF under a Contributor Agreement.
#
+import array
import hashlib
+import itertools
+import sys
+try:
+ import threading
+except ImportError:
+ threading = None
import unittest
+import warnings
from test import test_support
from test.test_support import _4G, precisionbigmemtest
+# Were we compiled --with-pydebug or with #define Py_DEBUG?
+COMPILED_WITH_PYDEBUG = hasattr(sys, 'gettotalrefcount')
+
+
def hexstr(s):
import string
h = string.hexdigits
@@ -26,24 +38,93 @@
'sha224', 'SHA224', 'sha256', 'SHA256',
'sha384', 'SHA384', 'sha512', 'SHA512' )
+ _warn_on_extension_import = COMPILED_WITH_PYDEBUG
+
+ def _conditional_import_module(self, module_name):
+ """Import a module and return a reference to it or None on failure."""
+ try:
+ exec('import '+module_name)
+ except ImportError, error:
+ if self._warn_on_extension_import:
+ warnings.warn('Did a C extension fail to compile? %s' % error)
+ return locals().get(module_name)
+
+ def __init__(self, *args, **kwargs):
+ algorithms = set()
+ for algorithm in self.supported_hash_names:
+ algorithms.add(algorithm.lower())
+ self.constructors_to_test = {}
+ for algorithm in algorithms:
+ self.constructors_to_test[algorithm] = set()
+
+ # For each algorithm, test the direct constructor and the use
+ # of hashlib.new given the algorithm name.
+ for algorithm, constructors in self.constructors_to_test.items():
+ constructors.add(getattr(hashlib, algorithm))
+ def _test_algorithm_via_hashlib_new(data=None, _alg=algorithm):
+ if data is None:
+ return hashlib.new(_alg)
+ return hashlib.new(_alg, data)
+ constructors.add(_test_algorithm_via_hashlib_new)
+
+ _hashlib = self._conditional_import_module('_hashlib')
+ if _hashlib:
+ # These two algorithms should always be present when this module
+ # is compiled. If not, something was compiled wrong.
+ assert hasattr(_hashlib, 'openssl_md5')
+ assert hasattr(_hashlib, 'openssl_sha1')
+ for algorithm, constructors in self.constructors_to_test.items():
+ constructor = getattr(_hashlib, 'openssl_'+algorithm, None)
+ if constructor:
+ constructors.add(constructor)
+
+ _md5 = self._conditional_import_module('_md5')
+ if _md5:
+ self.constructors_to_test['md5'].add(_md5.new)
+ _sha = self._conditional_import_module('_sha')
+ if _sha:
+ self.constructors_to_test['sha1'].add(_sha.new)
+ _sha256 = self._conditional_import_module('_sha256')
+ if _sha256:
+ self.constructors_to_test['sha224'].add(_sha256.sha224)
+ self.constructors_to_test['sha256'].add(_sha256.sha256)
+ _sha512 = self._conditional_import_module('_sha512')
+ if _sha512:
+ self.constructors_to_test['sha384'].add(_sha512.sha384)
+ self.constructors_to_test['sha512'].add(_sha512.sha512)
+
+ super(HashLibTestCase, self).__init__(*args, **kwargs)
+
+ def test_hash_array(self):
+ a = array.array("b", range(10))
+ constructors = self.constructors_to_test.itervalues()
+ for cons in itertools.chain.from_iterable(constructors):
+ c = cons(a)
+ c.hexdigest()
+
+ def test_algorithms_attribute(self):
+ self.assertEqual(hashlib.algorithms,
+ tuple([_algo for _algo in self.supported_hash_names if
+ _algo.islower()]))
+
def test_unknown_hash(self):
try:
hashlib.new('spam spam spam spam spam')
except ValueError:
pass
else:
- self.assert_(0 == "hashlib didn't reject bogus hash name")
+ self.assertTrue(0 == "hashlib didn't reject bogus hash name")
def test_hexdigest(self):
for name in self.supported_hash_names:
h = hashlib.new(name)
- self.assert_(hexstr(h.digest()) == h.hexdigest())
-
+ self.assertTrue(hexstr(h.digest()) == h.hexdigest())
def test_large_update(self):
aas = 'a' * 128
bees = 'b' * 127
cees = 'c' * 126
+ abcs = aas + bees + cees
for name in self.supported_hash_names:
m1 = hashlib.new(name)
@@ -52,17 +133,39 @@
m1.update(cees)
m2 = hashlib.new(name)
- m2.update(aas + bees + cees)
- self.assertEqual(m1.digest(), m2.digest())
+ m2.update(abcs)
+ self.assertEqual(m1.digest(), m2.digest(), name+' update problem.')
+
+ m3 = hashlib.new(name, abcs)
+ self.assertEqual(m1.digest(), m3.digest(), name+' new problem.')
def check(self, name, data, digest):
- # test the direct constructors
- computed = getattr(hashlib, name)(data).hexdigest()
- self.assert_(computed == digest)
- # test the general new() interface
- computed = hashlib.new(name, data).hexdigest()
- self.assert_(computed == digest)
+ constructors = self.constructors_to_test[name]
+ # 2 is for hashlib.name(...) and hashlib.new(name, ...)
+ self.assertGreaterEqual(len(constructors), 2)
+ for hash_object_constructor in constructors:
+ computed = hash_object_constructor(data).hexdigest()
+ self.assertEqual(
+ computed, digest,
+ "Hash algorithm %s constructed using %s returned hexdigest"
+ " %r for %d byte input data that should have hashed to %r."
+ % (name, hash_object_constructor,
+ computed, len(data), digest))
+ def check_unicode(self, algorithm_name):
+ # Unicode objects are not allowed as input.
+ expected = hashlib.new(algorithm_name, str(u'spam')).hexdigest()
+ self.check(algorithm_name, u'spam', expected)
+
+ def test_unicode(self):
+ # In python 2.x unicode is auto-encoded to the system default encoding
+ # when passed to hashlib functions.
+ self.check_unicode('md5')
+ self.check_unicode('sha1')
+ self.check_unicode('sha224')
+ self.check_unicode('sha256')
+ self.check_unicode('sha384')
+ self.check_unicode('sha512')
def test_case_md5_0(self):
self.check('md5', '', 'd41d8cd98f00b204e9800998ecf8427e')
@@ -196,6 +299,42 @@
"e718483d0ce769644e2e42c7bc15b4638e1f98b13b2044285632a803afa973eb"+
"de0ff244877ea60a4cb0432ce577c31beb009c5c2c49aa2e4eadb217ad8cc09b")
+ @unittest.skipUnless(threading, 'Threading required for this test.')
+ @test_support.reap_threads
+ def test_threaded_hashing(self):
+ # Updating the same hash object from several threads at once
+ # using data chunk sizes containing the same byte sequences.
+ #
+ # If the internal locks are working to prevent multiple
+ # updates on the same object from running at once, the resulting
+ # hash will be the same as doing it single threaded upfront.
+ hasher = hashlib.sha1()
+ num_threads = 5
+ smallest_data = 'swineflu'
+ data = smallest_data*200000
+ expected_hash = hashlib.sha1(data*num_threads).hexdigest()
+
+ def hash_in_chunks(chunk_size, event):
+ index = 0
+ while index < len(data):
+ hasher.update(data[index:index+chunk_size])
+ index += chunk_size
+ event.set()
+
+ events = []
+ for threadnum in xrange(num_threads):
+ chunk_size = len(data) // (10**threadnum)
+ assert chunk_size > 0
+ assert chunk_size % len(smallest_data) == 0
+ event = threading.Event()
+ events.append(event)
+ threading.Thread(target=hash_in_chunks,
+ args=(chunk_size, event)).start()
+
+ for event in events:
+ event.wait()
+
+ self.assertEqual(expected_hash, hasher.hexdigest())
def test_main():
if test_support.is_jython:
@@ -210,6 +349,5 @@
test_support.run_unittest(HashLibTestCase)
-
if __name__ == "__main__":
test_main()
diff --git a/Lib/test/test_hmac.py b/Lib/test/test_hmac.py
--- a/Lib/test/test_hmac.py
+++ b/Lib/test/test_hmac.py
@@ -213,19 +213,13 @@
with warnings.catch_warnings():
warnings.simplefilter('error', RuntimeWarning)
- try:
+ with self.assertRaises(RuntimeWarning):
hmac.HMAC('a', 'b', digestmod=MockCrazyHash)
- except RuntimeWarning:
- pass
- else:
self.fail('Expected warning about missing block_size')
MockCrazyHash.block_size = 1
- try:
+ with self.assertRaises(RuntimeWarning):
hmac.HMAC('a', 'b', digestmod=MockCrazyHash)
- except RuntimeWarning:
- pass
- else:
self.fail('Expected warning about small block_size')
@@ -260,7 +254,7 @@
# Testing if HMAC defaults to MD5 algorithm.
# NOTE: this whitebox test depends on the hmac class internals
h = hmac.HMAC("key")
- self.failUnless(h.digest_cons == hashlib.md5)
+ self.assertTrue(h.digest_cons == hashlib.md5)
def test_exercise_all_methods(self):
# Exercising all methods once.
@@ -280,11 +274,11 @@
# Testing if attributes are of same type.
h1 = hmac.HMAC("key")
h2 = h1.copy()
- self.failUnless(h1.digest_cons == h2.digest_cons,
+ self.assertTrue(h1.digest_cons == h2.digest_cons,
"digest constructors don't match.")
- self.failUnless(type(h1.inner) == type(h2.inner),
+ self.assertTrue(type(h1.inner) == type(h2.inner),
"Types of inner don't match.")
- self.failUnless(type(h1.outer) == type(h2.outer),
+ self.assertTrue(type(h1.outer) == type(h2.outer),
"Types of outer don't match.")
def test_realcopy(self):
@@ -292,10 +286,10 @@
h1 = hmac.HMAC("key")
h2 = h1.copy()
# Using id() in case somebody has overridden __cmp__.
- self.failUnless(id(h1) != id(h2), "No real copy of the HMAC instance.")
- self.failUnless(id(h1.inner) != id(h2.inner),
+ self.assertTrue(id(h1) != id(h2), "No real copy of the HMAC instance.")
+ self.assertTrue(id(h1.inner) != id(h2.inner),
"No real copy of the attribute 'inner'.")
- self.failUnless(id(h1.outer) != id(h2.outer),
+ self.assertTrue(id(h1.outer) != id(h2.outer),
"No real copy of the attribute 'outer'.")
def test_equality(self):
@@ -303,9 +297,9 @@
h1 = hmac.HMAC("key")
h1.update("some random text")
h2 = h1.copy()
- self.failUnless(h1.digest() == h2.digest(),
+ self.assertTrue(h1.digest() == h2.digest(),
"Digest of copy doesn't match original digest.")
- self.failUnless(h1.hexdigest() == h2.hexdigest(),
+ self.assertTrue(h1.hexdigest() == h2.hexdigest(),
"Hexdigest of copy doesn't match original hexdigest.")
def test_main():
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -26,27 +26,27 @@
import logging.config
import codecs
-import copy
import cPickle
import cStringIO
import gc
+import json
import os
import re
import select
import socket
from SocketServer import ThreadingTCPServer, StreamRequestHandler
-import string
import struct
import sys
import tempfile
from test.test_support import captured_stdout, run_with_locale, run_unittest
import textwrap
-import threading
-import time
-import types
import unittest
+import warnings
import weakref
-
+try:
+ import threading
+except ImportError:
+ threading = None
class BaseTest(unittest.TestCase):
@@ -71,7 +71,7 @@
# Set two unused loggers: one non-ASCII and one Unicode.
# This is to test correct operation when sorting existing
- # loggers in the configuration code. See issues 8201, 9310.
+ # loggers in the configuration code. See issue 8201.
logging.getLogger("\xab\xd7\xbb")
logging.getLogger(u"\u013f\u00d6\u0047")
@@ -90,6 +90,10 @@
level."""
self.stream.close()
self.root_logger.removeHandler(self.root_hdlr)
+ while self.root_logger.handlers:
+ h = self.root_logger.handlers[0]
+ self.root_logger.removeHandler(h)
+ h.close()
self.root_logger.setLevel(self.original_logging_level)
logging._acquireLock()
try:
@@ -116,13 +120,13 @@
except AttributeError:
# StringIO.StringIO lacks a reset() method.
actual_lines = stream.getvalue().splitlines()
- self.assertEquals(len(actual_lines), len(expected_values))
+ self.assertEqual(len(actual_lines), len(expected_values))
for actual, expected in zip(actual_lines, expected_values):
match = pat.search(actual)
if not match:
self.fail("Log line does not match expected pattern:\n" +
actual)
- self.assertEquals(tuple(match.groups()), expected)
+ self.assertEqual(tuple(match.groups()), expected)
s = stream.read()
if s:
self.fail("Remaining output at end of log stream:\n" + s)
@@ -637,14 +641,8 @@
"""
def apply_config(self, conf):
- try:
- fn = tempfile.mktemp(".ini")
- f = open(fn, "w")
- f.write(textwrap.dedent(conf))
- f.close()
- logging.config.fileConfig(fn)
- finally:
- os.remove(fn)
+ file = cStringIO.StringIO(textwrap.dedent(conf))
+ logging.config.fileConfig(file)
def test_config0_ok(self):
# A simple config file which overrides the default settings.
@@ -694,7 +692,7 @@
except RuntimeError:
logging.exception("just testing")
sys.stdout.seek(0)
- self.assertEquals(output.getvalue(),
+ self.assertEqual(output.getvalue(),
"ERROR:root:just testing\nGot a [RuntimeError]\n")
# Original logger output is empty
self.assert_log_lines([])
@@ -776,6 +774,7 @@
self.server_close()
+ at unittest.skipUnless(threading, 'Threading required for this test.')
class SocketHandlerTest(BaseTest):
"""Test for SocketHandler objects."""
@@ -820,7 +819,7 @@
logger = logging.getLogger("tcp")
logger.error("spam")
logger.debug("eggs")
- self.assertEquals(self.get_output(), "spam\neggs\n")
+ self.assertEqual(self.get_output(), "spam\neggs\n")
class MemoryTest(BaseTest):
@@ -839,7 +838,7 @@
key = id(obj), repr(obj)
self._survivors[key] = weakref.ref(obj)
- def _assert_survival(self):
+ def _assertTruesurvival(self):
"""Assert that all objects watched for survival have survived."""
# Trigger cycle breaking.
gc.collect()
@@ -865,7 +864,7 @@
])
del foo
# foo has survived.
- self._assert_survival()
+ self._assertTruesurvival()
# foo has retained its settings.
bar = logging.getLogger("foo")
bar.debug(self.next_message())
@@ -894,7 +893,7 @@
# check we wrote exactly those bytes, ignoring trailing \n etc
f = open(fn)
try:
- self.failUnlessEqual(f.read().rstrip(), data)
+ self.assertEqual(f.read().rstrip(), data)
finally:
f.close()
finally:
@@ -923,15 +922,860 @@
self.assertEqual(s, '\xe4\xee \xf1\xe2\xe8\xe4\xe0\xed\xe8\xff\n')
+class WarningsTest(BaseTest):
+
+ def test_warnings(self):
+ with warnings.catch_warnings():
+ logging.captureWarnings(True)
+ try:
+ warnings.filterwarnings("always", category=UserWarning)
+ file = cStringIO.StringIO()
+ h = logging.StreamHandler(file)
+ logger = logging.getLogger("py.warnings")
+ logger.addHandler(h)
+ warnings.warn("I'm warning you...")
+ logger.removeHandler(h)
+ s = file.getvalue()
+ h.close()
+ self.assertTrue(s.find("UserWarning: I'm warning you...\n") > 0)
+
+ #See if an explicit file uses the original implementation
+ file = cStringIO.StringIO()
+ warnings.showwarning("Explicit", UserWarning, "dummy.py", 42,
+ file, "Dummy line")
+ s = file.getvalue()
+ file.close()
+ self.assertEqual(s,
+ "dummy.py:42: UserWarning: Explicit\n Dummy line\n")
+ finally:
+ logging.captureWarnings(False)
+
+
+def formatFunc(format, datefmt=None):
+ return logging.Formatter(format, datefmt)
+
+def handlerFunc():
+ return logging.StreamHandler()
+
+class CustomHandler(logging.StreamHandler):
+ pass
+
+class ConfigDictTest(BaseTest):
+
+ """Reading logging config from a dictionary."""
+
+ expected_log_pat = r"^([\w]+) \+\+ ([\w]+)$"
+
+ # config0 is a standard configuration.
+ config0 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ 'handlers' : ['hand1'],
+ },
+ }
+
+ # config1 adds a little to the standard configuration.
+ config1 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ # config2 has a subtle configuration error that should be reported
+ config2 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdbout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ #As config1 but with a misspelt level on a handler
+ config2a = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NTOSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+
+ #As config1 but with a misspelt level on a logger
+ config2b = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WRANING',
+ },
+ }
+
+ # config3 has a less subtle configuration error
+ config3 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'misspelled_name',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ # config4 specifies a custom formatter class to be loaded
+ config4 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ '()' : __name__ + '.ExceptionFormatter',
+ 'format' : '%(levelname)s:%(name)s:%(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'root' : {
+ 'level' : 'NOTSET',
+ 'handlers' : ['hand1'],
+ },
+ }
+
+ # As config4 but using an actual callable rather than a string
+ config4a = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ '()' : ExceptionFormatter,
+ 'format' : '%(levelname)s:%(name)s:%(message)s',
+ },
+ 'form2' : {
+ '()' : __name__ + '.formatFunc',
+ 'format' : '%(levelname)s:%(name)s:%(message)s',
+ },
+ 'form3' : {
+ '()' : formatFunc,
+ 'format' : '%(levelname)s:%(name)s:%(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ 'hand2' : {
+ '()' : handlerFunc,
+ },
+ },
+ 'root' : {
+ 'level' : 'NOTSET',
+ 'handlers' : ['hand1'],
+ },
+ }
+
+ # config5 specifies a custom handler class to be loaded
+ config5 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : __name__ + '.CustomHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ # config6 specifies a custom handler class to be loaded
+ # but has bad arguments
+ config6 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : __name__ + '.CustomHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ '9' : 'invalid parameter name',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ #config 7 does not define compiler.parser but defines compiler.lexer
+ #so compiler.parser should be disabled after applying it
+ config7 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.lexer' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ config8 = {
+ 'version': 1,
+ 'disable_existing_loggers' : False,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ 'compiler.lexer' : {
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ config9 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'WARNING',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'WARNING',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'NOTSET',
+ },
+ }
+
+ config9a = {
+ 'version': 1,
+ 'incremental' : True,
+ 'handlers' : {
+ 'hand1' : {
+ 'level' : 'WARNING',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'INFO',
+ },
+ },
+ }
+
+ config9b = {
+ 'version': 1,
+ 'incremental' : True,
+ 'handlers' : {
+ 'hand1' : {
+ 'level' : 'INFO',
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'INFO',
+ },
+ },
+ }
+
+ #As config1 but with a filter added
+ config10 = {
+ 'version': 1,
+ 'formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'filters' : {
+ 'filt1' : {
+ 'name' : 'compiler.parser',
+ },
+ },
+ 'handlers' : {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ 'filters' : ['filt1'],
+ },
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'filters' : ['filt1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ 'handlers' : ['hand1'],
+ },
+ }
+
+ #As config1 but using cfg:// references
+ config11 = {
+ 'version': 1,
+ 'true_formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handler_configs': {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'formatters' : 'cfg://true_formatters',
+ 'handlers' : {
+ 'hand1' : 'cfg://handler_configs[hand1]',
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ #As config11 but missing the version key
+ config12 = {
+ 'true_formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handler_configs': {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'formatters' : 'cfg://true_formatters',
+ 'handlers' : {
+ 'hand1' : 'cfg://handler_configs[hand1]',
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ #As config11 but using an unsupported version
+ config13 = {
+ 'version': 2,
+ 'true_formatters': {
+ 'form1' : {
+ 'format' : '%(levelname)s ++ %(message)s',
+ },
+ },
+ 'handler_configs': {
+ 'hand1' : {
+ 'class' : 'logging.StreamHandler',
+ 'formatter' : 'form1',
+ 'level' : 'NOTSET',
+ 'stream' : 'ext://sys.stdout',
+ },
+ },
+ 'formatters' : 'cfg://true_formatters',
+ 'handlers' : {
+ 'hand1' : 'cfg://handler_configs[hand1]',
+ },
+ 'loggers' : {
+ 'compiler.parser' : {
+ 'level' : 'DEBUG',
+ 'handlers' : ['hand1'],
+ },
+ },
+ 'root' : {
+ 'level' : 'WARNING',
+ },
+ }
+
+ def apply_config(self, conf):
+ logging.config.dictConfig(conf)
+
+ def test_config0_ok(self):
+ # A simple config which overrides the default settings.
+ with captured_stdout() as output:
+ self.apply_config(self.config0)
+ logger = logging.getLogger()
+ # Won't output anything
+ logger.info(self.next_message())
+ # Outputs a message
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('ERROR', '2'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+
+ def test_config1_ok(self, config=config1):
+ # A config defining a sub-parser as well.
+ with captured_stdout() as output:
+ self.apply_config(config)
+ logger = logging.getLogger("compiler.parser")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '1'),
+ ('ERROR', '2'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+
+ def test_config2_failure(self):
+ # A simple config which overrides the default settings.
+ self.assertRaises(StandardError, self.apply_config, self.config2)
+
+ def test_config2a_failure(self):
+ # A simple config which overrides the default settings.
+ self.assertRaises(StandardError, self.apply_config, self.config2a)
+
+ def test_config2b_failure(self):
+ # A simple config which overrides the default settings.
+ self.assertRaises(StandardError, self.apply_config, self.config2b)
+
+ def test_config3_failure(self):
+ # A simple config which overrides the default settings.
+ self.assertRaises(StandardError, self.apply_config, self.config3)
+
+ def test_config4_ok(self):
+ # A config specifying a custom formatter class.
+ with captured_stdout() as output:
+ self.apply_config(self.config4)
+ #logger = logging.getLogger()
+ try:
+ raise RuntimeError()
+ except RuntimeError:
+ logging.exception("just testing")
+ sys.stdout.seek(0)
+ self.assertEqual(output.getvalue(),
+ "ERROR:root:just testing\nGot a [RuntimeError]\n")
+ # Original logger output is empty
+ self.assert_log_lines([])
+
+ def test_config4a_ok(self):
+ # A config specifying a custom formatter class.
+ with captured_stdout() as output:
+ self.apply_config(self.config4a)
+ #logger = logging.getLogger()
+ try:
+ raise RuntimeError()
+ except RuntimeError:
+ logging.exception("just testing")
+ sys.stdout.seek(0)
+ self.assertEqual(output.getvalue(),
+ "ERROR:root:just testing\nGot a [RuntimeError]\n")
+ # Original logger output is empty
+ self.assert_log_lines([])
+
+ def test_config5_ok(self):
+ self.test_config1_ok(config=self.config5)
+
+ def test_config6_failure(self):
+ self.assertRaises(StandardError, self.apply_config, self.config6)
+
+ def test_config7_ok(self):
+ with captured_stdout() as output:
+ self.apply_config(self.config1)
+ logger = logging.getLogger("compiler.parser")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '1'),
+ ('ERROR', '2'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+ with captured_stdout() as output:
+ self.apply_config(self.config7)
+ logger = logging.getLogger("compiler.parser")
+ self.assertTrue(logger.disabled)
+ logger = logging.getLogger("compiler.lexer")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '3'),
+ ('ERROR', '4'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+
+ #Same as test_config_7_ok but don't disable old loggers.
+ def test_config_8_ok(self):
+ with captured_stdout() as output:
+ self.apply_config(self.config1)
+ logger = logging.getLogger("compiler.parser")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '1'),
+ ('ERROR', '2'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+ with captured_stdout() as output:
+ self.apply_config(self.config8)
+ logger = logging.getLogger("compiler.parser")
+ self.assertFalse(logger.disabled)
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ logger = logging.getLogger("compiler.lexer")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '3'),
+ ('ERROR', '4'),
+ ('INFO', '5'),
+ ('ERROR', '6'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+
+ def test_config_9_ok(self):
+ with captured_stdout() as output:
+ self.apply_config(self.config9)
+ logger = logging.getLogger("compiler.parser")
+ #Nothing will be output since both handler and logger are set to WARNING
+ logger.info(self.next_message())
+ self.assert_log_lines([], stream=output)
+ self.apply_config(self.config9a)
+ #Nothing will be output since both handler is still set to WARNING
+ logger.info(self.next_message())
+ self.assert_log_lines([], stream=output)
+ self.apply_config(self.config9b)
+ #Message should now be output
+ logger.info(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '3'),
+ ], stream=output)
+
+ def test_config_10_ok(self):
+ with captured_stdout() as output:
+ self.apply_config(self.config10)
+ logger = logging.getLogger("compiler.parser")
+ logger.warning(self.next_message())
+ logger = logging.getLogger('compiler')
+ #Not output, because filtered
+ logger.warning(self.next_message())
+ logger = logging.getLogger('compiler.lexer')
+ #Not output, because filtered
+ logger.warning(self.next_message())
+ logger = logging.getLogger("compiler.parser.codegen")
+ #Output, as not filtered
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('WARNING', '1'),
+ ('ERROR', '4'),
+ ], stream=output)
+
+ def test_config11_ok(self):
+ self.test_config1_ok(self.config11)
+
+ def test_config12_failure(self):
+ self.assertRaises(StandardError, self.apply_config, self.config12)
+
+ def test_config13_failure(self):
+ self.assertRaises(StandardError, self.apply_config, self.config13)
+
+ @unittest.skipUnless(threading, 'listen() needs threading to work')
+ def setup_via_listener(self, text):
+ # Ask for a randomly assigned port (by using port 0)
+ t = logging.config.listen(0)
+ t.start()
+ t.ready.wait()
+ # Now get the port allocated
+ port = t.port
+ t.ready.clear()
+ try:
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock.settimeout(2.0)
+ sock.connect(('localhost', port))
+
+ slen = struct.pack('>L', len(text))
+ s = slen + text
+ sentsofar = 0
+ left = len(s)
+ while left > 0:
+ sent = sock.send(s[sentsofar:])
+ sentsofar += sent
+ left -= sent
+ sock.close()
+ finally:
+ t.ready.wait(2.0)
+ logging.config.stopListening()
+ t.join(2.0)
+
+ def test_listen_config_10_ok(self):
+ with captured_stdout() as output:
+ self.setup_via_listener(json.dumps(self.config10))
+ logger = logging.getLogger("compiler.parser")
+ logger.warning(self.next_message())
+ logger = logging.getLogger('compiler')
+ #Not output, because filtered
+ logger.warning(self.next_message())
+ logger = logging.getLogger('compiler.lexer')
+ #Not output, because filtered
+ logger.warning(self.next_message())
+ logger = logging.getLogger("compiler.parser.codegen")
+ #Output, as not filtered
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('WARNING', '1'),
+ ('ERROR', '4'),
+ ], stream=output)
+
+ def test_listen_config_1_ok(self):
+ with captured_stdout() as output:
+ self.setup_via_listener(textwrap.dedent(ConfigFileTest.config1))
+ logger = logging.getLogger("compiler.parser")
+ # Both will output a message
+ logger.info(self.next_message())
+ logger.error(self.next_message())
+ self.assert_log_lines([
+ ('INFO', '1'),
+ ('ERROR', '2'),
+ ], stream=output)
+ # Original logger output is empty.
+ self.assert_log_lines([])
+
+
+class ManagerTest(BaseTest):
+ def test_manager_loggerclass(self):
+ logged = []
+
+ class MyLogger(logging.Logger):
+ def _log(self, level, msg, args, exc_info=None, extra=None):
+ logged.append(msg)
+
+ man = logging.Manager(None)
+ self.assertRaises(TypeError, man.setLoggerClass, int)
+ man.setLoggerClass(MyLogger)
+ logger = man.getLogger('test')
+ logger.warning('should appear in logged')
+ logging.warning('should not appear in logged')
+
+ self.assertEqual(logged, ['should appear in logged'])
+
+
+class ChildLoggerTest(BaseTest):
+ def test_child_loggers(self):
+ r = logging.getLogger()
+ l1 = logging.getLogger('abc')
+ l2 = logging.getLogger('def.ghi')
+ c1 = r.getChild('xyz')
+ c2 = r.getChild('uvw.xyz')
+ self.assertTrue(c1 is logging.getLogger('xyz'))
+ self.assertTrue(c2 is logging.getLogger('uvw.xyz'))
+ c1 = l1.getChild('def')
+ c2 = c1.getChild('ghi')
+ c3 = l1.getChild('def.ghi')
+ self.assertTrue(c1 is logging.getLogger('abc.def'))
+ self.assertTrue(c2 is logging.getLogger('abc.def.ghi'))
+ self.assertTrue(c2 is c3)
+
+
# Set the locale to the platform-dependent default. I have no idea
# why the test does this, but in any case we save the current locale
# first and restore it at the end.
@run_with_locale('LC_ALL', '')
def test_main():
run_unittest(BuiltinLevelsTest, BasicFilterTest,
- CustomLevelsAndFiltersTest, MemoryHandlerTest,
- ConfigFileTest, SocketHandlerTest, MemoryTest,
- EncodingTest)
+ CustomLevelsAndFiltersTest, MemoryHandlerTest,
+ ConfigFileTest, SocketHandlerTest, MemoryTest,
+ EncodingTest, WarningsTest, ConfigDictTest, ManagerTest,
+ ChildLoggerTest)
if __name__ == "__main__":
test_main()
diff --git a/Lib/test/test_new.py b/Lib/test/test_new.py
--- a/Lib/test/test_new.py
+++ b/Lib/test/test_new.py
@@ -153,7 +153,7 @@
d = new.code(argcount, nlocals, stacksize, flags, codestring,
constants, t, varnames, filename, name,
firstlineno, lnotab)
- self.assert_(type(t[0]) is S, "eek, tuple changed under us!")
+ self.assertTrue(type(t[0]) is S, "eek, tuple changed under us!")
def test_main():
test_support.run_unittest(NewTest)
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -3,10 +3,16 @@
# portable than they had been thought to be.
import os
+import errno
import unittest
import warnings
import sys
+import signal
+import subprocess
+import time
from test import test_support
+import mmap
+import uuid
warnings.filterwarnings("ignore", "tempnam", RuntimeWarning, __name__)
warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, __name__)
@@ -21,7 +27,7 @@
def test_access(self):
f = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
os.close(f)
- self.assert_(os.access(test_support.TESTFN, os.W_OK))
+ self.assertTrue(os.access(test_support.TESTFN, os.W_OK))
def test_closerange(self):
first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
@@ -36,10 +42,7 @@
retries += 1
if retries > 10:
# XXX test skipped
- print >> sys.stderr, (
- "couldn't allocate two consecutive fds, "
- "skipping test_closerange")
- return
+ self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
@@ -47,6 +50,7 @@
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, "a")
+ @test_support.cpython_only
def test_rename(self):
path = unicode(test_support.TESTFN)
if not test_support.is_jython:
@@ -69,7 +73,7 @@
def check_tempfile(self, name):
# make sure it doesn't already exist:
- self.failIf(os.path.exists(name),
+ self.assertFalse(os.path.exists(name),
"file already exists for temporary file")
# make sure we can create the file
open(name, "w")
@@ -78,16 +82,18 @@
def test_tempnam(self):
if not hasattr(os, "tempnam"):
return
- warnings.filterwarnings("ignore", "tempnam", RuntimeWarning,
- r"test_os$")
- self.check_tempfile(os.tempnam())
+ with warnings.catch_warnings():
+ warnings.filterwarnings("ignore", "tempnam", RuntimeWarning,
+ r"test_os$")
+ warnings.filterwarnings("ignore", "tempnam", DeprecationWarning)
+ self.check_tempfile(os.tempnam())
- name = os.tempnam(test_support.TESTFN)
- self.check_tempfile(name)
+ name = os.tempnam(test_support.TESTFN)
+ self.check_tempfile(name)
- name = os.tempnam(test_support.TESTFN, "pfx")
- self.assert_(os.path.basename(name)[:3] == "pfx")
- self.check_tempfile(name)
+ name = os.tempnam(test_support.TESTFN, "pfx")
+ self.assertTrue(os.path.basename(name)[:3] == "pfx")
+ self.check_tempfile(name)
def test_tmpfile(self):
if not hasattr(os, "tmpfile"):
@@ -106,64 +112,69 @@
# test that a subsequent call to os.tmpfile() raises the same error. If
# it doesn't, assume we're on XP or below and the user running the test
# has administrative privileges, and proceed with the test as normal.
- if sys.platform == 'win32':
- name = '\\python_test_os_test_tmpfile.txt'
- if os.path.exists(name):
- os.remove(name)
- try:
- fp = open(name, 'w')
- except IOError, first:
- # open() failed, assert tmpfile() fails in the same way.
- # Although open() raises an IOError and os.tmpfile() raises an
- # OSError(), 'args' will be (13, 'Permission denied') in both
- # cases.
+ with warnings.catch_warnings():
+ warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
+
+ if sys.platform == 'win32':
+ name = '\\python_test_os_test_tmpfile.txt'
+ if os.path.exists(name):
+ os.remove(name)
try:
- fp = os.tmpfile()
- except OSError, second:
- self.assertEqual(first.args, second.args)
+ fp = open(name, 'w')
+ except IOError, first:
+ # open() failed, assert tmpfile() fails in the same way.
+ # Although open() raises an IOError and os.tmpfile() raises an
+ # OSError(), 'args' will be (13, 'Permission denied') in both
+ # cases.
+ try:
+ fp = os.tmpfile()
+ except OSError, second:
+ self.assertEqual(first.args, second.args)
+ else:
+ self.fail("expected os.tmpfile() to raise OSError")
+ return
else:
- self.fail("expected os.tmpfile() to raise OSError")
- return
- else:
- # open() worked, therefore, tmpfile() should work. Close our
- # dummy file and proceed with the test as normal.
- fp.close()
- os.remove(name)
+ # open() worked, therefore, tmpfile() should work. Close our
+ # dummy file and proceed with the test as normal.
+ fp.close()
+ os.remove(name)
- fp = os.tmpfile()
- fp.write("foobar")
- fp.seek(0,0)
- s = fp.read()
- fp.close()
- self.assert_(s == "foobar")
+ fp = os.tmpfile()
+ fp.write("foobar")
+ fp.seek(0,0)
+ s = fp.read()
+ fp.close()
+ self.assertTrue(s == "foobar")
def test_tmpnam(self):
- import sys
if not hasattr(os, "tmpnam"):
return
- warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning,
- r"test_os$")
- name = os.tmpnam()
- if sys.platform in ("win32",):
- # The Windows tmpnam() seems useless. From the MS docs:
- #
- # The character string that tmpnam creates consists of
- # the path prefix, defined by the entry P_tmpdir in the
- # file STDIO.H, followed by a sequence consisting of the
- # digit characters '0' through '9'; the numerical value
- # of this string is in the range 1 - 65,535. Changing the
- # definitions of L_tmpnam or P_tmpdir in STDIO.H does not
- # change the operation of tmpnam.
- #
- # The really bizarre part is that, at least under MSVC6,
- # P_tmpdir is "\\". That is, the path returned refers to
- # the root of the current drive. That's a terrible place to
- # put temp files, and, depending on privileges, the user
- # may not even be able to open a file in the root directory.
- self.failIf(os.path.exists(name),
- "file already exists for temporary file")
- else:
- self.check_tempfile(name)
+ with warnings.catch_warnings():
+ warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning,
+ r"test_os$")
+ warnings.filterwarnings("ignore", "tmpnam", DeprecationWarning)
+
+ name = os.tmpnam()
+ if sys.platform in ("win32",):
+ # The Windows tmpnam() seems useless. From the MS docs:
+ #
+ # The character string that tmpnam creates consists of
+ # the path prefix, defined by the entry P_tmpdir in the
+ # file STDIO.H, followed by a sequence consisting of the
+ # digit characters '0' through '9'; the numerical value
+ # of this string is in the range 1 - 65,535. Changing the
+ # definitions of L_tmpnam or P_tmpdir in STDIO.H does not
+ # change the operation of tmpnam.
+ #
+ # The really bizarre part is that, at least under MSVC6,
+ # P_tmpdir is "\\". That is, the path returned refers to
+ # the root of the current drive. That's a terrible place to
+ # put temp files, and, depending on privileges, the user
+ # may not even be able to open a file in the root directory.
+ self.assertFalse(os.path.exists(name),
+ "file already exists for temporary file")
+ else:
+ self.check_tempfile(name)
# Test attributes on return values from os.*stat* family.
class StatAttributeTests(unittest.TestCase):
@@ -186,10 +197,8 @@
result = os.stat(self.fname)
# Make sure direct access works
- self.assertEquals(result[stat.ST_SIZE], 3)
- self.assertEquals(result.st_size, 3)
-
- import sys
+ self.assertEqual(result[stat.ST_SIZE], 3)
+ self.assertEqual(result.st_size, 3)
# Make sure all the attributes are there
members = dir(result)
@@ -200,9 +209,9 @@
def trunc(x): return int(x)
else:
def trunc(x): return x
- self.assertEquals(trunc(getattr(result, attr)),
- result[getattr(stat, name)])
- self.assert_(attr in members)
+ self.assertEqual(trunc(getattr(result, attr)),
+ result[getattr(stat, name)])
+ self.assertIn(attr, members)
try:
result[200]
@@ -214,7 +223,7 @@
try:
result.st_mode = 1
self.fail("No exception thrown")
- except TypeError:
+ except (AttributeError, TypeError):
pass
try:
@@ -251,18 +260,17 @@
result = os.statvfs(self.fname)
except OSError, e:
# On AtheOS, glibc always returns ENOSYS
- import errno
if e.errno == errno.ENOSYS:
return
# Make sure direct access works
- self.assertEquals(result.f_bfree, result[3])
+ self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
- self.assertEquals(getattr(result, 'f_' + member), result[value])
+ self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
@@ -297,7 +305,7 @@
# time stamps in stat, but not in utime.
os.utime(test_support.TESTFN, (st.st_atime, int(st.st_mtime-delta)))
st2 = os.stat(test_support.TESTFN)
- self.assertEquals(st2.st_mtime, int(st.st_mtime-delta))
+ self.assertEqual(st2.st_mtime, int(st.st_mtime-delta))
# Restrict test to Win32, since there is no guarantee other
# systems support centiseconds
@@ -314,7 +322,12 @@
def test_1565150(self):
t1 = 1159195039.25
os.utime(self.fname, (t1, t1))
- self.assertEquals(os.stat(self.fname).st_mtime, t1)
+ self.assertEqual(os.stat(self.fname).st_mtime, t1)
+
+ def test_large_time(self):
+ t1 = 5000000000 # some day in 2128
+ os.utime(self.fname, (t1, t1))
+ self.assertEqual(os.stat(self.fname).st_mtime, t1)
def test_1686475(self):
# Verify that an open file can be stat'ed
@@ -346,8 +359,9 @@
def test_update2(self):
if os.path.exists("/bin/sh"):
os.environ.update(HELLO="World")
- value = os.popen("/bin/sh -c 'echo $HELLO'").read().strip()
- self.assertEquals(value, "World")
+ with os.popen("/bin/sh -c 'echo $HELLO'") as popen:
+ value = popen.read().strip()
+ self.assertEqual(value, "World")
class WalkTests(unittest.TestCase):
"""Tests for os.walk()."""
@@ -469,7 +483,7 @@
os.makedirs(path)
# Try paths with a '.' in them
- self.failUnlessRaises(OSError, os.makedirs, os.curdir)
+ self.assertRaises(OSError, os.makedirs, os.curdir)
path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', os.curdir)
os.makedirs(path)
path = os.path.join(base, 'dir1', os.curdir, 'dir2', 'dir3', 'dir4',
@@ -502,18 +516,20 @@
class URandomTests (unittest.TestCase):
def test_urandom(self):
try:
- with test_support.check_warnings():
- self.assertEqual(len(os.urandom(1)), 1)
- self.assertEqual(len(os.urandom(10)), 10)
- self.assertEqual(len(os.urandom(100)), 100)
- self.assertEqual(len(os.urandom(1000)), 1000)
- # see http://bugs.python.org/issue3708
- self.assertEqual(len(os.urandom(0.9)), 0)
- self.assertEqual(len(os.urandom(1.1)), 1)
- self.assertEqual(len(os.urandom(2.0)), 2)
+ self.assertEqual(len(os.urandom(1)), 1)
+ self.assertEqual(len(os.urandom(10)), 10)
+ self.assertEqual(len(os.urandom(100)), 100)
+ self.assertEqual(len(os.urandom(1000)), 1000)
+ # see http://bugs.python.org/issue3708
+ self.assertRaises(TypeError, os.urandom, 0.9)
+ self.assertRaises(TypeError, os.urandom, 1.1)
+ self.assertRaises(TypeError, os.urandom, 2.0)
except NotImplementedError:
pass
+ def test_execvpe_with_bad_arglist(self):
+ self.assertRaises(ValueError, os.execvpe, 'notepad', [], None)
+
class Win32ErrorTests(unittest.TestCase):
def test_rename(self):
self.assertRaises(WindowsError, os.rename, test_support.TESTFN, test_support.TESTFN+".bak")
@@ -525,16 +541,18 @@
self.assertRaises(WindowsError, os.chdir, test_support.TESTFN)
def test_mkdir(self):
- self.assertRaises(WindowsError, os.chdir, test_support.TESTFN)
+ f = open(test_support.TESTFN, "w")
+ try:
+ self.assertRaises(WindowsError, os.mkdir, test_support.TESTFN)
+ finally:
+ f.close()
+ os.unlink(test_support.TESTFN)
def test_utime(self):
self.assertRaises(WindowsError, os.utime, test_support.TESTFN, None)
- def test_access(self):
- self.assertRaises(WindowsError, os.utime, test_support.TESTFN, 0)
-
def test_chmod(self):
- self.assertRaises(WindowsError, os.utime, test_support.TESTFN, 0)
+ self.assertRaises(WindowsError, os.chmod, test_support.TESTFN, 0)
class TestInvalidFD(unittest.TestCase):
singles = ["fchdir", "fdopen", "dup", "fdatasync", "fstat",
@@ -550,7 +568,13 @@
locals()["test_"+f] = get_single(f)
def check(self, f, *args):
- self.assertRaises(OSError, f, test_support.make_bad_fd(), *args)
+ try:
+ f(test_support.make_bad_fd(), *args)
+ except OSError as e:
+ self.assertEqual(e.errno, errno.EBADF)
+ else:
+ self.fail("%r didn't raise a OSError with a bad file descriptor"
+ % f)
def test_isatty(self):
if hasattr(os, "isatty"):
@@ -568,9 +592,8 @@
else:
break
if i < 2:
- # Unable to acquire a range of invalid file descriptors,
- # so skip the test (in 2.6+ this is a unittest.SkipTest).
- return
+ raise unittest.SkipTest(
+ "Unable to acquire a range of invalid file descriptors")
self.assertEqual(os.closerange(fd, fd + i-1), None)
def test_dup2(self):
@@ -589,11 +612,9 @@
if hasattr(os, "fpathconf"):
self.check(os.fpathconf, "PC_NAME_MAX")
- #this is a weird one, it raises IOError unlike the others
def test_ftruncate(self):
if hasattr(os, "ftruncate"):
- self.assertRaises(IOError, os.ftruncate, test_support.make_bad_fd(),
- 0)
+ self.check(os.ftruncate, 0)
def test_lseek(self):
if hasattr(os, "lseek"):
@@ -650,7 +671,6 @@
def test_setreuid_neg1(self):
# Needs to accept -1. We run this in a subprocess to avoid
# altering the test runner's process state (issue8045).
- import subprocess
subprocess.check_call([
sys.executable, '-c',
'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
@@ -665,7 +685,6 @@
def test_setregid_neg1(self):
# Needs to accept -1. We run this in a subprocess to avoid
# altering the test runner's process state (issue8045).
- import subprocess
subprocess.check_call([
sys.executable, '-c',
'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
@@ -673,6 +692,120 @@
class PosixUidGidTests(unittest.TestCase):
pass
+ at unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+class Win32KillTests(unittest.TestCase):
+ def _kill(self, sig):
+ # Start sys.executable as a subprocess and communicate from the
+ # subprocess to the parent that the interpreter is ready. When it
+ # becomes ready, send *sig* via os.kill to the subprocess and check
+ # that the return code is equal to *sig*.
+ import ctypes
+ from ctypes import wintypes
+ import msvcrt
+
+ # Since we can't access the contents of the process' stdout until the
+ # process has exited, use PeekNamedPipe to see what's inside stdout
+ # without waiting. This is done so we can tell that the interpreter
+ # is started and running at a point where it could handle a signal.
+ PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
+ PeekNamedPipe.restype = wintypes.BOOL
+ PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
+ ctypes.POINTER(ctypes.c_char), # stdout buf
+ wintypes.DWORD, # Buffer size
+ ctypes.POINTER(wintypes.DWORD), # bytes read
+ ctypes.POINTER(wintypes.DWORD), # bytes avail
+ ctypes.POINTER(wintypes.DWORD)) # bytes left
+ msg = "running"
+ proc = subprocess.Popen([sys.executable, "-c",
+ "import sys;"
+ "sys.stdout.write('{}');"
+ "sys.stdout.flush();"
+ "input()".format(msg)],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ stdin=subprocess.PIPE)
+ self.addCleanup(proc.stdout.close)
+ self.addCleanup(proc.stderr.close)
+ self.addCleanup(proc.stdin.close)
+
+ count, max = 0, 100
+ while count < max and proc.poll() is None:
+ # Create a string buffer to store the result of stdout from the pipe
+ buf = ctypes.create_string_buffer(len(msg))
+ # Obtain the text currently in proc.stdout
+ # Bytes read/avail/left are left as NULL and unused
+ rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
+ buf, ctypes.sizeof(buf), None, None, None)
+ self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
+ if buf.value:
+ self.assertEqual(msg, buf.value)
+ break
+ time.sleep(0.1)
+ count += 1
+ else:
+ self.fail("Did not receive communication from the subprocess")
+
+ os.kill(proc.pid, sig)
+ self.assertEqual(proc.wait(), sig)
+
+ def test_kill_sigterm(self):
+ # SIGTERM doesn't mean anything special, but make sure it works
+ self._kill(signal.SIGTERM)
+
+ def test_kill_int(self):
+ # os.kill on Windows can take an int which gets set as the exit code
+ self._kill(100)
+
+ def _kill_with_event(self, event, name):
+ tagname = "test_os_%s" % uuid.uuid1()
+ m = mmap.mmap(-1, 1, tagname)
+ m[0] = '0'
+ # Run a script which has console control handling enabled.
+ proc = subprocess.Popen([sys.executable,
+ os.path.join(os.path.dirname(__file__),
+ "win_console_handler.py"), tagname],
+ creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
+ # Let the interpreter startup before we send signals. See #3137.
+ count, max = 0, 20
+ while count < max and proc.poll() is None:
+ if m[0] == '1':
+ break
+ time.sleep(0.5)
+ count += 1
+ else:
+ self.fail("Subprocess didn't finish initialization")
+ os.kill(proc.pid, event)
+ # proc.send_signal(event) could also be done here.
+ # Allow time for the signal to be passed and the process to exit.
+ time.sleep(0.5)
+ if not proc.poll():
+ # Forcefully kill the process if we weren't able to signal it.
+ os.kill(proc.pid, signal.SIGINT)
+ self.fail("subprocess did not stop on {}".format(name))
+
+ @unittest.skip("subprocesses aren't inheriting CTRL+C property")
+ def test_CTRL_C_EVENT(self):
+ from ctypes import wintypes
+ import ctypes
+
+ # Make a NULL value by creating a pointer with no argument.
+ NULL = ctypes.POINTER(ctypes.c_int)()
+ SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
+ SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
+ wintypes.BOOL)
+ SetConsoleCtrlHandler.restype = wintypes.BOOL
+
+ # Calling this with NULL and FALSE causes the calling process to
+ # handle CTRL+C, rather than ignore it. This property is inherited
+ # by subprocesses.
+ SetConsoleCtrlHandler(NULL, 0)
+
+ self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
+
+ def test_CTRL_BREAK_EVENT(self):
+ self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
+
+
def test_main():
test_support.run_unittest(
FileTests,
@@ -685,7 +818,8 @@
URandomTests,
Win32ErrorTests,
TestInvalidFD,
- PosixUidGidTests
+ PosixUidGidTests,
+ Win32KillTests
)
if __name__ == "__main__":
diff --git a/Lib/test/test_random.py b/Lib/test/test_random.py
--- a/Lib/test/test_random.py
+++ b/Lib/test/test_random.py
@@ -5,7 +5,7 @@
import time
import pickle
import warnings
-from math import log, exp, sqrt, pi, fsum as msum
+from math import log, exp, pi, fsum, sin
from functools import reduce
from test import test_support
@@ -53,10 +53,8 @@
state3 = self.gen.getstate() # s/b distinct from state2
self.assertNotEqual(state2, state3)
- with test_support._check_py3k_warnings(quiet=True):
+ with test_support.check_py3k_warnings(quiet=True):
self.assertRaises(TypeError, self.gen.jumpahead) # needs an arg
- self.assertRaises(TypeError, self.gen.jumpahead, "ick") # wrong type
- self.assertRaises(TypeError, self.gen.jumpahead, 2.3) # wrong type
self.assertRaises(TypeError, self.gen.jumpahead, 2, 3) # too many
def test_sample(self):
@@ -69,7 +67,7 @@
self.assertEqual(len(s), k)
uniq = set(s)
self.assertEqual(len(uniq), k)
- self.failUnless(uniq <= set(population))
+ self.assertTrue(uniq <= set(population))
self.assertEqual(self.gen.sample([], 0), []) # test edge case N==k==0
def test_sample_distribution(self):
@@ -114,7 +112,7 @@
samp = self.gen.sample(d, k)
# Verify that we got ints back (keys); the values are complex.
for x in samp:
- self.assert_(type(x) is int)
+ self.assertTrue(type(x) is int)
samp.sort()
self.assertEqual(samp, range(N))
@@ -239,7 +237,7 @@
cum = 0
for i in xrange(100):
r = self.gen.randrange(span)
- self.assert_(0 <= r < span)
+ self.assertTrue(0 <= r < span)
cum |= r
self.assertEqual(cum, span-1)
@@ -249,7 +247,7 @@
stop = self.gen.randrange(2 ** (i-2))
if stop <= start:
return
- self.assert_(start <= self.gen.randrange(start, stop) < stop)
+ self.assertTrue(start <= self.gen.randrange(start, stop) < stop)
def test_rangelimits(self):
for start, stop in [(-2,0), (-(2**60)-2,-(2**60)), (2**60,2**60+2)]:
@@ -259,7 +257,7 @@
def test_genrandbits(self):
# Verify ranges
for k in xrange(1, 1000):
- self.assert_(0 <= self.gen.getrandbits(k) < 2**k)
+ self.assertTrue(0 <= self.gen.getrandbits(k) < 2**k)
# Verify all bits active
getbits = self.gen.getrandbits
@@ -285,17 +283,17 @@
numbits = i+1
k = int(1.00001 + _log(n, 2))
self.assertEqual(k, numbits)
- self.assert_(n == 2**(k-1))
+ self.assertTrue(n == 2**(k-1))
n += n - 1 # check 1 below the next power of two
k = int(1.00001 + _log(n, 2))
- self.assert_(k in [numbits, numbits+1])
- self.assert_(2**k > n > 2**(k-2))
+ self.assertIn(k, [numbits, numbits+1])
+ self.assertTrue(2**k > n > 2**(k-2))
n -= n >> 15 # check a little farther below the next power of two
k = int(1.00001 + _log(n, 2))
self.assertEqual(k, numbits) # note the stronger assertion
- self.assert_(2**k > n > 2**(k-1)) # note the stronger assertion
+ self.assertTrue(2**k > n > 2**(k-1)) # note the stronger assertion
class MersenneTwister_TestBasicOps(TestBasicOps):
@@ -391,7 +389,7 @@
cum = 0
for i in xrange(100):
r = self.gen.randrange(span)
- self.assert_(0 <= r < span)
+ self.assertTrue(0 <= r < span)
cum |= r
self.assertEqual(cum, span-1)
@@ -401,7 +399,7 @@
stop = self.gen.randrange(2 ** (i-2))
if stop <= start:
return
- self.assert_(start <= self.gen.randrange(start, stop) < stop)
+ self.assertTrue(start <= self.gen.randrange(start, stop) < stop)
def test_rangelimits(self):
for start, stop in [(-2,0), (-(2**60)-2,-(2**60)), (2**60,2**60+2)]:
@@ -415,7 +413,7 @@
97904845777343510404718956115L)
# Verify ranges
for k in xrange(1, 1000):
- self.assert_(0 <= self.gen.getrandbits(k) < 2**k)
+ self.assertTrue(0 <= self.gen.getrandbits(k) < 2**k)
# Verify all bits active
getbits = self.gen.getrandbits
@@ -441,35 +439,43 @@
numbits = i+1
k = int(1.00001 + _log(n, 2))
self.assertEqual(k, numbits)
- self.assert_(n == 2**(k-1))
+ self.assertTrue(n == 2**(k-1))
n += n - 1 # check 1 below the next power of two
k = int(1.00001 + _log(n, 2))
- self.assert_(k in [numbits, numbits+1])
- self.assert_(2**k > n > 2**(k-2))
+ self.assertIn(k, [numbits, numbits+1])
+ self.assertTrue(2**k > n > 2**(k-2))
n -= n >> 15 # check a little farther below the next power of two
k = int(1.00001 + _log(n, 2))
self.assertEqual(k, numbits) # note the stronger assertion
- self.assert_(2**k > n > 2**(k-1)) # note the stronger assertion
+ self.assertTrue(2**k > n > 2**(k-1)) # note the stronger assertion
def test_randrange_bug_1590891(self):
start = 1000000000000
stop = -100000000000000000000
step = -200
x = self.gen.randrange(start, stop, step)
- self.assert_(stop < x <= start)
+ self.assertTrue(stop < x <= start)
self.assertEqual((x+stop)%step, 0)
-_gammacoeff = (0.9999999999995183, 676.5203681218835, -1259.139216722289,
- 771.3234287757674, -176.6150291498386, 12.50734324009056,
- -0.1385710331296526, 0.9934937113930748e-05, 0.1659470187408462e-06)
-
-def gamma(z, cof=_gammacoeff, g=7):
- z -= 1.0
- s = msum([cof[0]] + [cof[i] / (z+i) for i in range(1,len(cof))])
- z += 0.5
- return (z+g)**z / exp(z+g) * sqrt(2.0*pi) * s
+def gamma(z, sqrt2pi=(2.0*pi)**0.5):
+ # Reflection to right half of complex plane
+ if z < 0.5:
+ return pi / sin(pi*z) / gamma(1.0-z)
+ # Lanczos approximation with g=7
+ az = z + (7.0 - 0.5)
+ return az ** (z-0.5) / exp(az) * sqrt2pi * fsum([
+ 0.9999999999995183,
+ 676.5203681218835 / z,
+ -1259.139216722289 / (z+1.0),
+ 771.3234287757674 / (z+2.0),
+ -176.6150291498386 / (z+3.0),
+ 12.50734324009056 / (z+4.0),
+ -0.1385710331296526 / (z+5.0),
+ 0.9934937113930748e-05 / (z+6.0),
+ 0.1659470187408462e-06 / (z+7.0),
+ ])
class TestDistributions(unittest.TestCase):
def test_zeroinputs(self):
@@ -528,7 +534,7 @@
def test__all__(self):
# tests validity but not completeness of the __all__ list
- self.failUnless(set(random.__all__) <= set(dir(random)))
+ self.assertTrue(set(random.__all__) <= set(dir(random)))
def test_random_subclass_with_kwargs(self):
# SF bug #1486663 -- this used to erroneously raise a TypeError
diff --git a/Lib/test/test_repr.py b/Lib/test/test_repr.py
--- a/Lib/test/test_repr.py
+++ b/Lib/test/test_repr.py
@@ -8,7 +8,7 @@
import shutil
import unittest
-from test.test_support import run_unittest, _check_py3k_warnings
+from test.test_support import run_unittest, check_py3k_warnings
from repr import repr as r # Don't shadow builtin repr
from repr import Repr
@@ -22,7 +22,7 @@
class ReprTests(unittest.TestCase):
def test_string(self):
- eq = self.assertEquals
+ eq = self.assertEqual
eq(r("abc"), "'abc'")
eq(r("abcdefghijklmnop"),"'abcdefghijklmnop'")
@@ -36,7 +36,7 @@
eq(r(s), expected)
def test_tuple(self):
- eq = self.assertEquals
+ eq = self.assertEqual
eq(r((1,)), "(1,)")
t3 = (1, 2, 3)
@@ -51,7 +51,7 @@
from array import array
from collections import deque
- eq = self.assertEquals
+ eq = self.assertEqual
# Tuples give up after 6 elements
eq(r(()), "()")
eq(r((1,)), "(1,)")
@@ -101,7 +101,7 @@
"array('i', [1, 2, 3, 4, 5, ...])")
def test_numbers(self):
- eq = self.assertEquals
+ eq = self.assertEqual
eq(r(123), repr(123))
eq(r(123L), repr(123L))
eq(r(1.0/3), repr(1.0/3))
@@ -111,7 +111,7 @@
eq(r(n), expected)
def test_instance(self):
- eq = self.assertEquals
+ eq = self.assertEqual
i1 = ClassWithRepr("a")
eq(r(i1), repr(i1))
@@ -123,39 +123,39 @@
eq(r(i3), ("<ClassWithFailingRepr instance at %x>"%id(i3)))
s = r(ClassWithFailingRepr)
- self.failUnless(s.startswith("<class "))
- self.failUnless(s.endswith(">"))
- self.failUnless(s.find("...") == 8)
+ self.assertTrue(s.startswith("<class "))
+ self.assertTrue(s.endswith(">"))
+ self.assertTrue(s.find("...") == 8)
def test_file(self):
fp = open(unittest.__file__)
- self.failUnless(repr(fp).startswith(
+ self.assertTrue(repr(fp).startswith(
"<open file '%s', mode 'r' at 0x" % unittest.__file__))
fp.close()
- self.failUnless(repr(fp).startswith(
+ self.assertTrue(repr(fp).startswith(
"<closed file '%s', mode 'r' at 0x" % unittest.__file__))
def test_lambda(self):
- self.failUnless(repr(lambda x: x).startswith(
+ self.assertTrue(repr(lambda x: x).startswith(
"<function <lambda"))
# XXX anonymous functions? see func_repr
def test_builtin_function(self):
- eq = self.assertEquals
+ eq = self.assertEqual
# Functions
eq(repr(hash), '<built-in function hash>')
# Methods
- self.failUnless(repr(''.split).startswith(
+ self.assertTrue(repr(''.split).startswith(
'<built-in method split of str object at 0x'))
def test_xrange(self):
- eq = self.assertEquals
+ eq = self.assertEqual
eq(repr(xrange(1)), 'xrange(1)')
eq(repr(xrange(1, 2)), 'xrange(1, 2)')
eq(repr(xrange(1, 2, 3)), 'xrange(1, 4, 3)')
def test_nesting(self):
- eq = self.assertEquals
+ eq = self.assertEqual
# everything is meant to give up after 6 levels.
eq(r([[[[[[[]]]]]]]), "[[[[[[[]]]]]]]")
eq(r([[[[[[[[]]]]]]]]), "[[[[[[[...]]]]]]]")
@@ -174,16 +174,16 @@
def test_buffer(self):
# XXX doesn't test buffers with no b_base or read-write buffers (see
# bufferobject.c). The test is fairly incomplete too. Sigh.
- with _check_py3k_warnings():
+ with check_py3k_warnings():
x = buffer('foo')
- self.failUnless(repr(x).startswith('<read-only buffer for 0x'))
+ self.assertTrue(repr(x).startswith('<read-only buffer for 0x'))
def test_cell(self):
# XXX Hmm? How to get at a cell object?
pass
def test_descriptors(self):
- eq = self.assertEquals
+ eq = self.assertEqual
# method descriptors
eq(repr(dict.items), "<method 'items' of 'dict' objects>")
# XXX member descriptors
@@ -193,9 +193,9 @@
class C:
def foo(cls): pass
x = staticmethod(C.foo)
- self.failUnless(repr(x).startswith('<staticmethod object at 0x'))
+ self.assertTrue(repr(x).startswith('<staticmethod object at 0x'))
x = classmethod(C.foo)
- self.failUnless(repr(x).startswith('<classmethod object at 0x'))
+ self.assertTrue(repr(x).startswith('<classmethod object at 0x'))
def test_unsortable(self):
# Repr.repr() used to call sorted() on sets, frozensets and dicts
@@ -244,7 +244,7 @@
del sys.path[0]
def test_module(self):
- eq = self.assertEquals
+ eq = self.assertEqual
touch(os.path.join(self.subpkgname, self.pkgname + os.extsep + 'py'))
from areallylongpackageandmodulenametotestreprtruncation.areallylongpackageandmodulenametotestreprtruncation import areallylongpackageandmodulenametotestreprtruncation
eq(repr(areallylongpackageandmodulenametotestreprtruncation),
@@ -253,7 +253,7 @@
#eq(repr(sys), "<module 'sys' (built-in)>")
def test_type(self):
- eq = self.assertEquals
+ eq = self.assertEqual
touch(os.path.join(self.subpkgname, 'foo'+os.extsep+'py'), '''\
class foo(object):
pass
@@ -274,7 +274,7 @@
''')
from areallylongpackageandmodulenametotestreprtruncation.areallylongpackageandmodulenametotestreprtruncation import bar
# Module name may be prefixed with "test.", depending on how run.
- self.failUnless(repr(bar.bar).startswith(
+ self.assertTrue(repr(bar.bar).startswith(
"<class %s.bar at 0x" % bar.__name__))
def test_instance(self):
@@ -284,11 +284,11 @@
''')
from areallylongpackageandmodulenametotestreprtruncation.areallylongpackageandmodulenametotestreprtruncation import baz
ibaz = baz.baz()
- self.failUnless(repr(ibaz).startswith(
+ self.assertTrue(repr(ibaz).startswith(
"<%s.baz instance at 0x" % baz.__name__))
def test_method(self):
- eq = self.assertEquals
+ eq = self.assertEqual
touch(os.path.join(self.subpkgname, 'qux'+os.extsep+'py'), '''\
class aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:
def amethod(self): pass
@@ -299,7 +299,7 @@
'<unbound method aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.amethod>')
# Bound method next
iqux = qux.aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa()
- self.failUnless(repr(iqux.amethod).startswith(
+ self.assertTrue(repr(iqux.amethod).startswith(
'<bound method aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.amethod of <%s.aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa instance at 0x' \
% (qux.__name__,) ))
@@ -325,8 +325,7 @@
# XXX: Jython lacks the buffer type
del ReprTests.test_buffer
run_unittest(ReprTests)
- if os.name != 'mac':
- run_unittest(LongReprTest)
+ run_unittest(LongReprTest)
if __name__ == "__main__":
diff --git a/Lib/test/test_robotparser.py b/Lib/test/test_robotparser.py
--- a/Lib/test/test_robotparser.py
+++ b/Lib/test/test_robotparser.py
@@ -20,9 +20,9 @@
url = self.url
agent = self.agent
if self.good:
- self.failUnless(self.parser.can_fetch(agent, url))
+ self.assertTrue(self.parser.can_fetch(agent, url))
else:
- self.failIf(self.parser.can_fetch(agent, url))
+ self.assertFalse(self.parser.can_fetch(agent, url))
def __str__(self):
return self.str
@@ -202,7 +202,18 @@
RobotTest(13, doc, good, bad, agent="googlebot")
-# 14. For issue #4108 (obey first * entry)
+# 14. For issue #6325 (query string support)
+doc = """
+User-agent: *
+Disallow: /some/path?name=value
+"""
+
+good = ['/some/path']
+bad = ['/some/path?name=value']
+
+RobotTest(14, doc, good, bad)
+
+# 15. For issue #4108 (obey first * entry)
doc = """
User-agent: *
Disallow: /some/path
@@ -214,22 +225,36 @@
good = ['/another/path']
bad = ['/some/path']
-RobotTest(14, doc, good, bad)
+RobotTest(15, doc, good, bad)
-class TestCase(unittest.TestCase):
- def runTest(self):
+class NetworkTestCase(unittest.TestCase):
+
+ def testPasswordProtectedSite(self):
test_support.requires('network')
- # whole site is password-protected.
- url = 'http://mueblesmoraleda.com'
- parser = robotparser.RobotFileParser()
- parser.set_url(url)
- parser.read()
- self.assertEqual(parser.can_fetch("*", url+"/robots.txt"), False)
+ with test_support.transient_internet('mueblesmoraleda.com'):
+ url = 'http://mueblesmoraleda.com'
+ parser = robotparser.RobotFileParser()
+ parser.set_url(url)
+ try:
+ parser.read()
+ except IOError:
+ self.skipTest('%s is unavailable' % url)
+ self.assertEqual(parser.can_fetch("*", url+"/robots.txt"), False)
+
+ def testPythonOrg(self):
+ test_support.requires('network')
+ with test_support.transient_internet('www.python.org'):
+ parser = robotparser.RobotFileParser(
+ "http://www.python.org/robots.txt")
+ parser.read()
+ self.assertTrue(
+ parser.can_fetch("*", "http://www.python.org/robots.txt"))
+
def test_main():
test_support.run_unittest(tests)
- TestCase().run()
+ test_support.run_unittest(NetworkTestCase)
if __name__=='__main__':
test_support.verbose = 1
diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py
--- a/Lib/test/test_shutil.py
+++ b/Lib/test/test_shutil.py
@@ -7,10 +7,71 @@
import stat
import os
import os.path
+from os.path import splitdrive
+from distutils.spawn import find_executable, spawn
+from shutil import (_make_tarball, _make_zipfile, make_archive,
+ register_archive_format, unregister_archive_format,
+ get_archive_formats)
+import tarfile
+import warnings
+
from test import test_support
-from test.test_support import TESTFN
+from test.test_support import TESTFN, check_warnings, captured_stdout
+
+TESTFN2 = TESTFN + "2"
+
+try:
+ import grp
+ import pwd
+ UID_GID_SUPPORT = True
+except ImportError:
+ UID_GID_SUPPORT = False
+
+try:
+ import zlib
+except ImportError:
+ zlib = None
+
+try:
+ import zipfile
+ ZIP_SUPPORT = True
+except ImportError:
+ ZIP_SUPPORT = find_executable('zip')
class TestShutil(unittest.TestCase):
+
+ def setUp(self):
+ super(TestShutil, self).setUp()
+ self.tempdirs = []
+
+ def tearDown(self):
+ super(TestShutil, self).tearDown()
+ while self.tempdirs:
+ d = self.tempdirs.pop()
+ shutil.rmtree(d, os.name in ('nt', 'cygwin'))
+
+ def write_file(self, path, content='xxx'):
+ """Writes a file in the given path.
+
+
+ path can be a string or a sequence.
+ """
+ if isinstance(path, (list, tuple)):
+ path = os.path.join(*path)
+ f = open(path, 'w')
+ try:
+ f.write(content)
+ finally:
+ f.close()
+
+ def mkdtemp(self):
+ """Create a temporary directory that will be cleaned up.
+
+ Returns the path of the directory.
+ """
+ d = tempfile.mkdtemp()
+ self.tempdirs.append(d)
+ return d
def test_rmtree_errors(self):
# filename is guaranteed not to exist
filename = tempfile.mktemp()
@@ -48,15 +109,29 @@
shutil.rmtree(TESTFN)
def check_args_to_onerror(self, func, arg, exc):
+ # test_rmtree_errors deliberately runs rmtree
+ # on a directory that is chmod 400, which will fail.
+ # This function is run when shutil.rmtree fails.
+ # 99.9% of the time it initially fails to remove
+ # a file in the directory, so the first time through
+ # func is os.remove.
+ # However, some Linux machines running ZFS on
+ # FUSE experienced a failure earlier in the process
+ # at os.listdir. The first failure may legally
+ # be either.
if self.errorState == 0:
- self.assertEqual(func, os.remove)
- self.assertEqual(arg, self.childpath)
- self.failUnless(issubclass(exc[0], OSError))
+ if func is os.remove:
+ self.assertEqual(arg, self.childpath)
+ else:
+ self.assertIs(func, os.listdir,
+ "func must be either os.remove or os.listdir")
+ self.assertEqual(arg, TESTFN)
+ self.assertTrue(issubclass(exc[0], OSError))
self.errorState = 1
else:
self.assertEqual(func, os.rmdir)
self.assertEqual(arg, TESTFN)
- self.failUnless(issubclass(exc[0], OSError))
+ self.assertTrue(issubclass(exc[0], OSError))
self.errorState = 2
def test_rmtree_dont_delete_file(self):
@@ -141,6 +216,7 @@
write_data(join(src_dir, 'test_dir2', 'subdir', 'test.txt'), '456')
write_data(join(src_dir, 'test_dir2', 'subdir2', 'test.py'), '456')
+
# testing glob-like patterns
try:
patterns = shutil.ignore_patterns('*.tmp', 'test_dir2')
@@ -204,7 +280,8 @@
os.link(src, dst)
self.assertRaises(shutil.Error, shutil.copyfile, src, dst)
- self.assertEqual(open(src,'r').read(), 'cheddar')
+ with open(src, 'r') as f:
+ self.assertEqual(f.read(), 'cheddar')
os.remove(dst)
# Using `src` here would mean we end up with a symlink pointing
@@ -212,7 +289,8 @@
# TESTFN/cheese.
os.symlink('cheese', dst)
self.assertRaises(shutil.Error, shutil.copyfile, src, dst)
- self.assertEqual(open(src,'r').read(), 'cheddar')
+ with open(src, 'r') as f:
+ self.assertEqual(f.read(), 'cheddar')
os.remove(dst)
finally:
try:
@@ -232,6 +310,260 @@
finally:
shutil.rmtree(TESTFN, ignore_errors=True)
+ if hasattr(os, "mkfifo"):
+ # Issue #3002: copyfile and copytree block indefinitely on named pipes
+ def test_copyfile_named_pipe(self):
+ os.mkfifo(TESTFN)
+ try:
+ self.assertRaises(shutil.SpecialFileError,
+ shutil.copyfile, TESTFN, TESTFN2)
+ self.assertRaises(shutil.SpecialFileError,
+ shutil.copyfile, __file__, TESTFN)
+ finally:
+ os.remove(TESTFN)
+
+ def test_copytree_named_pipe(self):
+ os.mkdir(TESTFN)
+ try:
+ subdir = os.path.join(TESTFN, "subdir")
+ os.mkdir(subdir)
+ pipe = os.path.join(subdir, "mypipe")
+ os.mkfifo(pipe)
+ try:
+ shutil.copytree(TESTFN, TESTFN2)
+ except shutil.Error as e:
+ errors = e.args[0]
+ self.assertEqual(len(errors), 1)
+ src, dst, error_msg = errors[0]
+ self.assertEqual("`%s` is a named pipe" % pipe, error_msg)
+ else:
+ self.fail("shutil.Error should have been raised")
+ finally:
+ shutil.rmtree(TESTFN, ignore_errors=True)
+ shutil.rmtree(TESTFN2, ignore_errors=True)
+
+ @unittest.skipUnless(zlib, "requires zlib")
+ def test_make_tarball(self):
+ # creating something to tar
+ tmpdir = self.mkdtemp()
+ self.write_file([tmpdir, 'file1'], 'xxx')
+ self.write_file([tmpdir, 'file2'], 'xxx')
+ os.mkdir(os.path.join(tmpdir, 'sub'))
+ self.write_file([tmpdir, 'sub', 'file3'], 'xxx')
+
+ tmpdir2 = self.mkdtemp()
+ unittest.skipUnless(splitdrive(tmpdir)[0] == splitdrive(tmpdir2)[0],
+ "source and target should be on same drive")
+
+ base_name = os.path.join(tmpdir2, 'archive')
+
+ # working with relative paths to avoid tar warnings
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ _make_tarball(splitdrive(base_name)[1], '.')
+ finally:
+ os.chdir(old_dir)
+
+ # check if the compressed tarball was created
+ tarball = base_name + '.tar.gz'
+ self.assertTrue(os.path.exists(tarball))
+
+ # trying an uncompressed one
+ base_name = os.path.join(tmpdir2, 'archive')
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ _make_tarball(splitdrive(base_name)[1], '.', compress=None)
+ finally:
+ os.chdir(old_dir)
+ tarball = base_name + '.tar'
+ self.assertTrue(os.path.exists(tarball))
+
+ def _tarinfo(self, path):
+ tar = tarfile.open(path)
+ try:
+ names = tar.getnames()
+ names.sort()
+ return tuple(names)
+ finally:
+ tar.close()
+
+ def _create_files(self):
+ # creating something to tar
+ tmpdir = self.mkdtemp()
+ dist = os.path.join(tmpdir, 'dist')
+ os.mkdir(dist)
+ self.write_file([dist, 'file1'], 'xxx')
+ self.write_file([dist, 'file2'], 'xxx')
+ os.mkdir(os.path.join(dist, 'sub'))
+ self.write_file([dist, 'sub', 'file3'], 'xxx')
+ os.mkdir(os.path.join(dist, 'sub2'))
+ tmpdir2 = self.mkdtemp()
+ base_name = os.path.join(tmpdir2, 'archive')
+ return tmpdir, tmpdir2, base_name
+
+ @unittest.skipUnless(zlib, "Requires zlib")
+ @unittest.skipUnless(find_executable('tar') and find_executable('gzip'),
+ 'Need the tar command to run')
+ def test_tarfile_vs_tar(self):
+ tmpdir, tmpdir2, base_name = self._create_files()
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ _make_tarball(base_name, 'dist')
+ finally:
+ os.chdir(old_dir)
+
+ # check if the compressed tarball was created
+ tarball = base_name + '.tar.gz'
+ self.assertTrue(os.path.exists(tarball))
+
+ # now create another tarball using `tar`
+ tarball2 = os.path.join(tmpdir, 'archive2.tar.gz')
+ tar_cmd = ['tar', '-cf', 'archive2.tar', 'dist']
+ gzip_cmd = ['gzip', '-f9', 'archive2.tar']
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ with captured_stdout() as s:
+ spawn(tar_cmd)
+ spawn(gzip_cmd)
+ finally:
+ os.chdir(old_dir)
+
+ self.assertTrue(os.path.exists(tarball2))
+ # let's compare both tarballs
+ self.assertEqual(self._tarinfo(tarball), self._tarinfo(tarball2))
+
+ # trying an uncompressed one
+ base_name = os.path.join(tmpdir2, 'archive')
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ _make_tarball(base_name, 'dist', compress=None)
+ finally:
+ os.chdir(old_dir)
+ tarball = base_name + '.tar'
+ self.assertTrue(os.path.exists(tarball))
+
+ # now for a dry_run
+ base_name = os.path.join(tmpdir2, 'archive')
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ try:
+ _make_tarball(base_name, 'dist', compress=None, dry_run=True)
+ finally:
+ os.chdir(old_dir)
+ tarball = base_name + '.tar'
+ self.assertTrue(os.path.exists(tarball))
+
+ @unittest.skipUnless(zlib, "Requires zlib")
+ @unittest.skipUnless(ZIP_SUPPORT, 'Need zip support to run')
+ def test_make_zipfile(self):
+ # creating something to tar
+ tmpdir = self.mkdtemp()
+ self.write_file([tmpdir, 'file1'], 'xxx')
+ self.write_file([tmpdir, 'file2'], 'xxx')
+
+ tmpdir2 = self.mkdtemp()
+ base_name = os.path.join(tmpdir2, 'archive')
+ _make_zipfile(base_name, tmpdir)
+
+ # check if the compressed tarball was created
+ tarball = base_name + '.zip'
+ self.assertTrue(os.path.exists(tarball))
+
+
+ def test_make_archive(self):
+ tmpdir = self.mkdtemp()
+ base_name = os.path.join(tmpdir, 'archive')
+ self.assertRaises(ValueError, make_archive, base_name, 'xxx')
+
+ @unittest.skipUnless(zlib, "Requires zlib")
+ def test_make_archive_owner_group(self):
+ # testing make_archive with owner and group, with various combinations
+ # this works even if there's not gid/uid support
+ if UID_GID_SUPPORT:
+ group = grp.getgrgid(0)[0]
+ owner = pwd.getpwuid(0)[0]
+ else:
+ group = owner = 'root'
+
+ base_dir, root_dir, base_name = self._create_files()
+ base_name = os.path.join(self.mkdtemp() , 'archive')
+ res = make_archive(base_name, 'zip', root_dir, base_dir, owner=owner,
+ group=group)
+ self.assertTrue(os.path.exists(res))
+
+ res = make_archive(base_name, 'zip', root_dir, base_dir)
+ self.assertTrue(os.path.exists(res))
+
+ res = make_archive(base_name, 'tar', root_dir, base_dir,
+ owner=owner, group=group)
+ self.assertTrue(os.path.exists(res))
+
+ res = make_archive(base_name, 'tar', root_dir, base_dir,
+ owner='kjhkjhkjg', group='oihohoh')
+ self.assertTrue(os.path.exists(res))
+
+ @unittest.skipUnless(zlib, "Requires zlib")
+ @unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support")
+ def test_tarfile_root_owner(self):
+ tmpdir, tmpdir2, base_name = self._create_files()
+ old_dir = os.getcwd()
+ os.chdir(tmpdir)
+ group = grp.getgrgid(0)[0]
+ owner = pwd.getpwuid(0)[0]
+ try:
+ archive_name = _make_tarball(base_name, 'dist', compress=None,
+ owner=owner, group=group)
+ finally:
+ os.chdir(old_dir)
+
+ # check if the compressed tarball was created
+ self.assertTrue(os.path.exists(archive_name))
+
+ # now checks the rights
+ archive = tarfile.open(archive_name)
+ try:
+ for member in archive.getmembers():
+ self.assertEqual(member.uid, 0)
+ self.assertEqual(member.gid, 0)
+ finally:
+ archive.close()
+
+ def test_make_archive_cwd(self):
+ current_dir = os.getcwd()
+ def _breaks(*args, **kw):
+ raise RuntimeError()
+
+ register_archive_format('xxx', _breaks, [], 'xxx file')
+ try:
+ try:
+ make_archive('xxx', 'xxx', root_dir=self.mkdtemp())
+ except Exception:
+ pass
+ self.assertEqual(os.getcwd(), current_dir)
+ finally:
+ unregister_archive_format('xxx')
+
+ def test_register_archive_format(self):
+
+ self.assertRaises(TypeError, register_archive_format, 'xxx', 1)
+ self.assertRaises(TypeError, register_archive_format, 'xxx', lambda: x,
+ 1)
+ self.assertRaises(TypeError, register_archive_format, 'xxx', lambda: x,
+ [(1, 2), (1, 2, 3)])
+
+ register_archive_format('xxx', lambda: x, [(1, 2)], 'xxx file')
+ formats = [name for name, params in get_archive_formats()]
+ self.assertIn('xxx', formats)
+
+ unregister_archive_format('xxx')
+ formats = [name for name, params in get_archive_formats()]
+ self.assertNotIn('xxx', formats)
+
class TestMove(unittest.TestCase):
@@ -262,9 +594,11 @@
pass
def _check_move_file(self, src, dst, real_dst):
- contents = open(src, "rb").read()
+ with open(src, "rb") as f:
+ contents = f.read()
shutil.move(src, dst)
- self.assertEqual(contents, open(real_dst, "rb").read())
+ with open(real_dst, "rb") as f:
+ self.assertEqual(contents, f.read())
self.assertFalse(os.path.exists(src))
def _check_move_dir(self, src, dst, real_dst):
@@ -352,8 +686,8 @@
for src, dst in [('srcdir', 'srcdir/dest')]:
src = os.path.join(TESTFN, src)
dst = os.path.join(TESTFN, dst)
- self.assert_(shutil.destinsrc(src, dst),
- msg='destinsrc() wrongly concluded that '
+ self.assertTrue(shutil._destinsrc(src, dst),
+ msg='_destinsrc() wrongly concluded that '
'dst (%s) is not in src (%s)' % (dst, src))
finally:
shutil.rmtree(TESTFN, ignore_errors=True)
@@ -364,8 +698,8 @@
for src, dst in [('srcdir', 'src/dest'), ('srcdir', 'srcdir.new')]:
src = os.path.join(TESTFN, src)
dst = os.path.join(TESTFN, dst)
- self.failIf(shutil.destinsrc(src, dst),
- msg='destinsrc() wrongly concluded that '
+ self.assertFalse(shutil._destinsrc(src, dst),
+ msg='_destinsrc() wrongly concluded that '
'dst (%s) is in src (%s)' % (dst, src))
finally:
shutil.rmtree(TESTFN, ignore_errors=True)
diff --git a/Lib/test/test_time.py b/Lib/test/test_time.py
--- a/Lib/test/test_time.py
+++ b/Lib/test/test_time.py
@@ -18,9 +18,9 @@
time.clock()
def test_conversions(self):
- self.assert_(time.ctime(self.t)
+ self.assertTrue(time.ctime(self.t)
== time.asctime(time.localtime(self.t)))
- self.assert_(long(time.mktime(time.localtime(self.t)))
+ self.assertTrue(long(time.mktime(time.localtime(self.t)))
== long(self.t))
def test_sleep(self):
@@ -89,11 +89,6 @@
(1900, 1, 1, 0, 0, 0, 0, -1, -1))
self.assertRaises(ValueError, time.strftime, '',
(1900, 1, 1, 0, 0, 0, 0, 367, -1))
- # Check daylight savings flag [-1, 1]
- self.assertRaises(ValueError, time.strftime, '',
- (1900, 1, 1, 0, 0, 0, 0, 1, -2))
- self.assertRaises(ValueError, time.strftime, '',
- (1900, 1, 1, 0, 0, 0, 0, 1, 2))
def test_default_values_for_zero(self):
# Make sure that using all zeros uses the proper default values.
@@ -107,7 +102,7 @@
# and it is not converted to 2000.
expected = "0000 01 01 00 00 00 1 001"
result = time.strftime("%Y %m %d %H %M %S %w %j", (0,)*9)
- self.assertEquals(expected, result)
+ self.assertEqual(expected, result)
def test_strptime(self):
# Should be able to go round-trip from strftime to strptime without
@@ -140,13 +135,13 @@
# (12345, 1, 0, 0, 0, 0, 0, 0, 0))
# XXX: For now, just make sure we don't have a crash:
try:
- time.asctime((12345, 1, 0, 0, 0, 0, 0, 0, 0))
+ time.asctime((12345, 1, 1, 0, 0, 0, 0, 1, 0))
except ValueError:
pass
+ @unittest.skipIf(not hasattr(time, "tzset"),
+ "time module has no attribute tzset")
def test_tzset(self):
- if not hasattr(time, "tzset"):
- return # Can't test this; don't want the test suite to fail
from os import environ
@@ -172,36 +167,36 @@
time.tzset()
environ['TZ'] = utc
time.tzset()
- self.failUnlessEqual(
+ self.assertEqual(
time.gmtime(xmas2002), time.localtime(xmas2002)
)
- self.failUnlessEqual(time.daylight, 0)
- self.failUnlessEqual(time.timezone, 0)
- self.failUnlessEqual(time.localtime(xmas2002).tm_isdst, 0)
+ self.assertEqual(time.daylight, 0)
+ self.assertEqual(time.timezone, 0)
+ self.assertEqual(time.localtime(xmas2002).tm_isdst, 0)
# Make sure we can switch to US/Eastern
environ['TZ'] = eastern
time.tzset()
- self.failIfEqual(time.gmtime(xmas2002), time.localtime(xmas2002))
- self.failUnlessEqual(time.tzname, ('EST', 'EDT'))
- self.failUnlessEqual(len(time.tzname), 2)
- self.failUnlessEqual(time.daylight, 1)
- self.failUnlessEqual(time.timezone, 18000)
- self.failUnlessEqual(time.altzone, 14400)
- self.failUnlessEqual(time.localtime(xmas2002).tm_isdst, 0)
- self.failUnlessEqual(len(time.tzname), 2)
+ self.assertNotEqual(time.gmtime(xmas2002), time.localtime(xmas2002))
+ self.assertEqual(time.tzname, ('EST', 'EDT'))
+ self.assertEqual(len(time.tzname), 2)
+ self.assertEqual(time.daylight, 1)
+ self.assertEqual(time.timezone, 18000)
+ self.assertEqual(time.altzone, 14400)
+ self.assertEqual(time.localtime(xmas2002).tm_isdst, 0)
+ self.assertEqual(len(time.tzname), 2)
# Now go to the southern hemisphere.
environ['TZ'] = victoria
time.tzset()
- self.failIfEqual(time.gmtime(xmas2002), time.localtime(xmas2002))
- self.failUnless(time.tzname[0] == 'AEST', str(time.tzname[0]))
- self.failUnless(time.tzname[1] == 'AEDT', str(time.tzname[1]))
- self.failUnlessEqual(len(time.tzname), 2)
- self.failUnlessEqual(time.daylight, 1)
- self.failUnlessEqual(time.timezone, -36000)
- self.failUnlessEqual(time.altzone, -39600)
- self.failUnlessEqual(time.localtime(xmas2002).tm_isdst, 1)
+ self.assertNotEqual(time.gmtime(xmas2002), time.localtime(xmas2002))
+ self.assertTrue(time.tzname[0] == 'AEST', str(time.tzname[0]))
+ self.assertTrue(time.tzname[1] == 'AEDT', str(time.tzname[1]))
+ self.assertEqual(len(time.tzname), 2)
+ self.assertEqual(time.daylight, 1)
+ self.assertEqual(time.timezone, -36000)
+ self.assertEqual(time.altzone, -39600)
+ self.assertEqual(time.localtime(xmas2002).tm_isdst, 1)
finally:
# Repair TZ environment variable in case any other tests
@@ -233,14 +228,25 @@
gt1 = time.gmtime(None)
t0 = time.mktime(gt0)
t1 = time.mktime(gt1)
- self.assert_(0 <= (t1-t0) < 0.2)
+ self.assertTrue(0 <= (t1-t0) < 0.2)
def test_localtime_without_arg(self):
lt0 = time.localtime()
lt1 = time.localtime(None)
t0 = time.mktime(lt0)
t1 = time.mktime(lt1)
- self.assert_(0 <= (t1-t0) < 0.2)
+ self.assertTrue(0 <= (t1-t0) < 0.2)
+
+ def test_mktime(self):
+ # Issue #1726687
+ for t in (-2, -1, 0, 1):
+ try:
+ tt = time.localtime(t)
+ except (OverflowError, ValueError):
+ pass
+ else:
+ self.assertEqual(time.mktime(tt), t)
+
def test_main():
test_support.run_unittest(TimeTestCase)
diff --git a/Lib/test/test_urllib2.py b/Lib/test/test_urllib2.py
--- a/Lib/test/test_urllib2.py
+++ b/Lib/test/test_urllib2.py
@@ -24,9 +24,7 @@
# And more hacking to get it to work on MacOS. This assumes
# urllib.pathname2url works, unfortunately...
- if os.name == 'mac':
- fname = '/' + fname.replace(':', '/')
- elif os.name == 'riscos':
+ if os.name == 'riscos':
import string
fname = os.expand(fname)
fname = fname.translate(string.maketrans("/.", "./"))
@@ -47,7 +45,7 @@
('a, b, "c", "d", "e,f", g, h', ['a', 'b', '"c"', '"d"', '"e,f"', 'g', 'h']),
('a="b\\"c", d="e\\,f", g="h\\\\i"', ['a="b"c"', 'd="e,f"', 'g="h\\i"'])]
for string, list in tests:
- self.assertEquals(urllib2.parse_http_list(string), list)
+ self.assertEqual(urllib2.parse_http_list(string), list)
def test_request_headers_dict():
@@ -288,7 +286,7 @@
def set_debuglevel(self, level):
self.level = level
- def _set_tunnel(self, host, port=None, headers=None):
+ def set_tunnel(self, host, port=None, headers=None):
self._tunnel_host = host
self._tunnel_port = port
if headers:
@@ -580,15 +578,15 @@
# *_request
self.assertEqual((handler, name), calls[i])
self.assertEqual(len(args), 1)
- self.assert_(isinstance(args[0], Request))
+ self.assertIsInstance(args[0], Request)
else:
# *_response
self.assertEqual((handler, name), calls[i])
self.assertEqual(len(args), 2)
- self.assert_(isinstance(args[0], Request))
+ self.assertIsInstance(args[0], Request)
# response from opener.open is None, because there's no
# handler that defines http_open to handle it
- self.assert_(args[1] is None or
+ self.assertTrue(args[1] is None or
isinstance(args[1], MockResponse))
@@ -625,22 +623,32 @@
h = NullFTPHandler(data)
o = h.parent = MockOpener()
- for url, host, port, type_, dirs, filename, mimetype in [
+ for url, host, port, user, passwd, type_, dirs, filename, mimetype in [
("ftp://localhost/foo/bar/baz.html",
- "localhost", ftplib.FTP_PORT, "I",
+ "localhost", ftplib.FTP_PORT, "", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "parrot", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://%25parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "%parrot", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://%2542parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "%42parrot", "", "I",
["foo", "bar"], "baz.html", "text/html"),
("ftp://localhost:80/foo/bar/",
- "localhost", 80, "D",
+ "localhost", 80, "", "", "D",
["foo", "bar"], "", None),
("ftp://localhost/baz.gif;type=a",
- "localhost", ftplib.FTP_PORT, "A",
+ "localhost", ftplib.FTP_PORT, "", "", "A",
[], "baz.gif", None), # XXX really this should guess image/gif
]:
req = Request(url)
req.timeout = None
r = h.ftp_open(req)
# ftp authentication not yet implemented by FTPHandler
- self.assert_(h.user == h.passwd == "")
+ self.assertEqual(h.user, user)
+ self.assertEqual(h.passwd, passwd)
self.assertEqual(h.host, socket.gethostbyname(host))
self.assertEqual(h.port, port)
self.assertEqual(h.dirs, dirs)
@@ -737,11 +745,11 @@
h.file_open(req)
# XXXX remove OSError when bug fixed
except (urllib2.URLError, OSError):
- self.assert_(not ftp)
+ self.assertTrue(not ftp)
else:
- self.assert_(o.req is req)
+ self.assertTrue(o.req is req)
self.assertEqual(req.type, "ftp")
- self.assertEqual(req.type is "ftp", ftp)
+ self.assertEqual(req.type == "ftp", ftp)
def test_http(self):
@@ -784,8 +792,8 @@
r = MockResponse(200, "OK", {}, "")
newreq = h.do_request_(req)
if data is None: # GET
- self.assert_("Content-length" not in req.unredirected_hdrs)
- self.assert_("Content-type" not in req.unredirected_hdrs)
+ self.assertNotIn("Content-length", req.unredirected_hdrs)
+ self.assertNotIn("Content-type", req.unredirected_hdrs)
else: # POST
self.assertEqual(req.unredirected_hdrs["Content-length"], "0")
self.assertEqual(req.unredirected_hdrs["Content-type"],
@@ -831,6 +839,25 @@
p_ds_req = h.do_request_(ds_req)
self.assertEqual(p_ds_req.unredirected_hdrs["Host"],"example.com")
+ def test_fixpath_in_weirdurls(self):
+ # Issue4493: urllib2 to supply '/' when to urls where path does not
+ # start with'/'
+
+ h = urllib2.AbstractHTTPHandler()
+ o = h.parent = MockOpener()
+
+ weird_url = 'http://www.python.org?getspam'
+ req = Request(weird_url)
+ newreq = h.do_request_(req)
+ self.assertEqual(newreq.get_host(),'www.python.org')
+ self.assertEqual(newreq.get_selector(),'/?getspam')
+
+ url_without_path = 'http://www.python.org'
+ req = Request(url_without_path)
+ newreq = h.do_request_(req)
+ self.assertEqual(newreq.get_host(),'www.python.org')
+ self.assertEqual(newreq.get_selector(),'')
+
def test_errors(self):
h = urllib2.HTTPErrorProcessor()
o = h.parent = MockOpener()
@@ -840,19 +867,19 @@
# all 2xx are passed through
r = MockResponse(200, "OK", {}, "", url)
newr = h.http_response(req, r)
- self.assert_(r is newr)
- self.assert_(not hasattr(o, "proto")) # o.error not called
+ self.assertTrue(r is newr)
+ self.assertTrue(not hasattr(o, "proto")) # o.error not called
r = MockResponse(202, "Accepted", {}, "", url)
newr = h.http_response(req, r)
- self.assert_(r is newr)
- self.assert_(not hasattr(o, "proto")) # o.error not called
+ self.assertTrue(r is newr)
+ self.assertTrue(not hasattr(o, "proto")) # o.error not called
r = MockResponse(206, "Partial content", {}, "", url)
newr = h.http_response(req, r)
- self.assert_(r is newr)
- self.assert_(not hasattr(o, "proto")) # o.error not called
+ self.assertTrue(r is newr)
+ self.assertTrue(not hasattr(o, "proto")) # o.error not called
# anything else calls o.error (and MockOpener returns None, here)
r = MockResponse(502, "Bad gateway", {}, "", url)
- self.assert_(h.http_response(req, r) is None)
+ self.assertTrue(h.http_response(req, r) is None)
self.assertEqual(o.proto, "http") # o.error called
self.assertEqual(o.args, (req, r, 502, "Bad gateway", {}))
@@ -864,12 +891,12 @@
req = Request("http://example.com/")
r = MockResponse(200, "OK", {}, "")
newreq = h.http_request(req)
- self.assert_(cj.ach_req is req is newreq)
- self.assertEquals(req.get_origin_req_host(), "example.com")
- self.assert_(not req.is_unverifiable())
+ self.assertTrue(cj.ach_req is req is newreq)
+ self.assertEqual(req.get_origin_req_host(), "example.com")
+ self.assertTrue(not req.is_unverifiable())
newr = h.http_response(req, r)
- self.assert_(cj.ec_req is req)
- self.assert_(cj.ec_r is r is newr)
+ self.assertTrue(cj.ec_req is req)
+ self.assertTrue(cj.ec_r is r is newr)
def test_redirect(self):
from_url = "http://example.com/a.html"
@@ -892,23 +919,23 @@
MockHeaders({"location": to_url}))
except urllib2.HTTPError:
# 307 in response to POST requires user OK
- self.assert_(code == 307 and data is not None)
+ self.assertTrue(code == 307 and data is not None)
self.assertEqual(o.req.get_full_url(), to_url)
try:
self.assertEqual(o.req.get_method(), "GET")
except AttributeError:
- self.assert_(not o.req.has_data())
+ self.assertTrue(not o.req.has_data())
# now it's a GET, there should not be headers regarding content
# (possibly dragged from before being a POST)
headers = [x.lower() for x in o.req.headers]
- self.assertTrue("content-length" not in headers)
- self.assertTrue("content-type" not in headers)
+ self.assertNotIn("content-length", headers)
+ self.assertNotIn("content-type", headers)
self.assertEqual(o.req.headers["Nonsense"],
"viking=withhold")
- self.assert_("Spam" not in o.req.headers)
- self.assert_("Spam" not in o.req.unredirected_hdrs)
+ self.assertNotIn("Spam", o.req.headers)
+ self.assertNotIn("Spam", o.req.unredirected_hdrs)
# loop detection
req = Request(from_url)
@@ -943,28 +970,6 @@
self.assertEqual(count,
urllib2.HTTPRedirectHandler.max_redirections)
- def test_invalid_redirect(self):
- from_url = "http://example.com/a.html"
- valid_schemes = ['http', 'https', 'ftp']
- invalid_schemes = ['file', 'imap', 'ldap']
- schemeless_url = "example.com/b.html"
- h = urllib2.HTTPRedirectHandler()
- o = h.parent = MockOpener()
- req = Request(from_url)
- req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
-
- for scheme in invalid_schemes:
- invalid_url = scheme + '://' + schemeless_url
- self.assertRaises(urllib2.HTTPError, h.http_error_302,
- req, MockFile(), 302, "Security Loophole",
- MockHeaders({"location": invalid_url}))
-
- for scheme in valid_schemes:
- valid_url = scheme + '://' + schemeless_url
- h.http_error_302(req, MockFile(), 302, "That's fine",
- MockHeaders({"location": valid_url}))
- self.assertEqual(o.req.get_full_url(), valid_url)
-
def test_cookie_redirect(self):
# cookies shouldn't leak into redirected requests
from cookielib import CookieJar
@@ -979,7 +984,7 @@
cp = urllib2.HTTPCookieProcessor(cj)
o = build_test_opener(hh, hdeh, hrh, cp)
o.open("http://www.example.com/")
- self.assert_(not hh.req.has_header("Cookie"))
+ self.assertTrue(not hh.req.has_header("Cookie"))
def test_proxy(self):
o = OpenerDirector()
@@ -1039,16 +1044,16 @@
req.add_header("Proxy-Authorization","FooBar")
req.add_header("User-Agent","Grail")
self.assertEqual(req.get_host(), "www.example.com")
- self.assertTrue(req._tunnel_host is None)
+ self.assertIsNone(req._tunnel_host)
r = o.open(req)
# Verify Proxy-Authorization gets tunneled to request.
# httpsconn req_headers do not have the Proxy-Authorization header but
# the req will have.
- self.assertFalse(("Proxy-Authorization","FooBar") in
+ self.assertNotIn(("Proxy-Authorization","FooBar"),
https_handler.httpconn.req_headers)
- self.assertTrue(("User-Agent","Grail") in
- https_handler.httpconn.req_headers)
- self.assertFalse(req._tunnel_host is None)
+ self.assertIn(("User-Agent","Grail"),
+ https_handler.httpconn.req_headers)
+ self.assertIsNotNone(req._tunnel_host)
self.assertEqual(req.get_host(), "proxy.example.com:3128")
self.assertEqual(req.get_header("Proxy-authorization"),"FooBar")
@@ -1215,7 +1220,7 @@
if h.__class__ == handler_class:
break
else:
- self.assert_(False)
+ self.assertTrue(False)
class RequestTests(unittest.TestCase):
@@ -1230,10 +1235,10 @@
self.assertEqual("GET", self.get.get_method())
def test_add_data(self):
- self.assert_(not self.get.has_data())
+ self.assertTrue(not self.get.has_data())
self.assertEqual("GET", self.get.get_method())
self.get.add_data("spam")
- self.assert_(self.get.has_data())
+ self.assertTrue(self.get.has_data())
self.assertEqual("POST", self.get.get_method())
def test_get_full_url(self):
@@ -1256,12 +1261,22 @@
self.assertEqual("www.python.org", req.get_host())
def test_proxy(self):
- self.assert_(not self.get.has_proxy())
+ self.assertTrue(not self.get.has_proxy())
self.get.set_proxy("www.perl.org", "http")
- self.assert_(self.get.has_proxy())
+ self.assertTrue(self.get.has_proxy())
self.assertEqual("www.python.org", self.get.get_origin_req_host())
self.assertEqual("www.perl.org", self.get.get_host())
+ def test_wrapped_url(self):
+ req = Request("<URL:http://www.python.org>")
+ self.assertEqual("www.python.org", req.get_host())
+
+ def test_urlwith_fragment(self):
+ req = Request("http://www.python.org/?qs=query#fragment=true")
+ self.assertEqual("/?qs=query", req.get_selector())
+ req = Request("http://www.python.org/#fun=true")
+ self.assertEqual("/", req.get_selector())
+
def test_main(verbose=None):
from test import test_urllib2
diff --git a/Lib/timeit.py b/Lib/timeit.py
--- a/Lib/timeit.py
+++ b/Lib/timeit.py
@@ -127,7 +127,7 @@
if isinstance(setup, basestring):
setup = reindent(setup, 4)
src = template % {'stmt': stmt, 'setup': setup}
- elif callable(setup):
+ elif hasattr(setup, '__call__'):
src = template % {'stmt': stmt, 'setup': '_setup()'}
ns['_setup'] = setup
else:
@@ -136,13 +136,13 @@
code = compile(src, dummy_src_name, "exec")
exec code in globals(), ns
self.inner = ns["inner"]
- elif callable(stmt):
+ elif hasattr(stmt, '__call__'):
self.src = None
if isinstance(setup, basestring):
_setup = setup
def setup():
exec _setup in globals(), ns
- elif not callable(setup):
+ elif not hasattr(setup, '__call__'):
raise ValueError("setup is neither a string nor callable")
self.inner = _template_func(setup, stmt)
else:
diff --git a/Lib/types.py b/Lib/types.py
--- a/Lib/types.py
+++ b/Lib/types.py
@@ -43,11 +43,7 @@
def _f(): pass
FunctionType = type(_f)
LambdaType = type(lambda: None) # Same as FunctionType
-try:
- CodeType = type(_f.func_code)
-except RuntimeError:
- # Execution in restricted environment
- pass
+CodeType = type(_f.func_code)
def _g():
yield 1
@@ -73,15 +69,10 @@
try:
raise TypeError
except TypeError:
- try:
- tb = sys.exc_info()[2]
- TracebackType = type(tb)
- FrameType = type(tb.tb_frame)
- except AttributeError:
- # In the restricted environment, exc_info returns (None, None,
- # None) Then, tb.tb_frame gives an attribute error
- pass
- tb = None; del tb
+ tb = sys.exc_info()[2]
+ TracebackType = type(tb)
+ FrameType = type(tb.tb_frame)
+ del tb
SliceType = slice
EllipsisType = type(Ellipsis)
diff --git a/Lib/warnings.py b/Lib/warnings.py
--- a/Lib/warnings.py
+++ b/Lib/warnings.py
@@ -46,7 +46,14 @@
append=0):
"""Insert an entry into the list of warnings filters (at the front).
- Use assertions to check that all arguments have the right type."""
+ 'action' -- one of "error", "ignore", "always", "default", "module",
+ or "once"
+ 'message' -- a regex that the warning message must match
+ 'category' -- a class that the warning must be a subclass of
+ 'module' -- a regex that the module name must match
+ 'lineno' -- an integer line number, 0 matches all warnings
+ 'append' -- if true, append to the list of filters
+ """
import re
assert action in ("error", "ignore", "always", "default", "module",
"once"), "invalid action: %r" % (action,)
@@ -68,6 +75,11 @@
"""Insert a simple entry into the list of warnings filters (at the front).
A simple filter matches all modules and messages.
+ 'action' -- one of "error", "ignore", "always", "default", "module",
+ or "once"
+ 'category' -- a class that the warning must be a subclass of
+ 'lineno' -- an integer line number, 0 matches all warnings
+ 'append' -- if true, append to the list of filters
"""
assert action in ("error", "ignore", "always", "default", "module",
"once"), "invalid action: %r" % (action,)
@@ -264,24 +276,6 @@
raise RuntimeError(
"Unrecognized action (%r) in warnings.filters:\n %s" %
(action, item))
- # Warn if showwarning() does not support the 'line' argument.
- # Don't use 'inspect' as it relies on an extension module, which break the
- # build thanks to 'warnings' being imported by setup.py.
- fxn_code = None
- if hasattr(showwarning, 'func_code'):
- fxn_code = showwarning.func_code
- elif hasattr(showwarning, '__func__'):
- fxn_code = showwarning.__func__.func_code
- if fxn_code:
- args = fxn_code.co_varnames[:fxn_code.co_argcount]
- CO_VARARGS = 0x4
- if 'line' not in args and not fxn_code.co_flags & CO_VARARGS:
- showwarning_msg = ("functions overriding warnings.showwarning() "
- "must support the 'line' argument")
- if message == showwarning_msg:
- _show_warning(message, category, filename, lineno)
- else:
- warn(showwarning_msg, DeprecationWarning)
# Print message and context
showwarning(message, category, filename, lineno)
@@ -391,8 +385,12 @@
# Module initialization
_processoptions(sys.warnoptions)
if not _warnings_defaults:
- simplefilter("ignore", category=PendingDeprecationWarning, append=1)
- simplefilter("ignore", category=ImportWarning, append=1)
+ silence = [ImportWarning, PendingDeprecationWarning]
+ # Don't silence DeprecationWarning if -3 or -Q was used.
+ if not sys.py3kwarning and not sys.flags.division_warning:
+ silence.append(DeprecationWarning)
+ for cls in silence:
+ simplefilter("ignore", category=cls)
bytes_warning = sys.flags.bytes_warning
if bytes_warning > 1:
bytes_action = "error"
diff --git a/Lib/weakref.py b/Lib/weakref.py
--- a/Lib/weakref.py
+++ b/Lib/weakref.py
@@ -20,6 +20,8 @@
ProxyType,
ReferenceType)
+from _weakrefset import WeakSet
+
from exceptions import ReferenceError
@@ -27,7 +29,7 @@
__all__ = ["ref", "proxy", "getweakrefcount", "getweakrefs",
"WeakKeyDictionary", "ReferenceError", "ReferenceType", "ProxyType",
- "CallableProxyType", "ProxyTypes", "WeakValueDictionary"]
+ "CallableProxyType", "ProxyTypes", "WeakValueDictionary", 'WeakSet']
class WeakValueDictionary(UserDict.UserDict):
@@ -88,6 +90,17 @@
new[key] = o
return new
+ __copy__ = copy
+
+ def __deepcopy__(self, memo):
+ from copy import deepcopy
+ new = self.__class__()
+ for key, wr in self.data.items():
+ o = wr()
+ if o is not None:
+ new[deepcopy(key, memo)] = o
+ return new
+
def get(self, key, default=None):
try:
wr = self.data[key]
@@ -262,6 +275,17 @@
new[o] = value
return new
+ __copy__ = copy
+
+ def __deepcopy__(self, memo):
+ from copy import deepcopy
+ new = self.__class__()
+ for key, value in self.data.items():
+ o = key()
+ if o is not None:
+ new[o] = deepcopy(value, memo)
+ return new
+
def get(self, key, default=None):
return self.data.get(ref(key),default)
diff --git a/Lib/zipfile.py b/Lib/zipfile.py
--- a/Lib/zipfile.py
+++ b/Lib/zipfile.py
@@ -3,6 +3,8 @@
"""
import struct, os, time, sys, shutil
import binascii, cStringIO, stat
+import io
+import re
try:
import zlib # We may need its compression method
@@ -130,23 +132,41 @@
_CD64_DIRECTORY_SIZE = 8
_CD64_OFFSET_START_CENTDIR = 9
-def is_zipfile(filename):
- """Quickly see if file is a ZIP file by checking the magic number."""
+def _check_zipfile(fp):
try:
- fpin = open(filename, "rb")
- endrec = _EndRecData(fpin)
- fpin.close()
- if endrec:
- return True # file has correct magic number
+ if _EndRecData(fp):
+ return True # file has correct magic number
except IOError:
pass
return False
+def is_zipfile(filename):
+ """Quickly see if a file is a ZIP file by checking the magic number.
+
+ The filename argument may be a file or file-like object too.
+ """
+ result = False
+ try:
+ if hasattr(filename, "read"):
+ result = _check_zipfile(fp=filename)
+ else:
+ with open(filename, "rb") as fp:
+ result = _check_zipfile(fp)
+ except IOError:
+ pass
+ return result
+
def _EndRecData64(fpin, offset, endrec):
"""
Read the ZIP64 end-of-archive records and use that to update endrec
"""
- fpin.seek(offset - sizeEndCentDir64Locator, 2)
+ try:
+ fpin.seek(offset - sizeEndCentDir64Locator, 2)
+ except IOError:
+ # If the seek fails, the file is not large enough to contain a ZIP64
+ # end-of-archive record, so just return the end record we were given.
+ return endrec
+
data = fpin.read(sizeEndCentDir64Locator)
sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data)
if sig != stringEndArchive64Locator:
@@ -441,198 +461,194 @@
self._UpdateKeys(c)
return c
-class ZipExtFile:
+class ZipExtFile(io.BufferedIOBase):
"""File-like object for reading an archive member.
Is returned by ZipFile.open().
"""
- def __init__(self, fileobj, zipinfo, decrypt=None):
- self.fileobj = fileobj
- self.decrypter = decrypt
- self.bytes_read = 0L
- self.rawbuffer = ''
- self.readbuffer = ''
- self.linebuffer = ''
- self.eof = False
- self.univ_newlines = False
- self.nlSeps = ("\n", )
- self.lastdiscard = ''
+ # Max size supported by decompressor.
+ MAX_N = 1 << 31 - 1
- self.compress_type = zipinfo.compress_type
- self.compress_size = zipinfo.compress_size
+ # Read from compressed files in 4k blocks.
+ MIN_READ_SIZE = 4096
- self.closed = False
- self.mode = "r"
+ # Search for universal newlines or line chunks.
+ PATTERN = re.compile(r'^(?P<chunk>[^\r\n]+)|(?P<newline>\n|\r\n?)')
+
+ def __init__(self, fileobj, mode, zipinfo, decrypter=None):
+ self._fileobj = fileobj
+ self._decrypter = decrypter
+
+ self._compress_type = zipinfo.compress_type
+ self._compress_size = zipinfo.compress_size
+ self._compress_left = zipinfo.compress_size
+
+ if self._compress_type == ZIP_DEFLATED:
+ self._decompressor = zlib.decompressobj(-15)
+ self._unconsumed = ''
+
+ self._readbuffer = ''
+ self._offset = 0
+
+ self._universal = 'U' in mode
+ self.newlines = None
+
+ # Adjust read size for encrypted files since the first 12 bytes
+ # are for the encryption/password information.
+ if self._decrypter is not None:
+ self._compress_left -= 12
+
+ self.mode = mode
self.name = zipinfo.filename
- # read from compressed files in 64k blocks
- self.compreadsize = 64*1024
- if self.compress_type == ZIP_DEFLATED:
- self.dc = zlib.decompressobj(-15)
+ if hasattr(zipinfo, 'CRC'):
+ self._expected_crc = zipinfo.CRC
+ self._running_crc = crc32(b'') & 0xffffffff
+ else:
+ self._expected_crc = None
- def set_univ_newlines(self, univ_newlines):
- self.univ_newlines = univ_newlines
+ def readline(self, limit=-1):
+ """Read and return a line from the stream.
- # pick line separator char(s) based on universal newlines flag
- self.nlSeps = ("\n", )
- if self.univ_newlines:
- self.nlSeps = ("\r\n", "\r", "\n")
+ If limit is specified, at most limit bytes will be read.
+ """
- def __iter__(self):
- return self
+ if not self._universal and limit < 0:
+ # Shortcut common case - newline found in buffer.
+ i = self._readbuffer.find('\n', self._offset) + 1
+ if i > 0:
+ line = self._readbuffer[self._offset: i]
+ self._offset = i
+ return line
- def next(self):
- nextline = self.readline()
- if not nextline:
- raise StopIteration()
+ if not self._universal:
+ return io.BufferedIOBase.readline(self, limit)
- return nextline
+ line = ''
+ while limit < 0 or len(line) < limit:
+ readahead = self.peek(2)
+ if readahead == '':
+ return line
- def close(self):
- self.closed = True
+ #
+ # Search for universal newlines or line chunks.
+ #
+ # The pattern returns either a line chunk or a newline, but not
+ # both. Combined with peek(2), we are assured that the sequence
+ # '\r\n' is always retrieved completely and never split into
+ # separate newlines - '\r', '\n' due to coincidental readaheads.
+ #
+ match = self.PATTERN.search(readahead)
+ newline = match.group('newline')
+ if newline is not None:
+ if self.newlines is None:
+ self.newlines = []
+ if newline not in self.newlines:
+ self.newlines.append(newline)
+ self._offset += len(newline)
+ return line + '\n'
- def _checkfornewline(self):
- nl, nllen = -1, -1
- if self.linebuffer:
- # ugly check for cases where half of an \r\n pair was
- # read on the last pass, and the \r was discarded. In this
- # case we just throw away the \n at the start of the buffer.
- if (self.lastdiscard, self.linebuffer[0]) == ('\r','\n'):
- self.linebuffer = self.linebuffer[1:]
+ chunk = match.group('chunk')
+ if limit >= 0:
+ chunk = chunk[: limit - len(line)]
- for sep in self.nlSeps:
- nl = self.linebuffer.find(sep)
- if nl >= 0:
- nllen = len(sep)
- return nl, nllen
+ self._offset += len(chunk)
+ line += chunk
- return nl, nllen
+ return line
- def readline(self, size = -1):
- """Read a line with approx. size. If size is negative,
- read a whole line.
+ def peek(self, n=1):
+ """Returns buffered bytes without advancing the position."""
+ if n > len(self._readbuffer) - self._offset:
+ chunk = self.read(n)
+ self._offset -= len(chunk)
+
+ # Return up to 512 bytes to reduce allocation overhead for tight loops.
+ return self._readbuffer[self._offset: self._offset + 512]
+
+ def readable(self):
+ return True
+
+ def read(self, n=-1):
+ """Read and return up to n bytes.
+ If the argument is omitted, None, or negative, data is read and returned until EOF is reached..
"""
- if size < 0:
- size = sys.maxint
- elif size == 0:
- return ''
+ buf = ''
+ if n is None:
+ n = -1
+ while True:
+ if n < 0:
+ data = self.read1(n)
+ elif n > len(buf):
+ data = self.read1(n - len(buf))
+ else:
+ return buf
+ if len(data) == 0:
+ return buf
+ buf += data
- # check for a newline already in buffer
- nl, nllen = self._checkfornewline()
+ def _update_crc(self, newdata, eof):
+ # Update the CRC using the given data.
+ if self._expected_crc is None:
+ # No need to compute the CRC if we don't have a reference value
+ return
+ self._running_crc = crc32(newdata, self._running_crc) & 0xffffffff
+ # Check the CRC if we're at the end of the file
+ if eof and self._running_crc != self._expected_crc:
+ raise BadZipfile("Bad CRC-32 for file %r" % self.name)
- if nl >= 0:
- # the next line was already in the buffer
- nl = min(nl, size)
- else:
- # no line break in buffer - try to read more
- size -= len(self.linebuffer)
- while nl < 0 and size > 0:
- buf = self.read(min(size, 100))
- if not buf:
- break
- self.linebuffer += buf
- size -= len(buf)
+ def read1(self, n):
+ """Read up to n bytes with at most one read() system call."""
- # check for a newline in buffer
- nl, nllen = self._checkfornewline()
+ # Simplify algorithm (branching) by transforming negative n to large n.
+ if n < 0 or n is None:
+ n = self.MAX_N
- # we either ran out of bytes in the file, or
- # met the specified size limit without finding a newline,
- # so return current buffer
- if nl < 0:
- s = self.linebuffer
- self.linebuffer = ''
- return s
+ # Bytes available in read buffer.
+ len_readbuffer = len(self._readbuffer) - self._offset
- buf = self.linebuffer[:nl]
- self.lastdiscard = self.linebuffer[nl:nl + nllen]
- self.linebuffer = self.linebuffer[nl + nllen:]
+ # Read from file.
+ if self._compress_left > 0 and n > len_readbuffer + len(self._unconsumed):
+ nbytes = n - len_readbuffer - len(self._unconsumed)
+ nbytes = max(nbytes, self.MIN_READ_SIZE)
+ nbytes = min(nbytes, self._compress_left)
- # line is always returned with \n as newline char (except possibly
- # for a final incomplete line in the file, which is handled above).
- return buf + "\n"
+ data = self._fileobj.read(nbytes)
+ self._compress_left -= len(data)
- def readlines(self, sizehint = -1):
- """Return a list with all (following) lines. The sizehint parameter
- is ignored in this implementation.
- """
- result = []
- while True:
- line = self.readline()
- if not line: break
- result.append(line)
- return result
+ if data and self._decrypter is not None:
+ data = ''.join(map(self._decrypter, data))
- def read(self, size = None):
- # act like file() obj and return empty string if size is 0
- if size == 0:
- return ''
+ if self._compress_type == ZIP_STORED:
+ self._update_crc(data, eof=(self._compress_left==0))
+ self._readbuffer = self._readbuffer[self._offset:] + data
+ self._offset = 0
+ else:
+ # Prepare deflated bytes for decompression.
+ self._unconsumed += data
- # determine read size
- bytesToRead = self.compress_size - self.bytes_read
+ # Handle unconsumed data.
+ if (len(self._unconsumed) > 0 and n > len_readbuffer and
+ self._compress_type == ZIP_DEFLATED):
+ data = self._decompressor.decompress(
+ self._unconsumed,
+ max(n - len_readbuffer, self.MIN_READ_SIZE)
+ )
- # adjust read size for encrypted files since the first 12 bytes
- # are for the encryption/password information
- if self.decrypter is not None:
- bytesToRead -= 12
+ self._unconsumed = self._decompressor.unconsumed_tail
+ eof = len(self._unconsumed) == 0 and self._compress_left == 0
+ if eof:
+ data += self._decompressor.flush()
- if size is not None and size >= 0:
- if self.compress_type == ZIP_STORED:
- lr = len(self.readbuffer)
- bytesToRead = min(bytesToRead, size - lr)
- elif self.compress_type == ZIP_DEFLATED:
- if len(self.readbuffer) > size:
- # the user has requested fewer bytes than we've already
- # pulled through the decompressor; don't read any more
- bytesToRead = 0
- else:
- # user will use up the buffer, so read some more
- lr = len(self.rawbuffer)
- bytesToRead = min(bytesToRead, self.compreadsize - lr)
+ self._update_crc(data, eof=eof)
+ self._readbuffer = self._readbuffer[self._offset:] + data
+ self._offset = 0
- # avoid reading past end of file contents
- if bytesToRead + self.bytes_read > self.compress_size:
- bytesToRead = self.compress_size - self.bytes_read
+ # Read from buffer.
+ data = self._readbuffer[self._offset: self._offset + n]
+ self._offset += len(data)
+ return data
- # try to read from file (if necessary)
- if bytesToRead > 0:
- bytes = self.fileobj.read(bytesToRead)
- self.bytes_read += len(bytes)
- self.rawbuffer += bytes
-
- # handle contents of raw buffer
- if self.rawbuffer:
- newdata = self.rawbuffer
- self.rawbuffer = ''
-
- # decrypt new data if we were given an object to handle that
- if newdata and self.decrypter is not None:
- newdata = ''.join(map(self.decrypter, newdata))
-
- # decompress newly read data if necessary
- if newdata and self.compress_type == ZIP_DEFLATED:
- newdata = self.dc.decompress(newdata)
- self.rawbuffer = self.dc.unconsumed_tail
- if self.eof and len(self.rawbuffer) == 0:
- # we're out of raw bytes (both from the file and
- # the local buffer); flush just to make sure the
- # decompressor is done
- newdata += self.dc.flush()
- # prevent decompressor from being used again
- self.dc = None
-
- self.readbuffer += newdata
-
-
- # return what the user asked for
- if size is None or len(self.readbuffer) <= size:
- bytes = self.readbuffer
- self.readbuffer = ''
- else:
- bytes = self.readbuffer[:size]
- self.readbuffer = self.readbuffer[size:]
-
- return bytes
class ZipFile:
@@ -697,20 +713,34 @@
if key == 'r':
self._GetContents()
elif key == 'w':
- pass
+ # set the modified flag so central directory gets written
+ # even if no files are added to the archive
+ self._didModify = True
elif key == 'a':
- try: # See if file is a zip file
+ try:
+ # See if file is a zip file
self._RealGetContents()
# seek to start of directory and overwrite
self.fp.seek(self.start_dir, 0)
- except BadZipfile: # file is not a zip file, just append
+ except BadZipfile:
+ # file is not a zip file, just append
self.fp.seek(0, 2)
+
+ # set the modified flag so central directory gets written
+ # even if no files are added to the archive
+ self._didModify = True
else:
if not self._filePassed:
self.fp.close()
self.fp = None
raise RuntimeError, 'Mode must be "r", "w" or "a"'
+ def __enter__(self):
+ return self
+
+ def __exit__(self, type, value, traceback):
+ self.close()
+
def _GetContents(self):
"""Read the directory, making sure we close the file if the format
is bad."""
@@ -725,7 +755,10 @@
def _RealGetContents(self):
"""Read in the table of contents for the ZIP file."""
fp = self.fp
- endrec = _EndRecData(fp)
+ try:
+ endrec = _EndRecData(fp)
+ except IOError:
+ raise BadZipfile("File is not a zip file")
if not endrec:
raise BadZipfile, "File is not a zip file"
if self.debug > 1:
@@ -902,16 +935,7 @@
if ord(h[11]) != check_byte:
raise RuntimeError("Bad password for file", name)
- # build and return a ZipExtFile
- if zd is None:
- zef = ZipExtFile(zef_file, zinfo)
- else:
- zef = ZipExtFile(zef_file, zinfo, zd)
-
- # set universal newlines on ZipExtFile if necessary
- if "U" in mode:
- zef.set_univ_newlines(True)
- return zef
+ return ZipExtFile(zef_file, mode, zinfo, zd)
def extract(self, member, path=None, pwd=None):
"""Extract a member from the archive to the current working directory,
@@ -1041,28 +1065,27 @@
self.fp.write(zinfo.FileHeader())
return
- fp = open(filename, "rb")
- # Must overwrite CRC and sizes with correct data later
- zinfo.CRC = CRC = 0
- zinfo.compress_size = compress_size = 0
- zinfo.file_size = file_size = 0
- self.fp.write(zinfo.FileHeader())
- if zinfo.compress_type == ZIP_DEFLATED:
- cmpr = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION,
- zlib.DEFLATED, -15)
- else:
- cmpr = None
- while 1:
- buf = fp.read(1024 * 8)
- if not buf:
- break
- file_size = file_size + len(buf)
- CRC = crc32(buf, CRC) & 0xffffffff
- if cmpr:
- buf = cmpr.compress(buf)
- compress_size = compress_size + len(buf)
- self.fp.write(buf)
- fp.close()
+ with open(filename, "rb") as fp:
+ # Must overwrite CRC and sizes with correct data later
+ zinfo.CRC = CRC = 0
+ zinfo.compress_size = compress_size = 0
+ zinfo.file_size = file_size = 0
+ self.fp.write(zinfo.FileHeader())
+ if zinfo.compress_type == ZIP_DEFLATED:
+ cmpr = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION,
+ zlib.DEFLATED, -15)
+ else:
+ cmpr = None
+ while 1:
+ buf = fp.read(1024 * 8)
+ if not buf:
+ break
+ file_size = file_size + len(buf)
+ CRC = crc32(buf, CRC) & 0xffffffff
+ if cmpr:
+ buf = cmpr.compress(buf)
+ compress_size = compress_size + len(buf)
+ self.fp.write(buf)
if cmpr:
buf = cmpr.flush()
compress_size = compress_size + len(buf)
@@ -1081,13 +1104,14 @@
self.filelist.append(zinfo)
self.NameToInfo[zinfo.filename] = zinfo
- def writestr(self, zinfo_or_arcname, bytes):
+ def writestr(self, zinfo_or_arcname, bytes, compress_type=None):
"""Write a file into the archive. The contents is the string
'bytes'. 'zinfo_or_arcname' is either a ZipInfo instance or
the name of the file in the archive."""
if not isinstance(zinfo_or_arcname, ZipInfo):
zinfo = ZipInfo(filename=zinfo_or_arcname,
date_time=time.localtime(time.time())[:6])
+
zinfo.compress_type = self.compression
zinfo.external_attr = 0600 << 16
else:
@@ -1097,6 +1121,9 @@
raise RuntimeError(
"Attempt to write to ZIP archive that was already closed")
+ if compress_type is not None:
+ zinfo.compress_type = compress_type
+
zinfo.file_size = len(bytes) # Uncompressed size
zinfo.header_offset = self.fp.tell() # Start of header bytes
self._writecheck(zinfo)
@@ -1363,7 +1390,9 @@
print USAGE
sys.exit(1)
zf = ZipFile(args[1], 'r')
- zf.testzip()
+ badfile = zf.testzip()
+ if badfile:
+ print("The following enclosed file is corrupted: {!r}".format(badfile))
print "Done testing"
elif args[0] == '-e':
@@ -1382,9 +1411,8 @@
tgtdir = os.path.dirname(tgt)
if not os.path.exists(tgtdir):
os.makedirs(tgtdir)
- fp = open(tgt, 'wb')
- fp.write(zf.read(path))
- fp.close()
+ with open(tgt, 'wb') as fp:
+ fp.write(zf.read(path))
zf.close()
elif args[0] == '-c':
--
Repository URL: http://hg.python.org/jython
More information about the Jython-checkins
mailing list