diff options
Diffstat (limited to 'Lib/test')
71 files changed, 2725 insertions, 456 deletions
diff --git a/Lib/test/crashers/README b/Lib/test/crashers/README index 2a73e1b..0259a06 100644 --- a/Lib/test/crashers/README +++ b/Lib/test/crashers/README @@ -14,3 +14,7 @@ note if the cause is system or environment dependent and what the variables are. Once the crash is fixed, the test case should be moved into an appropriate test (even if it was originally from the test suite). This ensures the regression doesn't happen again. And if it does, it should be easier to track down. + +Also see Lib/test_crashers.py which exercises the crashers in this directory. +In particular, make sure to add any new infinite loop crashers to the black +list so it doesn't try to run them. diff --git a/Lib/test/crashers/compiler_recursion.py b/Lib/test/crashers/compiler_recursion.py index 4954bdd..31f28a9 100644 --- a/Lib/test/crashers/compiler_recursion.py +++ b/Lib/test/crashers/compiler_recursion.py @@ -1,5 +1,13 @@ """ -The compiler (>= 2.5) recurses happily. +The compiler (>= 2.5) recurses happily until it blows the stack. + +Recorded on the tracker as http://bugs.python.org/issue11383 """ -compile('()'*9**5, '?', 'exec') +# The variant below blows up in compiler_call, but there are assorted +# other variations that blow up in other functions +# e.g. '1*'*10**5+'1' will die in compiler_visit_expr + +# The exact limit to destroy the stack will vary by platform +# but 10M should do the trick even with huge stack allocations +compile('()'*10**7, '?', 'exec') diff --git a/Lib/test/list_tests.py b/Lib/test/list_tests.py index e3a7845..0c656fd 100644 --- a/Lib/test/list_tests.py +++ b/Lib/test/list_tests.py @@ -425,6 +425,47 @@ class CommonTest(seq_tests.CommonTest): self.assertRaises(TypeError, u.reverse, 42) + def test_clear(self): + u = self.type2test([2, 3, 4]) + u.clear() + self.assertEqual(u, []) + + u = self.type2test([]) + u.clear() + self.assertEqual(u, []) + + u = self.type2test([]) + u.append(1) + u.clear() + u.append(2) + self.assertEqual(u, [2]) + + self.assertRaises(TypeError, u.clear, None) + + def test_copy(self): + u = self.type2test([1, 2, 3]) + v = u.copy() + self.assertEqual(v, [1, 2, 3]) + + u = self.type2test([]) + v = u.copy() + self.assertEqual(v, []) + + # test that it's indeed a copy and not a reference + u = self.type2test(['a', 'b']) + v = u.copy() + v.append('i') + self.assertEqual(u, ['a', 'b']) + self.assertEqual(v, u + ['i']) + + # test that it's a shallow, not a deep copy + u = self.type2test([1, 2, [3, 4], 5]) + v = u.copy() + self.assertEqual(u, v) + self.assertIs(v[3], u[3]) + + self.assertRaises(TypeError, u.copy, None) + def test_sort(self): u = self.type2test([1, 0]) u.sort() diff --git a/Lib/test/pickletester.py b/Lib/test/pickletester.py index a843486..45e8f74 100644 --- a/Lib/test/pickletester.py +++ b/Lib/test/pickletester.py @@ -3,9 +3,10 @@ import unittest import pickle import pickletools import copyreg +import weakref from http.cookies import SimpleCookie -from test.support import TestFailed, TESTFN, run_with_locale +from test.support import TestFailed, TESTFN, run_with_locale, no_tracing from pickle import bytes_types @@ -842,6 +843,25 @@ class AbstractPickleTests(unittest.TestCase): self.assertEqual(B(x), B(y), detail) self.assertEqual(x.__dict__, y.__dict__, detail) + def test_newobj_proxies(self): + # NEWOBJ should use the __class__ rather than the raw type + classes = myclasses[:] + # Cannot create weakproxies to these classes + for c in (MyInt, MyTuple): + classes.remove(c) + for proto in protocols: + for C in classes: + B = C.__base__ + x = C(C.sample) + x.foo = 42 + p = weakref.proxy(x) + s = self.dumps(p, proto) + y = self.loads(s) + self.assertEqual(type(y), type(x)) # rather than type(p) + detail = (proto, C, B, x, y, type(y)) + self.assertEqual(B(x), B(y), detail) + self.assertEqual(x.__dict__, y.__dict__, detail) + # Register a type with copyreg, with extension code extcode. Pickle # an object of that type. Check that the resulting pickle uses opcode # (EXT[124]) under proto 2, and not in proto 1. @@ -1002,13 +1022,13 @@ class AbstractPickleTests(unittest.TestCase): y = self.loads(s) self.assertEqual(y._reduce_called, 1) + @no_tracing def test_bad_getattr(self): x = BadGetattr() for proto in 0, 1: self.assertRaises(RuntimeError, self.dumps, x, proto) # protocol 2 don't raise a RuntimeError. d = self.dumps(x, 2) - self.assertRaises(RuntimeError, self.loads, d) def test_reduce_bad_iterator(self): # Issue4176: crash when 4th and 5th items of __reduce__() diff --git a/Lib/test/regrtest.py b/Lib/test/regrtest.py index b5288b5..6267702 100755 --- a/Lib/test/regrtest.py +++ b/Lib/test/regrtest.py @@ -827,7 +827,7 @@ class saved_test_environment: resources = ('sys.argv', 'cwd', 'sys.stdin', 'sys.stdout', 'sys.stderr', 'os.environ', 'sys.path', 'sys.path_hooks', '__import__', 'warnings.filters', 'asyncore.socket_map', - 'logging._handlers', 'logging._handlerList') + 'logging._handlers', 'logging._handlerList', 'sys.gettrace') def get_sys_argv(self): return id(sys.argv), sys.argv, sys.argv[:] @@ -874,6 +874,11 @@ class saved_test_environment: sys.path_hooks = saved_hooks[1] sys.path_hooks[:] = saved_hooks[2] + def get_sys_gettrace(self): + return sys.gettrace() + def restore_sys_gettrace(self, trace_fxn): + sys.settrace(trace_fxn) + def get___import__(self): return builtins.__import__ def restore___import__(self, import_): @@ -1051,7 +1056,8 @@ def dash_R(the_module, test, indirect_test, huntrleaks): False if the test didn't leak references; True if we detected refleaks. """ # This code is hackish and inelegant, but it seems to do the job. - import copyreg, _abcoll + import copyreg + import collections.abc if not hasattr(sys, 'gettotalrefcount'): raise Exception("Tracking reference leaks requires a debug build " @@ -1068,7 +1074,7 @@ def dash_R(the_module, test, indirect_test, huntrleaks): else: zdc = zipimport._zip_directory_cache.copy() abcs = {} - for abc in [getattr(_abcoll, a) for a in _abcoll.__all__]: + for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]: if not isabstract(abc): continue for obj in abc.__subclasses__() + [abc]: @@ -1114,7 +1120,7 @@ def dash_R_cleanup(fs, ps, pic, zdc, abcs): import gc, copyreg import _strptime, linecache import urllib.parse, urllib.request, mimetypes, doctest - import struct, filecmp, _abcoll + import struct, filecmp, collections.abc from distutils.dir_util import _path_created from weakref import WeakSet @@ -1141,7 +1147,7 @@ def dash_R_cleanup(fs, ps, pic, zdc, abcs): sys._clear_type_cache() # Clear ABC registries, restoring previously saved ABC registries. - for abc in [getattr(_abcoll, a) for a in _abcoll.__all__]: + for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]: if not isabstract(abc): continue for obj in abc.__subclasses__() + [abc]: diff --git a/Lib/test/support.py b/Lib/test/support.py index 8d8e187..66d7f93 100644 --- a/Lib/test/support.py +++ b/Lib/test/support.py @@ -15,7 +15,7 @@ import shutil import warnings import unittest import importlib -import collections +import collections.abc import re import subprocess import imp @@ -33,7 +33,7 @@ __all__ = [ "verbose", "use_resources", "max_memuse", "record_original_stdout", "get_original_stdout", "unload", "unlink", "rmtree", "forget", "is_resource_enabled", "requires", "find_unused_port", "bind_port", - "fcmp", "is_jython", "TESTFN", "HOST", "FUZZ", "SAVEDCWD", "temp_cwd", + "is_jython", "TESTFN", "HOST", "SAVEDCWD", "temp_cwd", "findfile", "sortdict", "check_syntax_error", "open_urlresource", "check_warnings", "CleanImport", "EnvironmentVarGuard", "TransientResource", "captured_output", "captured_stdout", @@ -381,24 +381,6 @@ def bind_port(sock, host=HOST): port = sock.getsockname()[1] return port -FUZZ = 1e-6 - -def fcmp(x, y): # fuzzy comparison function - if isinstance(x, float) or isinstance(y, float): - try: - fuzz = (abs(x) + abs(y)) * FUZZ - if abs(x-y) <= fuzz: - return 0 - except: - pass - elif type(x) == type(y) and isinstance(x, (tuple, list)): - for i in range(min(len(x), len(y))): - outcome = fcmp(x[i], y[i]) - if outcome != 0: - return outcome - return (len(x) > len(y)) - (len(x) < len(y)) - return (x > y) - (x < y) - # decorator for skipping tests on non-IEEE 754 platforms requires_IEEE_754 = unittest.skipUnless( float.__getformat__("double").startswith("IEEE"), @@ -714,7 +696,7 @@ class CleanImport(object): sys.modules.update(self.original_modules) -class EnvironmentVarGuard(collections.MutableMapping): +class EnvironmentVarGuard(collections.abc.MutableMapping): """Class to help protect the environment variable properly. Can be used as a context manager.""" @@ -1142,6 +1124,32 @@ def check_impl_detail(**guards): return guards.get(platform.python_implementation().lower(), default) +def no_tracing(func): + """Decorator to temporarily turn off tracing for the duration of a test.""" + if not hasattr(sys, 'gettrace'): + return func + else: + @functools.wraps(func) + def wrapper(*args, **kwargs): + original_trace = sys.gettrace() + try: + sys.settrace(None) + return func(*args, **kwargs) + finally: + sys.settrace(original_trace) + return wrapper + + +def refcount_test(test): + """Decorator for tests which involve reference counting. + + To start, the decorator does not run the test if is not run by CPython. + After that, any trace function is unset during the test to prevent + unexpected refcounts caused by the trace function. + + """ + return no_tracing(cpython_only(test)) + def _run_suite(suite): """Run tests from a unittest.TestSuite-derived class.""" @@ -1366,7 +1374,7 @@ def strip_python_stderr(stderr): def args_from_interpreter_flags(): """Return a list of command-line arguments reproducing the current - settings in sys.flags.""" + settings in sys.flags and sys.warnoptions.""" flag_opt_map = { 'bytes_warning': 'b', 'dont_write_bytecode': 'B', @@ -1381,6 +1389,9 @@ def args_from_interpreter_flags(): v = getattr(sys.flags, flag) if v > 0: args.append('-' + opt * v) + if sys.warnoptions: + args.append('-W') + args.extend(sys.warnoptions) return args #============================================================ @@ -1454,9 +1465,11 @@ def can_symlink(): global _can_symlink if _can_symlink is not None: return _can_symlink + symlink_path = TESTFN + "can_symlink" try: - os.symlink(TESTFN, TESTFN + "can_symlink") + os.symlink(TESTFN, symlink_path) can = True + os.remove(symlink_path) except (OSError, NotImplementedError): can = False _can_symlink = can diff --git a/Lib/test/test_abc.py b/Lib/test/test_abc.py index d86f97c..1319a64 100644 --- a/Lib/test/test_abc.py +++ b/Lib/test/test_abc.py @@ -121,11 +121,32 @@ class TestABC(unittest.TestCase): self.assertFalse(issubclass(B, (A,))) self.assertNotIsInstance(b, A) self.assertNotIsInstance(b, (A,)) - A.register(B) + B1 = A.register(B) + self.assertTrue(issubclass(B, A)) + self.assertTrue(issubclass(B, (A,))) + self.assertIsInstance(b, A) + self.assertIsInstance(b, (A,)) + self.assertIs(B1, B) + class C(B): + pass + c = C() + self.assertTrue(issubclass(C, A)) + self.assertTrue(issubclass(C, (A,))) + self.assertIsInstance(c, A) + self.assertIsInstance(c, (A,)) + + def test_register_as_class_deco(self): + class A(metaclass=abc.ABCMeta): + pass + @A.register + class B(object): + pass + b = B() self.assertTrue(issubclass(B, A)) self.assertTrue(issubclass(B, (A,))) self.assertIsInstance(b, A) self.assertIsInstance(b, (A,)) + @A.register class C(B): pass c = C() @@ -133,6 +154,7 @@ class TestABC(unittest.TestCase): self.assertTrue(issubclass(C, (A,))) self.assertIsInstance(c, A) self.assertIsInstance(c, (A,)) + self.assertIs(C, A.register(C)) def test_isinstance_invalidation(self): class A(metaclass=abc.ABCMeta): diff --git a/Lib/test/test_argparse.py b/Lib/test/test_argparse.py index 03c95fa..8d80336 100644 --- a/Lib/test/test_argparse.py +++ b/Lib/test/test_argparse.py @@ -4328,7 +4328,7 @@ class TestEncoding(TestCase): def _test_module_encoding(self, path): path, _ = os.path.splitext(path) path += ".py" - with codecs.open(path, 'r', 'utf8') as f: + with codecs.open(path, 'r', 'utf-8') as f: f.read() def test_argparse_module_encoding(self): diff --git a/Lib/test/test_asyncore.py b/Lib/test/test_asyncore.py index 53c49a8..389dc8a 100644 --- a/Lib/test/test_asyncore.py +++ b/Lib/test/test_asyncore.py @@ -352,7 +352,7 @@ class DispatcherWithSendTests(unittest.TestCase): @support.reap_threads def test_send(self): evt = threading.Event() - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket() sock.settimeout(3) port = support.bind_port(sock) @@ -367,7 +367,7 @@ class DispatcherWithSendTests(unittest.TestCase): data = b"Suppose there isn't a 16-ton weight?" d = dispatcherwithsend_noread() - d.create_socket(socket.AF_INET, socket.SOCK_STREAM) + d.create_socket() d.connect((HOST, port)) # give time for socket to connect @@ -474,7 +474,7 @@ class TCPServer(asyncore.dispatcher): def __init__(self, handler=BaseTestHandler, host=HOST, port=0): asyncore.dispatcher.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.create_socket() self.set_reuse_addr() self.bind((host, port)) self.listen(5) @@ -495,7 +495,7 @@ class BaseClient(BaseTestHandler): def __init__(self, address): BaseTestHandler.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.create_socket() self.connect(address) def handle_connect(self): @@ -536,7 +536,7 @@ class BaseTestAPI(unittest.TestCase): def __init__(self): BaseTestHandler.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.create_socket() self.bind((HOST, 0)) self.listen(5) self.address = self.socket.getsockname()[:2] @@ -555,7 +555,7 @@ class BaseTestAPI(unittest.TestCase): def __init__(self): BaseTestHandler.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.create_socket() self.bind((HOST, 0)) self.listen(5) self.address = self.socket.getsockname()[:2] @@ -693,20 +693,20 @@ class BaseTestAPI(unittest.TestCase): def test_create_socket(self): s = asyncore.dispatcher() - s.create_socket(socket.AF_INET, socket.SOCK_STREAM) + s.create_socket() self.assertEqual(s.socket.family, socket.AF_INET) SOCK_NONBLOCK = getattr(socket, 'SOCK_NONBLOCK', 0) self.assertEqual(s.socket.type, socket.SOCK_STREAM | SOCK_NONBLOCK) def test_bind(self): s1 = asyncore.dispatcher() - s1.create_socket(socket.AF_INET, socket.SOCK_STREAM) + s1.create_socket() s1.bind((HOST, 0)) s1.listen(5) port = s1.socket.getsockname()[1] s2 = asyncore.dispatcher() - s2.create_socket(socket.AF_INET, socket.SOCK_STREAM) + s2.create_socket() # EADDRINUSE indicates the socket was correctly bound self.assertRaises(socket.error, s2.bind, (HOST, port)) @@ -723,7 +723,7 @@ class BaseTestAPI(unittest.TestCase): self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)) s.socket.close() - s.create_socket(socket.AF_INET, socket.SOCK_STREAM) + s.create_socket() s.set_reuse_addr() self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)) diff --git a/Lib/test/test_bigmem.py b/Lib/test/test_bigmem.py index ac6b109..f9a0a3d 100644 --- a/Lib/test/test_bigmem.py +++ b/Lib/test/test_bigmem.py @@ -707,7 +707,7 @@ class StrTest(unittest.TestCase, BaseStrTest): class BytesTest(unittest.TestCase, BaseStrTest): def from_latin1(self, s): - return s.encode("latin1") + return s.encode("latin-1") @bigmemtest(minsize=_2G + 2, memuse=1 + character_size) def test_decode(self, size): @@ -718,7 +718,7 @@ class BytesTest(unittest.TestCase, BaseStrTest): class BytearrayTest(unittest.TestCase, BaseStrTest): def from_latin1(self, s): - return bytearray(s.encode("latin1")) + return bytearray(s.encode("latin-1")) @bigmemtest(minsize=_2G + 2, memuse=1 + character_size) def test_decode(self, size): diff --git a/Lib/test/test_builtin.py b/Lib/test/test_builtin.py index 1469e36..b094a65 100644 --- a/Lib/test/test_builtin.py +++ b/Lib/test/test_builtin.py @@ -10,7 +10,7 @@ import ast import types import builtins import random -from test.support import fcmp, TESTFN, unlink, run_unittest, check_warnings +from test.support import TESTFN, unlink, run_unittest, check_warnings from operator import neg @@ -394,10 +394,13 @@ class BuiltinTest(unittest.TestCase): self.assertEqual(divmod(-sys.maxsize-1, -1), (sys.maxsize+1, 0)) - self.assertTrue(not fcmp(divmod(3.25, 1.0), (3.0, 0.25))) - self.assertTrue(not fcmp(divmod(-3.25, 1.0), (-4.0, 0.75))) - self.assertTrue(not fcmp(divmod(3.25, -1.0), (-4.0, -0.75))) - self.assertTrue(not fcmp(divmod(-3.25, -1.0), (3.0, -0.25))) + for num, denom, exp_result in [ (3.25, 1.0, (3.0, 0.25)), + (-3.25, 1.0, (-4.0, 0.75)), + (3.25, -1.0, (-4.0, -0.75)), + (-3.25, -1.0, (3.0, -0.25))]: + result = divmod(num, denom) + self.assertAlmostEqual(result[0], exp_result[0]) + self.assertAlmostEqual(result[1], exp_result[1]) self.assertRaises(TypeError, divmod) @@ -1276,14 +1279,14 @@ class BuiltinTest(unittest.TestCase): # -------------------------------------------------------------------- # Issue #7994: object.__format__ with a non-empty format string is - # pending deprecated + # deprecated def test_deprecated_format_string(obj, fmt_str, should_raise_warning): with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always", PendingDeprecationWarning) + warnings.simplefilter("always", DeprecationWarning) format(obj, fmt_str) if should_raise_warning: self.assertEqual(len(w), 1) - self.assertIsInstance(w[0].message, PendingDeprecationWarning) + self.assertIsInstance(w[0].message, DeprecationWarning) self.assertIn('object.__format__ with a non-empty format ' 'string', str(w[0].message)) else: diff --git a/Lib/test/test_bytes.py b/Lib/test/test_bytes.py index 9be1008..0b70c3a 100644 --- a/Lib/test/test_bytes.py +++ b/Lib/test/test_bytes.py @@ -188,24 +188,26 @@ class BaseBytesTest(unittest.TestCase): def test_encoding(self): sample = "Hello world\n\u1234\u5678\u9abc" - for enc in ("utf8", "utf16"): + for enc in ("utf-8", "utf-16"): b = self.type2test(sample, enc) self.assertEqual(b, self.type2test(sample.encode(enc))) - self.assertRaises(UnicodeEncodeError, self.type2test, sample, "latin1") - b = self.type2test(sample, "latin1", "ignore") + self.assertRaises(UnicodeEncodeError, self.type2test, sample, "latin-1") + b = self.type2test(sample, "latin-1", "ignore") self.assertEqual(b, self.type2test(sample[:-3], "utf-8")) def test_decode(self): sample = "Hello world\n\u1234\u5678\u9abc\def0\def0" - for enc in ("utf8", "utf16"): + for enc in ("utf-8", "utf-16"): b = self.type2test(sample, enc) self.assertEqual(b.decode(enc), sample) sample = "Hello world\n\x80\x81\xfe\xff" - b = self.type2test(sample, "latin1") - self.assertRaises(UnicodeDecodeError, b.decode, "utf8") - self.assertEqual(b.decode("utf8", "ignore"), "Hello world\n") - self.assertEqual(b.decode(errors="ignore", encoding="utf8"), + b = self.type2test(sample, "latin-1") + self.assertRaises(UnicodeDecodeError, b.decode, "utf-8") + self.assertEqual(b.decode("utf-8", "ignore"), "Hello world\n") + self.assertEqual(b.decode(errors="ignore", encoding="utf-8"), "Hello world\n") + # Default encoding is utf-8 + self.assertEqual(self.type2test(b'\xe2\x98\x83').decode(), '\u2603') def test_from_int(self): b = self.type2test(0) @@ -562,6 +564,39 @@ class ByteArrayTest(BaseBytesTest): b.reverse() self.assertFalse(b) + def test_clear(self): + b = bytearray(b'python') + b.clear() + self.assertEqual(b, b'') + + b = bytearray(b'') + b.clear() + self.assertEqual(b, b'') + + b = bytearray(b'') + b.append(ord('r')) + b.clear() + b.append(ord('p')) + self.assertEqual(b, b'p') + + def test_copy(self): + b = bytearray(b'abc') + bb = b.copy() + self.assertEqual(bb, b'abc') + + b = bytearray(b'') + bb = b.copy() + self.assertEqual(bb, b'') + + # test that it's indeed a copy and not a reference + b = bytearray(b'abc') + bb = b.copy() + self.assertEqual(b, bb) + self.assertIsNot(b, bb) + bb.append(ord('d')) + self.assertEqual(bb, b'abcd') + self.assertEqual(b, b'abc') + def test_regexps(self): def by(s): return bytearray(map(ord, s)) diff --git a/Lib/test/test_cmd_line.py b/Lib/test/test_cmd_line.py index c4e3adf..a0a85ae 100644 --- a/Lib/test/test_cmd_line.py +++ b/Lib/test/test_cmd_line.py @@ -151,7 +151,7 @@ class CmdLineTest(unittest.TestCase): @unittest.skipUnless(sys.platform == 'darwin', 'test specific to Mac OS X') def test_osx_utf8(self): def check_output(text): - decoded = text.decode('utf8', 'surrogateescape') + decoded = text.decode('utf-8', 'surrogateescape') expected = ascii(decoded).encode('ascii') + b'\n' env = os.environ.copy() @@ -223,7 +223,7 @@ class CmdLineTest(unittest.TestCase): self.assertIn(path2.encode('ascii'), out) def test_displayhook_unencodable(self): - for encoding in ('ascii', 'latin1', 'utf8'): + for encoding in ('ascii', 'latin-1', 'utf-8'): env = os.environ.copy() env['PYTHONIOENCODING'] = encoding p = subprocess.Popen( diff --git a/Lib/test/test_codecs.py b/Lib/test/test_codecs.py index d560d7a..67a5aed 100644 --- a/Lib/test/test_codecs.py +++ b/Lib/test/test_codecs.py @@ -1250,7 +1250,7 @@ class EncodedFileTest(unittest.TestCase): self.assertEqual(ef.read(), b'\\\xd5\n\x00\x00\xae') f = io.BytesIO() - ef = codecs.EncodedFile(f, 'utf-8', 'latin1') + ef = codecs.EncodedFile(f, 'utf-8', 'latin-1') ef.write(b'\xc3\xbc') self.assertEqual(f.getvalue(), b'\xfc') @@ -1611,7 +1611,7 @@ class SurrogateEscapeTest(unittest.TestCase): def test_latin1(self): # Issue6373 - self.assertEqual("\udce4\udceb\udcef\udcf6\udcfc".encode("latin1", "surrogateescape"), + self.assertEqual("\udce4\udceb\udcef\udcf6\udcfc".encode("latin-1", "surrogateescape"), b"\xe4\xeb\xef\xf6\xfc") diff --git a/Lib/test/test_collections.py b/Lib/test/test_collections.py index b3e0907..d71fb01 100644 --- a/Lib/test/test_collections.py +++ b/Lib/test/test_collections.py @@ -10,17 +10,18 @@ from random import randrange, shuffle import keyword import re import sys -from collections import _ChainMap as ChainMap -from collections import Hashable, Iterable, Iterator -from collections import Sized, Container, Callable -from collections import Set, MutableSet -from collections import Mapping, MutableMapping, KeysView, ItemsView, UserDict -from collections import Sequence, MutableSequence -from collections import ByteString +from collections import UserDict +from collections import ChainMap +from collections.abc import Hashable, Iterable, Iterator +from collections.abc import Sized, Container, Callable +from collections.abc import Set, MutableSet +from collections.abc import Mapping, MutableMapping, KeysView, ItemsView +from collections.abc import Sequence, MutableSequence +from collections.abc import ByteString ################################################################################ -### _ChainMap (helper class for configparser) +### ChainMap (helper class for configparser and the string module) ################################################################################ class TestChainMap(unittest.TestCase): @@ -71,17 +72,23 @@ class TestChainMap(unittest.TestCase): for m1, m2 in zip(d.maps, e.maps): self.assertIsNot(m1, m2, e) - d.new_child() - d['b'] = 5 - self.assertEqual(d.maps, [{'b': 5}, {'c':30}, {'a':1, 'b':2}]) - self.assertEqual(d.parents.maps, [{'c':30}, {'a':1, 'b':2}]) # check parents - self.assertEqual(d['b'], 5) # find first in chain - self.assertEqual(d.parents['b'], 2) # look beyond maps[0] + f = d.new_child() + f['b'] = 5 + self.assertEqual(f.maps, [{'b': 5}, {'c':30}, {'a':1, 'b':2}]) + self.assertEqual(f.parents.maps, [{'c':30}, {'a':1, 'b':2}]) # check parents + self.assertEqual(f['b'], 5) # find first in chain + self.assertEqual(f.parents['b'], 2) # look beyond maps[0] def test_contructor(self): - self.assertEqual(ChainedContext().maps, [{}]) # no-args --> one new dict + self.assertEqual(ChainMap().maps, [{}]) # no-args --> one new dict self.assertEqual(ChainMap({1:2}).maps, [{1:2}]) # 1 arg --> list + def test_bool(self): + self.assertFalse(ChainMap()) + self.assertFalse(ChainMap({}, {})) + self.assertTrue(ChainMap({1:2}, {})) + self.assertTrue(ChainMap({}, {1:2})) + def test_missing(self): class DefaultChainMap(ChainMap): def __missing__(self, key): @@ -721,6 +728,44 @@ class TestCollectionABCs(ABCTestCase): self.validate_abstract_methods(MutableSequence, '__contains__', '__iter__', '__len__', '__getitem__', '__setitem__', '__delitem__', 'insert') + def test_MutableSequence_mixins(self): + # Test the mixins of MutableSequence by creating a miminal concrete + # class inherited from it. + class MutableSequenceSubclass(MutableSequence): + def __init__(self): + self.lst = [] + + def __setitem__(self, index, value): + self.lst[index] = value + + def __getitem__(self, index): + return self.lst[index] + + def __len__(self): + return len(self.lst) + + def __delitem__(self, index): + del self.lst[index] + + def insert(self, index, value): + self.lst.insert(index, value) + + mss = MutableSequenceSubclass() + mss.append(0) + mss.extend((1, 2, 3, 4)) + self.assertEqual(len(mss), 5) + self.assertEqual(mss[3], 3) + mss.reverse() + self.assertEqual(mss[3], 1) + mss.pop() + self.assertEqual(len(mss), 4) + mss.remove(3) + self.assertEqual(len(mss), 3) + mss += (10, 20, 30) + self.assertEqual(len(mss), 6) + self.assertEqual(mss[-1], 30) + mss.clear() + self.assertEqual(len(mss), 0) ################################################################################ ### Counter @@ -1181,7 +1226,7 @@ import doctest, collections def test_main(verbose=None): NamedTupleDocs = doctest.DocTestSuite(module=collections) test_classes = [TestNamedTuple, NamedTupleDocs, TestOneTrickPonyABCs, - TestCollectionABCs, TestCounter, + TestCollectionABCs, TestCounter, TestChainMap, TestOrderedDict, GeneralMappingTests, SubclassMappingTests] support.run_unittest(*test_classes) support.run_doctest(collections, verbose) diff --git a/Lib/test/test_compileall.py b/Lib/test/test_compileall.py index 250d31b..a63af4c 100644 --- a/Lib/test/test_compileall.py +++ b/Lib/test/test_compileall.py @@ -345,7 +345,7 @@ class CommandLineTests(unittest.TestCase): def test_invalid_arg_produces_message(self): out = self.assertRunOK('badfilename') - self.assertRegex(out, b"Can't list badfilename") + self.assertRegex(out, b"Can't list 'badfilename'") def test_main(): diff --git a/Lib/test/test_cfgparser.py b/Lib/test/test_configparser.py index f7d9a26..f7d9a26 100644 --- a/Lib/test/test_cfgparser.py +++ b/Lib/test/test_configparser.py diff --git a/Lib/test/test_crashers.py b/Lib/test/test_crashers.py new file mode 100644 index 0000000..86f88f7 --- /dev/null +++ b/Lib/test/test_crashers.py @@ -0,0 +1,37 @@ +# Tests that the crashers in the Lib/test/crashers directory actually +# do crash the interpreter as expected +# +# If a crasher is fixed, it should be moved elsewhere in the test suite to +# ensure it continues to work correctly. + +import unittest +import glob +import os.path +import test.support +from test.script_helper import assert_python_failure + +CRASHER_DIR = os.path.join(os.path.dirname(__file__), "crashers") +CRASHER_FILES = os.path.join(CRASHER_DIR, "*.py") + +infinite_loops = ["infinite_loop_re.py", "nasty_eq_vs_dict.py"] + +class CrasherTest(unittest.TestCase): + + @test.support.cpython_only + def test_crashers_crash(self): + for fname in glob.glob(CRASHER_FILES): + if os.path.basename(fname) in infinite_loops: + continue + # Some "crashers" only trigger an exception rather than a + # segfault. Consider that an acceptable outcome. + if test.support.verbose: + print("Checking crasher:", fname) + assert_python_failure(fname) + + +def test_main(): + test.support.run_unittest(CrasherTest) + test.support.reap_children() + +if __name__ == "__main__": + test_main() diff --git a/Lib/test/test_crypt.py b/Lib/test/test_crypt.py index 2adb28d..dc107d8 100644 --- a/Lib/test/test_crypt.py +++ b/Lib/test/test_crypt.py @@ -10,6 +10,25 @@ class CryptTestCase(unittest.TestCase): if support.verbose: print('Test encryption: ', c) + def test_salt(self): + self.assertEqual(len(crypt._saltchars), 64) + for method in crypt.methods: + salt = crypt.mksalt(method) + self.assertEqual(len(salt), + method.salt_chars + (3 if method.ident else 0)) + + def test_saltedcrypt(self): + for method in crypt.methods: + pw = crypt.crypt('assword', method) + self.assertEqual(len(pw), method.total_size) + pw = crypt.crypt('assword', crypt.mksalt(method)) + self.assertEqual(len(pw), method.total_size) + + def test_methods(self): + # Gurantee that METHOD_CRYPT is the last method in crypt.methods. + self.assertTrue(len(crypt.methods) >= 1) + self.assertEqual(crypt.METHOD_CRYPT, crypt.methods[-1]) + def test_main(): support.run_unittest(CryptTestCase) diff --git a/Lib/test/test_descr.py b/Lib/test/test_descr.py index 41dfa70..b0d531a 100644 --- a/Lib/test/test_descr.py +++ b/Lib/test/test_descr.py @@ -4243,6 +4243,8 @@ class DictProxyTests(unittest.TestCase): pass self.C = C + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __local__') def test_iter_keys(self): # Testing dict-proxy keys... it = self.C.__dict__.keys() @@ -4252,6 +4254,8 @@ class DictProxyTests(unittest.TestCase): self.assertEqual(keys, ['__dict__', '__doc__', '__module__', '__weakref__', 'meth']) + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __local__') def test_iter_values(self): # Testing dict-proxy values... it = self.C.__dict__.values() @@ -4259,6 +4263,8 @@ class DictProxyTests(unittest.TestCase): values = list(it) self.assertEqual(len(values), 5) + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __local__') def test_iter_items(self): # Testing dict-proxy iteritems... it = self.C.__dict__.items() diff --git a/Lib/test/test_descrtut.py b/Lib/test/test_descrtut.py index 2db3d33..c3355b9 100644 --- a/Lib/test/test_descrtut.py +++ b/Lib/test/test_descrtut.py @@ -199,6 +199,8 @@ You can get the information from the list type: '__str__', '__subclasshook__', 'append', + 'clear', + 'copy', 'count', 'extend', 'index', diff --git a/Lib/test/test_dis.py b/Lib/test/test_dis.py index 7a61493..643e2e6 100644 --- a/Lib/test/test_dis.py +++ b/Lib/test/test_dis.py @@ -1,11 +1,35 @@ # Minimal tests for dis module from test.support import run_unittest, captured_stdout +import difflib import unittest import sys import dis import io +class _C: + def __init__(self, x): + self.x = x == 1 + +dis_c_instance_method = """\ + %-4d 0 LOAD_FAST 1 (x) + 3 LOAD_CONST 1 (1) + 6 COMPARE_OP 2 (==) + 9 LOAD_FAST 0 (self) + 12 STORE_ATTR 0 (x) + 15 LOAD_CONST 0 (None) + 18 RETURN_VALUE +""" % (_C.__init__.__code__.co_firstlineno + 1,) + +dis_c_instance_method_bytes = """\ + 0 LOAD_FAST 1 (1) + 3 LOAD_CONST 1 (1) + 6 COMPARE_OP 2 (==) + 9 LOAD_FAST 0 (0) + 12 STORE_ATTR 0 (0) + 15 LOAD_CONST 0 (0) + 18 RETURN_VALUE +""" def _f(a): print(a) @@ -23,6 +47,16 @@ dis_f = """\ _f.__code__.co_firstlineno + 2) +dis_f_co_code = """\ + 0 LOAD_GLOBAL 0 (0) + 3 LOAD_FAST 0 (0) + 6 CALL_FUNCTION 1 + 9 POP_TOP + 10 LOAD_CONST 1 (1) + 13 RETURN_VALUE +""" + + def bug708901(): for res in range(1, 10): @@ -138,18 +172,27 @@ dis_compound_stmt_str = """\ """ class DisTests(unittest.TestCase): - def do_disassembly_test(self, func, expected): + + def get_disassembly(self, func, lasti=-1, wrapper=True): s = io.StringIO() save_stdout = sys.stdout sys.stdout = s - dis.dis(func) - sys.stdout = save_stdout - got = s.getvalue() + try: + if wrapper: + dis.dis(func) + else: + dis.disassemble(func, lasti) + finally: + sys.stdout = save_stdout # Trim trailing blanks (if any). - lines = got.split('\n') - lines = [line.rstrip() for line in lines] - expected = expected.split("\n") - import difflib + return [line.rstrip() for line in s.getvalue().splitlines()] + + def get_disassemble_as_string(self, func, lasti=-1): + return '\n'.join(self.get_disassembly(func, lasti, False)) + + def do_disassembly_test(self, func, expected): + lines = self.get_disassembly(func) + expected = expected.splitlines() if expected != lines: self.fail( "events did not match expectation:\n" + @@ -211,6 +254,44 @@ class DisTests(unittest.TestCase): self.do_disassembly_test(simple_stmt_str, dis_simple_stmt_str) self.do_disassembly_test(compound_stmt_str, dis_compound_stmt_str) + def test_disassemble_bytes(self): + self.do_disassembly_test(_f.__code__.co_code, dis_f_co_code) + + def test_disassemble_method(self): + self.do_disassembly_test(_C(1).__init__, dis_c_instance_method) + + def test_disassemble_method_bytes(self): + method_bytecode = _C(1).__init__.__code__.co_code + self.do_disassembly_test(method_bytecode, dis_c_instance_method_bytes) + + def test_dis_none(self): + try: + del sys.last_traceback + except AttributeError: + pass + self.assertRaises(RuntimeError, dis.dis, None) + + def test_dis_object(self): + self.assertRaises(TypeError, dis.dis, object()) + + def test_dis_traceback(self): + try: + del sys.last_traceback + except AttributeError: + pass + + try: + 1/0 + except Exception as e: + tb = e.__traceback__ + sys.last_traceback = tb + + tb_dis = self.get_disassemble_as_string(tb.tb_frame.f_code, tb.tb_lasti) + self.do_disassembly_test(None, tb_dis) + + def test_dis_object(self): + self.assertRaises(TypeError, dis.dis, object()) + code_info_code_info = """\ Name: code_info Filename: (.*) @@ -363,6 +444,13 @@ class CodeInfoTests(unittest.TestCase): dis.show_code(x) self.assertRegex(output.getvalue(), expected+"\n") + def test_code_info_object(self): + self.assertRaises(TypeError, dis.code_info, object()) + + def test_pretty_flags_no_flags(self): + self.assertEqual(dis.pretty_flags(0), '0x0') + + def test_main(): run_unittest(DisTests, CodeInfoTests) diff --git a/Lib/test/test_doctest.py b/Lib/test/test_doctest.py index 13836ba..cd87179 100644 --- a/Lib/test/test_doctest.py +++ b/Lib/test/test_doctest.py @@ -5,6 +5,7 @@ Test script for doctest. from test import support import doctest import os +import sys # NOTE: There are some additional tests relating to interaction with @@ -373,7 +374,7 @@ We'll simulate a __file__ attr that ends in pyc: >>> tests = finder.find(sample_func) >>> print(tests) # doctest: +ELLIPSIS - [<DocTest sample_func from ...:17 (1 example)>] + [<DocTest sample_func from ...:18 (1 example)>] The exact name depends on how test_doctest was invoked, so allow for leading path components. @@ -1686,226 +1687,227 @@ Run the debugger on the docstring, and then restore sys.stdin. """ -def test_pdb_set_trace(): - """Using pdb.set_trace from a doctest. - - You can use pdb.set_trace from a doctest. To do so, you must - retrieve the set_trace function from the pdb module at the time - you use it. The doctest module changes sys.stdout so that it can - capture program output. It also temporarily replaces pdb.set_trace - with a version that restores stdout. This is necessary for you to - see debugger output. - - >>> doc = ''' - ... >>> x = 42 - ... >>> raise Exception('clé') - ... Traceback (most recent call last): - ... Exception: clé - ... >>> import pdb; pdb.set_trace() - ... ''' - >>> parser = doctest.DocTestParser() - >>> test = parser.get_doctest(doc, {}, "foo-bar@baz", "foo-bar@baz.py", 0) - >>> runner = doctest.DocTestRunner(verbose=False) - - To demonstrate this, we'll create a fake standard input that - captures our debugger input: - - >>> import tempfile - >>> real_stdin = sys.stdin - >>> sys.stdin = _FakeInput([ - ... 'print(x)', # print data defined by the example - ... 'continue', # stop debugging - ... '']) - - >>> try: runner.run(test) - ... finally: sys.stdin = real_stdin - --Return-- - > <doctest foo-bar@baz[2]>(1)<module>()->None - -> import pdb; pdb.set_trace() - (Pdb) print(x) - 42 - (Pdb) continue - TestResults(failed=0, attempted=3) - - You can also put pdb.set_trace in a function called from a test: - - >>> def calls_set_trace(): - ... y=2 - ... import pdb; pdb.set_trace() - - >>> doc = ''' - ... >>> x=1 - ... >>> calls_set_trace() - ... ''' - >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) - >>> real_stdin = sys.stdin - >>> sys.stdin = _FakeInput([ - ... 'print(y)', # print data defined in the function - ... 'up', # out of function - ... 'print(x)', # print data defined by the example - ... 'continue', # stop debugging - ... '']) - - >>> try: - ... runner.run(test) - ... finally: - ... sys.stdin = real_stdin - --Return-- - > <doctest test.test_doctest.test_pdb_set_trace[8]>(3)calls_set_trace()->None - -> import pdb; pdb.set_trace() - (Pdb) print(y) - 2 - (Pdb) up - > <doctest foo-bar@baz[1]>(1)<module>() - -> calls_set_trace() - (Pdb) print(x) - 1 - (Pdb) continue - TestResults(failed=0, attempted=2) - - During interactive debugging, source code is shown, even for - doctest examples: - - >>> doc = ''' - ... >>> def f(x): - ... ... g(x*2) - ... >>> def g(x): - ... ... print(x+3) - ... ... import pdb; pdb.set_trace() - ... >>> f(3) - ... ''' - >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) - >>> real_stdin = sys.stdin - >>> sys.stdin = _FakeInput([ - ... 'list', # list source from example 2 - ... 'next', # return from g() - ... 'list', # list source from example 1 - ... 'next', # return from f() - ... 'list', # list source from example 3 - ... 'continue', # stop debugging - ... '']) - >>> try: runner.run(test) - ... finally: sys.stdin = real_stdin - ... # doctest: +NORMALIZE_WHITESPACE - --Return-- - > <doctest foo-bar@baz[1]>(3)g()->None - -> import pdb; pdb.set_trace() - (Pdb) list - 1 def g(x): - 2 print(x+3) - 3 -> import pdb; pdb.set_trace() - [EOF] - (Pdb) next - --Return-- - > <doctest foo-bar@baz[0]>(2)f()->None - -> g(x*2) - (Pdb) list - 1 def f(x): - 2 -> g(x*2) - [EOF] - (Pdb) next - --Return-- - > <doctest foo-bar@baz[2]>(1)<module>()->None - -> f(3) - (Pdb) list - 1 -> f(3) - [EOF] - (Pdb) continue - ********************************************************************** - File "foo-bar@baz.py", line 7, in foo-bar@baz - Failed example: - f(3) - Expected nothing - Got: - 9 - TestResults(failed=1, attempted=3) - """ - -def test_pdb_set_trace_nested(): - """This illustrates more-demanding use of set_trace with nested functions. - - >>> class C(object): - ... def calls_set_trace(self): - ... y = 1 - ... import pdb; pdb.set_trace() - ... self.f1() - ... y = 2 - ... def f1(self): - ... x = 1 - ... self.f2() - ... x = 2 - ... def f2(self): - ... z = 1 - ... z = 2 - - >>> calls_set_trace = C().calls_set_trace - - >>> doc = ''' - ... >>> a = 1 - ... >>> calls_set_trace() - ... ''' - >>> parser = doctest.DocTestParser() - >>> runner = doctest.DocTestRunner(verbose=False) - >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) - >>> real_stdin = sys.stdin - >>> sys.stdin = _FakeInput([ - ... 'print(y)', # print data defined in the function - ... 'step', 'step', 'step', 'step', 'step', 'step', 'print(z)', - ... 'up', 'print(x)', - ... 'up', 'print(y)', - ... 'up', 'print(foo)', - ... 'continue', # stop debugging - ... '']) - - >>> try: - ... runner.run(test) - ... finally: - ... sys.stdin = real_stdin - ... # doctest: +REPORT_NDIFF - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(5)calls_set_trace() - -> self.f1() - (Pdb) print(y) - 1 - (Pdb) step - --Call-- - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(7)f1() - -> def f1(self): - (Pdb) step - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(8)f1() - -> x = 1 - (Pdb) step - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(9)f1() - -> self.f2() - (Pdb) step - --Call-- - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(11)f2() - -> def f2(self): - (Pdb) step - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(12)f2() - -> z = 1 - (Pdb) step - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(13)f2() - -> z = 2 - (Pdb) print(z) - 1 - (Pdb) up - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(9)f1() - -> self.f2() - (Pdb) print(x) - 1 - (Pdb) up - > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(5)calls_set_trace() - -> self.f1() - (Pdb) print(y) - 1 - (Pdb) up - > <doctest foo-bar@baz[1]>(1)<module>() - -> calls_set_trace() - (Pdb) print(foo) - *** NameError: name 'foo' is not defined - (Pdb) continue - TestResults(failed=0, attempted=2) -""" +if not hasattr(sys, 'gettrace') or not sys.gettrace(): + def test_pdb_set_trace(): + """Using pdb.set_trace from a doctest. + + You can use pdb.set_trace from a doctest. To do so, you must + retrieve the set_trace function from the pdb module at the time + you use it. The doctest module changes sys.stdout so that it can + capture program output. It also temporarily replaces pdb.set_trace + with a version that restores stdout. This is necessary for you to + see debugger output. + + >>> doc = ''' + ... >>> x = 42 + ... >>> raise Exception('clé') + ... Traceback (most recent call last): + ... Exception: clé + ... >>> import pdb; pdb.set_trace() + ... ''' + >>> parser = doctest.DocTestParser() + >>> test = parser.get_doctest(doc, {}, "foo-bar@baz", "foo-bar@baz.py", 0) + >>> runner = doctest.DocTestRunner(verbose=False) + + To demonstrate this, we'll create a fake standard input that + captures our debugger input: + + >>> import tempfile + >>> real_stdin = sys.stdin + >>> sys.stdin = _FakeInput([ + ... 'print(x)', # print data defined by the example + ... 'continue', # stop debugging + ... '']) + + >>> try: runner.run(test) + ... finally: sys.stdin = real_stdin + --Return-- + > <doctest foo-bar@baz[2]>(1)<module>()->None + -> import pdb; pdb.set_trace() + (Pdb) print(x) + 42 + (Pdb) continue + TestResults(failed=0, attempted=3) + + You can also put pdb.set_trace in a function called from a test: + + >>> def calls_set_trace(): + ... y=2 + ... import pdb; pdb.set_trace() + + >>> doc = ''' + ... >>> x=1 + ... >>> calls_set_trace() + ... ''' + >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) + >>> real_stdin = sys.stdin + >>> sys.stdin = _FakeInput([ + ... 'print(y)', # print data defined in the function + ... 'up', # out of function + ... 'print(x)', # print data defined by the example + ... 'continue', # stop debugging + ... '']) + + >>> try: + ... runner.run(test) + ... finally: + ... sys.stdin = real_stdin + --Return-- + > <doctest test.test_doctest.test_pdb_set_trace[8]>(3)calls_set_trace()->None + -> import pdb; pdb.set_trace() + (Pdb) print(y) + 2 + (Pdb) up + > <doctest foo-bar@baz[1]>(1)<module>() + -> calls_set_trace() + (Pdb) print(x) + 1 + (Pdb) continue + TestResults(failed=0, attempted=2) + + During interactive debugging, source code is shown, even for + doctest examples: + + >>> doc = ''' + ... >>> def f(x): + ... ... g(x*2) + ... >>> def g(x): + ... ... print(x+3) + ... ... import pdb; pdb.set_trace() + ... >>> f(3) + ... ''' + >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) + >>> real_stdin = sys.stdin + >>> sys.stdin = _FakeInput([ + ... 'list', # list source from example 2 + ... 'next', # return from g() + ... 'list', # list source from example 1 + ... 'next', # return from f() + ... 'list', # list source from example 3 + ... 'continue', # stop debugging + ... '']) + >>> try: runner.run(test) + ... finally: sys.stdin = real_stdin + ... # doctest: +NORMALIZE_WHITESPACE + --Return-- + > <doctest foo-bar@baz[1]>(3)g()->None + -> import pdb; pdb.set_trace() + (Pdb) list + 1 def g(x): + 2 print(x+3) + 3 -> import pdb; pdb.set_trace() + [EOF] + (Pdb) next + --Return-- + > <doctest foo-bar@baz[0]>(2)f()->None + -> g(x*2) + (Pdb) list + 1 def f(x): + 2 -> g(x*2) + [EOF] + (Pdb) next + --Return-- + > <doctest foo-bar@baz[2]>(1)<module>()->None + -> f(3) + (Pdb) list + 1 -> f(3) + [EOF] + (Pdb) continue + ********************************************************************** + File "foo-bar@baz.py", line 7, in foo-bar@baz + Failed example: + f(3) + Expected nothing + Got: + 9 + TestResults(failed=1, attempted=3) + """ + + def test_pdb_set_trace_nested(): + """This illustrates more-demanding use of set_trace with nested functions. + + >>> class C(object): + ... def calls_set_trace(self): + ... y = 1 + ... import pdb; pdb.set_trace() + ... self.f1() + ... y = 2 + ... def f1(self): + ... x = 1 + ... self.f2() + ... x = 2 + ... def f2(self): + ... z = 1 + ... z = 2 + + >>> calls_set_trace = C().calls_set_trace + + >>> doc = ''' + ... >>> a = 1 + ... >>> calls_set_trace() + ... ''' + >>> parser = doctest.DocTestParser() + >>> runner = doctest.DocTestRunner(verbose=False) + >>> test = parser.get_doctest(doc, globals(), "foo-bar@baz", "foo-bar@baz.py", 0) + >>> real_stdin = sys.stdin + >>> sys.stdin = _FakeInput([ + ... 'print(y)', # print data defined in the function + ... 'step', 'step', 'step', 'step', 'step', 'step', 'print(z)', + ... 'up', 'print(x)', + ... 'up', 'print(y)', + ... 'up', 'print(foo)', + ... 'continue', # stop debugging + ... '']) + + >>> try: + ... runner.run(test) + ... finally: + ... sys.stdin = real_stdin + ... # doctest: +REPORT_NDIFF + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(5)calls_set_trace() + -> self.f1() + (Pdb) print(y) + 1 + (Pdb) step + --Call-- + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(7)f1() + -> def f1(self): + (Pdb) step + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(8)f1() + -> x = 1 + (Pdb) step + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(9)f1() + -> self.f2() + (Pdb) step + --Call-- + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(11)f2() + -> def f2(self): + (Pdb) step + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(12)f2() + -> z = 1 + (Pdb) step + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(13)f2() + -> z = 2 + (Pdb) print(z) + 1 + (Pdb) up + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(9)f1() + -> self.f2() + (Pdb) print(x) + 1 + (Pdb) up + > <doctest test.test_doctest.test_pdb_set_trace_nested[0]>(5)calls_set_trace() + -> self.f1() + (Pdb) print(y) + 1 + (Pdb) up + > <doctest foo-bar@baz[1]>(1)<module>() + -> calls_set_trace() + (Pdb) print(foo) + *** NameError: name 'foo' is not defined + (Pdb) continue + TestResults(failed=0, attempted=2) + """ def test_DocTestSuite(): """DocTestSuite creates a unittest test suite from a doctest. diff --git a/Lib/test/test_dummy_thread.py b/Lib/test/test_dummy_thread.py index c61078d..2fafe1d 100644 --- a/Lib/test/test_dummy_thread.py +++ b/Lib/test/test_dummy_thread.py @@ -35,8 +35,8 @@ class LockTests(unittest.TestCase): "Lock object did not release properly.") def test_improper_release(self): - #Make sure release of an unlocked thread raises _thread.error - self.assertRaises(_thread.error, self.lock.release) + #Make sure release of an unlocked thread raises RuntimeError + self.assertRaises(RuntimeError, self.lock.release) def test_cond_acquire_success(self): #Make sure the conditional acquiring of the lock works. diff --git a/Lib/test/test_exceptions.py b/Lib/test/test_exceptions.py index 76f4249..592c765 100644 --- a/Lib/test/test_exceptions.py +++ b/Lib/test/test_exceptions.py @@ -7,7 +7,7 @@ import pickle import weakref from test.support import (TESTFN, unlink, run_unittest, captured_output, - gc_collect, cpython_only) + gc_collect, cpython_only, no_tracing) # XXX This is not really enough, each *operation* should be tested! @@ -388,6 +388,7 @@ class ExceptionTests(unittest.TestCase): x = DerivedException(fancy_arg=42) self.assertEqual(x.fancy_arg, 42) + @no_tracing def testInfiniteRecursion(self): def f(): return f() @@ -631,6 +632,7 @@ class ExceptionTests(unittest.TestCase): u.start = 1000 self.assertEqual(str(u), "can't translate characters in position 1000-4: 965230951443685724997") + @no_tracing def test_badisinstance(self): # Bug #2542: if issubclass(e, MyException) raises an exception, # it should be ignored @@ -741,6 +743,7 @@ class ExceptionTests(unittest.TestCase): self.fail("MemoryError not raised") self.assertEqual(wr(), None) + @no_tracing def test_recursion_error_cleanup(self): # Same test as above, but with "recursion exceeded" errors class C: diff --git a/Lib/test/test_fileinput.py b/Lib/test/test_fileinput.py index f312882..76e4d16 100644 --- a/Lib/test/test_fileinput.py +++ b/Lib/test/test_fileinput.py @@ -2,18 +2,31 @@ Tests for fileinput module. Nick Mathewson ''' - +import os +import sys +import re +import fileinput +import collections +import gzip +import types +import codecs import unittest -from test.support import verbose, TESTFN, run_unittest -from test.support import unlink as safe_unlink -import sys, re + +try: + import bz2 +except ImportError: + bz2 = None + from io import StringIO from fileinput import FileInput, hook_encoded +from test.support import verbose, TESTFN, run_unittest +from test.support import unlink as safe_unlink + + # The fileinput module has 2 interfaces: the FileInput class which does # all the work, and a few functions (input, etc.) that use a global _state -# variable. We only test the FileInput class, since the other functions -# only provide a thin facade over FileInput. +# variable. # Write lines (a list of lines) to temp file number i, and return the # temp file's name. @@ -121,7 +134,16 @@ class BufferSizesTests(unittest.TestCase): self.assertEqual(int(m.group(1)), fi.filelineno()) fi.close() +class UnconditionallyRaise: + def __init__(self, exception_type): + self.exception_type = exception_type + self.invoked = False + def __call__(self, *args, **kwargs): + self.invoked = True + raise self.exception_type() + class FileInputTests(unittest.TestCase): + def test_zero_byte_files(self): t1 = t2 = t3 = t4 = None try: @@ -219,17 +241,20 @@ class FileInputTests(unittest.TestCase): self.fail("FileInput should check openhook for being callable") except ValueError: pass - # XXX The rot13 codec was removed. - # So this test needs to be changed to use something else. - # (Or perhaps the API needs to change so we can just pass - # an encoding rather than using a hook?) -## try: -## t1 = writeTmp(1, ["A\nB"], mode="wb") -## fi = FileInput(files=t1, openhook=hook_encoded("rot13")) -## lines = list(fi) -## self.assertEqual(lines, ["N\n", "O"]) -## finally: -## remove_tempfiles(t1) + + class CustomOpenHook: + def __init__(self): + self.invoked = False + def __call__(self, *args): + self.invoked = True + return open(*args) + + t = writeTmp(1, ["\n"]) + self.addCleanup(remove_tempfiles, t) + custom_open_hook = CustomOpenHook() + with FileInput([t], openhook=custom_open_hook) as fi: + fi.readline() + self.assertTrue(custom_open_hook.invoked, "openhook not invoked") def test_context_manager(self): try: @@ -254,9 +279,584 @@ class FileInputTests(unittest.TestCase): finally: remove_tempfiles(t1) + def test_empty_files_list_specified_to_constructor(self): + with FileInput(files=[]) as fi: + self.assertEqual(fi._files, ('-',)) + + def test__getitem__(self): + """Tests invoking FileInput.__getitem__() with the current + line number""" + t = writeTmp(1, ["line1\n", "line2\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t]) as fi: + retval1 = fi[0] + self.assertEqual(retval1, "line1\n") + retval2 = fi[1] + self.assertEqual(retval2, "line2\n") + + def test__getitem__invalid_key(self): + """Tests invoking FileInput.__getitem__() with an index unequal to + the line number""" + t = writeTmp(1, ["line1\n", "line2\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t]) as fi: + with self.assertRaises(RuntimeError) as cm: + fi[1] + self.assertEqual(cm.exception.args, ("accessing lines out of order",)) + + def test__getitem__eof(self): + """Tests invoking FileInput.__getitem__() with the line number but at + end-of-input""" + t = writeTmp(1, []) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t]) as fi: + with self.assertRaises(IndexError) as cm: + fi[0] + self.assertEqual(cm.exception.args, ("end of input reached",)) + + def test_nextfile_oserror_deleting_backup(self): + """Tests invoking FileInput.nextfile() when the attempt to delete + the backup file would raise OSError. This error is expected to be + silently ignored""" + + os_unlink_orig = os.unlink + os_unlink_replacement = UnconditionallyRaise(OSError) + try: + t = writeTmp(1, ["\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t], inplace=True) as fi: + next(fi) # make sure the file is opened + os.unlink = os_unlink_replacement + fi.nextfile() + finally: + os.unlink = os_unlink_orig + + # sanity check to make sure that our test scenario was actually hit + self.assertTrue(os_unlink_replacement.invoked, + "os.unlink() was not invoked") + + def test_readline_os_fstat_raises_OSError(self): + """Tests invoking FileInput.readline() when os.fstat() raises OSError. + This exception should be silently discarded.""" + + os_fstat_orig = os.fstat + os_fstat_replacement = UnconditionallyRaise(OSError) + try: + t = writeTmp(1, ["\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t], inplace=True) as fi: + os.fstat = os_fstat_replacement + fi.readline() + finally: + os.fstat = os_fstat_orig + + # sanity check to make sure that our test scenario was actually hit + self.assertTrue(os_fstat_replacement.invoked, + "os.fstat() was not invoked") + + @unittest.skipIf(not hasattr(os, "chmod"), "os.chmod does not exist") + def test_readline_os_chmod_raises_OSError(self): + """Tests invoking FileInput.readline() when os.chmod() raises OSError. + This exception should be silently discarded.""" + + os_chmod_orig = os.chmod + os_chmod_replacement = UnconditionallyRaise(OSError) + try: + t = writeTmp(1, ["\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t], inplace=True) as fi: + os.chmod = os_chmod_replacement + fi.readline() + finally: + os.chmod = os_chmod_orig + + # sanity check to make sure that our test scenario was actually hit + self.assertTrue(os_chmod_replacement.invoked, + "os.fstat() was not invoked") + + def test_fileno_when_ValueError_raised(self): + class FilenoRaisesValueError(UnconditionallyRaise): + def __init__(self): + UnconditionallyRaise.__init__(self, ValueError) + def fileno(self): + self.__call__() + + unconditionally_raise_ValueError = FilenoRaisesValueError() + t = writeTmp(1, ["\n"]) + self.addCleanup(remove_tempfiles, t) + with FileInput(files=[t]) as fi: + file_backup = fi._file + try: + fi._file = unconditionally_raise_ValueError + result = fi.fileno() + finally: + fi._file = file_backup # make sure the file gets cleaned up + + # sanity check to make sure that our test scenario was actually hit + self.assertTrue(unconditionally_raise_ValueError.invoked, + "_file.fileno() was not invoked") + + self.assertEqual(result, -1, "fileno() should return -1") + +class MockFileInput: + """A class that mocks out fileinput.FileInput for use during unit tests""" + + def __init__(self, files=None, inplace=False, backup="", bufsize=0, + mode="r", openhook=None): + self.files = files + self.inplace = inplace + self.backup = backup + self.bufsize = bufsize + self.mode = mode + self.openhook = openhook + self._file = None + self.invocation_counts = collections.defaultdict(lambda: 0) + self.return_values = {} + + def close(self): + self.invocation_counts["close"] += 1 + + def nextfile(self): + self.invocation_counts["nextfile"] += 1 + return self.return_values["nextfile"] + + def filename(self): + self.invocation_counts["filename"] += 1 + return self.return_values["filename"] + + def lineno(self): + self.invocation_counts["lineno"] += 1 + return self.return_values["lineno"] + + def filelineno(self): + self.invocation_counts["filelineno"] += 1 + return self.return_values["filelineno"] + + def fileno(self): + self.invocation_counts["fileno"] += 1 + return self.return_values["fileno"] + + def isfirstline(self): + self.invocation_counts["isfirstline"] += 1 + return self.return_values["isfirstline"] + + def isstdin(self): + self.invocation_counts["isstdin"] += 1 + return self.return_values["isstdin"] + +class BaseFileInputGlobalMethodsTest(unittest.TestCase): + """Base class for unit tests for the global function of + the fileinput module.""" + + def setUp(self): + self._orig_state = fileinput._state + self._orig_FileInput = fileinput.FileInput + fileinput.FileInput = MockFileInput + + def tearDown(self): + fileinput.FileInput = self._orig_FileInput + fileinput._state = self._orig_state + + def assertExactlyOneInvocation(self, mock_file_input, method_name): + # assert that the method with the given name was invoked once + actual_count = mock_file_input.invocation_counts[method_name] + self.assertEqual(actual_count, 1, method_name) + # assert that no other unexpected methods were invoked + actual_total_count = len(mock_file_input.invocation_counts) + self.assertEqual(actual_total_count, 1) + +class Test_fileinput_input(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.input()""" + + def test_state_is_not_None_and_state_file_is_not_None(self): + """Tests invoking fileinput.input() when fileinput._state is not None + and its _file attribute is also not None. Expect RuntimeError to + be raised with a meaningful error message and for fileinput._state + to *not* be modified.""" + instance = MockFileInput() + instance._file = object() + fileinput._state = instance + with self.assertRaises(RuntimeError) as cm: + fileinput.input() + self.assertEqual(("input() already active",), cm.exception.args) + self.assertIs(instance, fileinput._state, "fileinput._state") + + def test_state_is_not_None_and_state_file_is_None(self): + """Tests invoking fileinput.input() when fileinput._state is not None + but its _file attribute *is* None. Expect it to create and return + a new fileinput.FileInput object with all method parameters passed + explicitly to the __init__() method; also ensure that + fileinput._state is set to the returned instance.""" + instance = MockFileInput() + instance._file = None + fileinput._state = instance + self.do_test_call_input() + + def test_state_is_None(self): + """Tests invoking fileinput.input() when fileinput._state is None + Expect it to create and return a new fileinput.FileInput object + with all method parameters passed explicitly to the __init__() + method; also ensure that fileinput._state is set to the returned + instance.""" + fileinput._state = None + self.do_test_call_input() + + def do_test_call_input(self): + """Tests that fileinput.input() creates a new fileinput.FileInput + object, passing the given parameters unmodified to + fileinput.FileInput.__init__(). Note that this test depends on the + monkey patching of fileinput.FileInput done by setUp().""" + files = object() + inplace = object() + backup = object() + bufsize = object() + mode = object() + openhook = object() + + # call fileinput.input() with different values for each argument + result = fileinput.input(files=files, inplace=inplace, backup=backup, + bufsize=bufsize, + mode=mode, openhook=openhook) + + # ensure fileinput._state was set to the returned object + self.assertIs(result, fileinput._state, "fileinput._state") + + # ensure the parameters to fileinput.input() were passed directly + # to FileInput.__init__() + self.assertIs(files, result.files, "files") + self.assertIs(inplace, result.inplace, "inplace") + self.assertIs(backup, result.backup, "backup") + self.assertIs(bufsize, result.bufsize, "bufsize") + self.assertIs(mode, result.mode, "mode") + self.assertIs(openhook, result.openhook, "openhook") + +class Test_fileinput_close(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.close()""" + + def test_state_is_None(self): + """Tests that fileinput.close() does nothing if fileinput._state + is None""" + fileinput._state = None + fileinput.close() + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests that fileinput.close() invokes close() on fileinput._state + and sets _state=None""" + instance = MockFileInput() + fileinput._state = instance + fileinput.close() + self.assertExactlyOneInvocation(instance, "close") + self.assertIsNone(fileinput._state) + +class Test_fileinput_nextfile(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.nextfile()""" + + def test_state_is_None(self): + """Tests fileinput.nextfile() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.nextfile() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.nextfile() when fileinput._state is not None. + Ensure that it invokes fileinput._state.nextfile() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + nextfile_retval = object() + instance = MockFileInput() + instance.return_values["nextfile"] = nextfile_retval + fileinput._state = instance + retval = fileinput.nextfile() + self.assertExactlyOneInvocation(instance, "nextfile") + self.assertIs(retval, nextfile_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_filename(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.filename()""" + + def test_state_is_None(self): + """Tests fileinput.filename() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.filename() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.filename() when fileinput._state is not None. + Ensure that it invokes fileinput._state.filename() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + filename_retval = object() + instance = MockFileInput() + instance.return_values["filename"] = filename_retval + fileinput._state = instance + retval = fileinput.filename() + self.assertExactlyOneInvocation(instance, "filename") + self.assertIs(retval, filename_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_lineno(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.lineno()""" + + def test_state_is_None(self): + """Tests fileinput.lineno() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.lineno() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.lineno() when fileinput._state is not None. + Ensure that it invokes fileinput._state.lineno() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + lineno_retval = object() + instance = MockFileInput() + instance.return_values["lineno"] = lineno_retval + fileinput._state = instance + retval = fileinput.lineno() + self.assertExactlyOneInvocation(instance, "lineno") + self.assertIs(retval, lineno_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_filelineno(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.filelineno()""" + + def test_state_is_None(self): + """Tests fileinput.filelineno() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.filelineno() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.filelineno() when fileinput._state is not None. + Ensure that it invokes fileinput._state.filelineno() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + filelineno_retval = object() + instance = MockFileInput() + instance.return_values["filelineno"] = filelineno_retval + fileinput._state = instance + retval = fileinput.filelineno() + self.assertExactlyOneInvocation(instance, "filelineno") + self.assertIs(retval, filelineno_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_fileno(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.fileno()""" + + def test_state_is_None(self): + """Tests fileinput.fileno() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.fileno() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.fileno() when fileinput._state is not None. + Ensure that it invokes fileinput._state.fileno() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + fileno_retval = object() + instance = MockFileInput() + instance.return_values["fileno"] = fileno_retval + instance.fileno_retval = fileno_retval + fileinput._state = instance + retval = fileinput.fileno() + self.assertExactlyOneInvocation(instance, "fileno") + self.assertIs(retval, fileno_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_isfirstline(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.isfirstline()""" + + def test_state_is_None(self): + """Tests fileinput.isfirstline() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.isfirstline() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.isfirstline() when fileinput._state is not None. + Ensure that it invokes fileinput._state.isfirstline() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + isfirstline_retval = object() + instance = MockFileInput() + instance.return_values["isfirstline"] = isfirstline_retval + fileinput._state = instance + retval = fileinput.isfirstline() + self.assertExactlyOneInvocation(instance, "isfirstline") + self.assertIs(retval, isfirstline_retval) + self.assertIs(fileinput._state, instance) + +class Test_fileinput_isstdin(BaseFileInputGlobalMethodsTest): + """Unit tests for fileinput.isstdin()""" + + def test_state_is_None(self): + """Tests fileinput.isstdin() when fileinput._state is None. + Ensure that it raises RuntimeError with a meaningful error message + and does not modify fileinput._state""" + fileinput._state = None + with self.assertRaises(RuntimeError) as cm: + fileinput.isstdin() + self.assertEqual(("no active input()",), cm.exception.args) + self.assertIsNone(fileinput._state) + + def test_state_is_not_None(self): + """Tests fileinput.isstdin() when fileinput._state is not None. + Ensure that it invokes fileinput._state.isstdin() exactly once, + returns whatever it returns, and does not modify fileinput._state + to point to a different object.""" + isstdin_retval = object() + instance = MockFileInput() + instance.return_values["isstdin"] = isstdin_retval + fileinput._state = instance + retval = fileinput.isstdin() + self.assertExactlyOneInvocation(instance, "isstdin") + self.assertIs(retval, isstdin_retval) + self.assertIs(fileinput._state, instance) + +class InvocationRecorder: + def __init__(self): + self.invocation_count = 0 + def __call__(self, *args, **kwargs): + self.invocation_count += 1 + self.last_invocation = (args, kwargs) + +class Test_hook_compressed(unittest.TestCase): + """Unit tests for fileinput.hook_compressed()""" + + def setUp(self): + self.fake_open = InvocationRecorder() + + def test_empty_string(self): + self.do_test_use_builtin_open("", 1) + + def test_no_ext(self): + self.do_test_use_builtin_open("abcd", 2) + + def test_gz_ext_fake(self): + original_open = gzip.open + gzip.open = self.fake_open + try: + result = fileinput.hook_compressed("test.gz", 3) + finally: + gzip.open = original_open + + self.assertEqual(self.fake_open.invocation_count, 1) + self.assertEqual(self.fake_open.last_invocation, (("test.gz", 3), {})) + + @unittest.skipUnless(bz2, "Requires bz2") + def test_bz2_ext_fake(self): + original_open = bz2.BZ2File + bz2.BZ2File = self.fake_open + try: + result = fileinput.hook_compressed("test.bz2", 4) + finally: + bz2.BZ2File = original_open + + self.assertEqual(self.fake_open.invocation_count, 1) + self.assertEqual(self.fake_open.last_invocation, (("test.bz2", 4), {})) + + def test_blah_ext(self): + self.do_test_use_builtin_open("abcd.blah", 5) + + def test_gz_ext_builtin(self): + self.do_test_use_builtin_open("abcd.Gz", 6) + + def test_bz2_ext_builtin(self): + self.do_test_use_builtin_open("abcd.Bz2", 7) + + def do_test_use_builtin_open(self, filename, mode): + original_open = self.replace_builtin_open(self.fake_open) + try: + result = fileinput.hook_compressed(filename, mode) + finally: + self.replace_builtin_open(original_open) + + self.assertEqual(self.fake_open.invocation_count, 1) + self.assertEqual(self.fake_open.last_invocation, + ((filename, mode), {})) + + @staticmethod + def replace_builtin_open(new_open_func): + builtins_type = type(__builtins__) + if builtins_type is dict: + original_open = __builtins__["open"] + __builtins__["open"] = new_open_func + elif builtins_type is types.ModuleType: + original_open = __builtins__.open + __builtins__.open = new_open_func + else: + raise RuntimeError( + "unknown __builtins__ type: %r (unable to replace open)" % + builtins_type) + + return original_open + +class Test_hook_encoded(unittest.TestCase): + """Unit tests for fileinput.hook_encoded()""" + + def test(self): + encoding = object() + result = fileinput.hook_encoded(encoding) + + fake_open = InvocationRecorder() + original_open = codecs.open + codecs.open = fake_open + try: + filename = object() + mode = object() + open_result = result(filename, mode) + finally: + codecs.open = original_open + + self.assertEqual(fake_open.invocation_count, 1) + + args = fake_open.last_invocation[0] + self.assertIs(args[0], filename) + self.assertIs(args[1], mode) + self.assertIs(args[2], encoding) def test_main(): - run_unittest(BufferSizesTests, FileInputTests) + run_unittest( + BufferSizesTests, + FileInputTests, + Test_fileinput_input, + Test_fileinput_close, + Test_fileinput_nextfile, + Test_fileinput_filename, + Test_fileinput_lineno, + Test_fileinput_filelineno, + Test_fileinput_fileno, + Test_fileinput_isfirstline, + Test_fileinput_isstdin, + Test_hook_compressed, + Test_hook_encoded, + ) if __name__ == "__main__": test_main() diff --git a/Lib/test/test_float.py b/Lib/test/test_float.py index 1968b8a..4e6f854 100644 --- a/Lib/test/test_float.py +++ b/Lib/test/test_float.py @@ -88,7 +88,7 @@ class GeneralFloatCases(unittest.TestCase): self.assertRaises(ValueError, float, " -0x3.p-1 ") self.assertRaises(ValueError, float, " +0x3.p-1 ") self.assertEqual(float(" 25.e-1 "), 2.5) - self.assertEqual(support.fcmp(float(" .25e-1 "), .025), 0) + self.assertAlmostEqual(float(" .25e-1 "), .025) def test_floatconversion(self): # Make sure that calls to __float__() work properly diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py index 9d2eab7..127b1b9 100644 --- a/Lib/test/test_ftplib.py +++ b/Lib/test/test_ftplib.py @@ -608,6 +608,20 @@ class TestFTPClass(TestCase): self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') self.assertFalse(is_client_connected()) + def test_source_address(self): + self.client.quit() + port = support.find_unused_port() + self.client.connect(self.server.host, self.server.port, + source_address=(HOST, port)) + self.assertEqual(self.client.sock.getsockname()[1], port) + self.client.quit() + + def test_source_address_passive_connection(self): + port = support.find_unused_port() + self.client.source_address = (HOST, port) + with self.client.transfercmd('list') as sock: + self.assertEqual(sock.getsockname()[1], port) + def test_parse257(self): self.assertEqual(ftplib.parse257('257 "/foo/bar"'), '/foo/bar') self.assertEqual(ftplib.parse257('257 "/foo/bar" created'), '/foo/bar') diff --git a/Lib/test/test_gc.py b/Lib/test/test_gc.py index 0e5a397..f60d5d9 100644 --- a/Lib/test/test_gc.py +++ b/Lib/test/test_gc.py @@ -1,5 +1,6 @@ import unittest -from test.support import verbose, run_unittest, strip_python_stderr +from test.support import (verbose, refcount_test, run_unittest, + strip_python_stderr) import sys import gc import weakref @@ -175,6 +176,7 @@ class GCTests(unittest.TestCase): del d self.assertEqual(gc.collect(), 2) + @refcount_test def test_frame(self): def f(): frame = sys._getframe() @@ -242,6 +244,7 @@ class GCTests(unittest.TestCase): # For example: # - disposed tuples are not freed, but reused # - the call to assertEqual somehow avoids building its args tuple + @refcount_test def test_get_count(self): # Avoid future allocation of method object assertEqual = self._baseAssertEqual @@ -252,6 +255,7 @@ class GCTests(unittest.TestCase): # the dict, and the tuple returned by get_count() assertEqual(gc.get_count(), (2, 0, 0)) + @refcount_test def test_collect_generations(self): # Avoid future allocation of method object assertEqual = self.assertEqual diff --git a/Lib/test/test_genexps.py b/Lib/test/test_genexps.py index 1f46af1..413043c 100644 --- a/Lib/test/test_genexps.py +++ b/Lib/test/test_genexps.py @@ -257,11 +257,15 @@ Verify that genexps are weakly referencable """ +import sys -__test__ = {'doctests' : doctests} +# Trace function can throw off the tuple reuse test. +if hasattr(sys, 'gettrace') and sys.gettrace(): + __test__ = {} +else: + __test__ = {'doctests' : doctests} def test_main(verbose=None): - import sys from test import support from test import test_genexps support.run_doctest(test_genexps, verbose) diff --git a/Lib/test/test_import.py b/Lib/test/test_import.py index cf17c96..ebcc2ca 100644 --- a/Lib/test/test_import.py +++ b/Lib/test/test_import.py @@ -283,8 +283,6 @@ class ImportTests(unittest.TestCase): self.skipTest('path is not encodable to {}'.format(encoding)) with self.assertRaises(ImportError) as c: __import__(path) - self.assertEqual("Import by filename is not supported.", - c.exception.args[0]) def test_import_in_del_does_not_crash(self): # Issue 4236 diff --git a/Lib/test/test_importhooks.py b/Lib/test/test_importhooks.py index ec6730e..dca5685 100644 --- a/Lib/test/test_importhooks.py +++ b/Lib/test/test_importhooks.py @@ -51,7 +51,7 @@ class TestImporter: def __init__(self, path=test_path): if path != test_path: - # if out class is on sys.path_hooks, we must raise + # if our class is on sys.path_hooks, we must raise # ImportError for any path item that we can't handle. raise ImportError self.path = path diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index b665a6c..1821492 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -46,7 +46,7 @@ except ImportError: def _default_chunk_size(): """Get the default TextIOWrapper chunk size""" - with open(__file__, "r", encoding="latin1") as f: + with open(__file__, "r", encoding="latin-1") as f: return f._CHUNK_SIZE @@ -1684,11 +1684,11 @@ class TextIOWrapperTest(unittest.TestCase): r = self.BytesIO(b"\xc3\xa9\n\n") b = self.BufferedReader(r, 1000) t = self.TextIOWrapper(b) - t.__init__(b, encoding="latin1", newline="\r\n") - self.assertEqual(t.encoding, "latin1") + t.__init__(b, encoding="latin-1", newline="\r\n") + self.assertEqual(t.encoding, "latin-1") self.assertEqual(t.line_buffering, False) - t.__init__(b, encoding="utf8", line_buffering=True) - self.assertEqual(t.encoding, "utf8") + t.__init__(b, encoding="utf-8", line_buffering=True) + self.assertEqual(t.encoding, "utf-8") self.assertEqual(t.line_buffering, True) self.assertEqual("\xe9\n", t.readline()) self.assertRaises(TypeError, t.__init__, b, newline=42) @@ -1738,8 +1738,8 @@ class TextIOWrapperTest(unittest.TestCase): def test_encoding(self): # Check the encoding attribute is always set, and valid b = self.BytesIO() - t = self.TextIOWrapper(b, encoding="utf8") - self.assertEqual(t.encoding, "utf8") + t = self.TextIOWrapper(b, encoding="utf-8") + self.assertEqual(t.encoding, "utf-8") t = self.TextIOWrapper(b) self.assertTrue(t.encoding is not None) codecs.lookup(t.encoding) @@ -1918,7 +1918,7 @@ class TextIOWrapperTest(unittest.TestCase): def test_basic_io(self): for chunksize in (1, 2, 3, 4, 5, 15, 16, 17, 31, 32, 33, 63, 64, 65): - for enc in "ascii", "latin1", "utf8" :# , "utf-16-be", "utf-16-le": + for enc in "ascii", "latin-1", "utf-8" :# , "utf-16-be", "utf-16-le": f = self.open(support.TESTFN, "w+", encoding=enc) f._CHUNK_SIZE = chunksize self.assertEqual(f.write("abc"), 3) @@ -1968,7 +1968,7 @@ class TextIOWrapperTest(unittest.TestCase): self.assertEqual(rlines, wlines) def test_telling(self): - f = self.open(support.TESTFN, "w+", encoding="utf8") + f = self.open(support.TESTFN, "w+", encoding="utf-8") p0 = f.tell() f.write("\xff\n") p1 = f.tell() @@ -2214,6 +2214,7 @@ class TextIOWrapperTest(unittest.TestCase): with self.open(support.TESTFN, "w", errors="replace") as f: self.assertEqual(f.errors, "replace") + @support.no_tracing @unittest.skipUnless(threading, 'Threading required for this test.') def test_threads_write(self): # Issue6750: concurrent writes could duplicate data @@ -2670,6 +2671,7 @@ class SignalsTest(unittest.TestCase): def test_interrupted_write_text(self): self.check_interrupted_write("xy", b"xy", mode="w", encoding="ascii") + @support.no_tracing def check_reentrant_write(self, data, **fdopen_kwargs): def on_alarm(*args): # Will be called reentrantly from the same thread diff --git a/Lib/test/test_mailbox.py b/Lib/test/test_mailbox.py index 1e4f887..03f814a 100644 --- a/Lib/test/test_mailbox.py +++ b/Lib/test/test_mailbox.py @@ -95,14 +95,14 @@ class TestMailbox(TestBase): """) def test_add_invalid_8bit_bytes_header(self): - key = self._box.add(self._nonascii_msg.encode('latin1')) + key = self._box.add(self._nonascii_msg.encode('latin-1')) self.assertEqual(len(self._box), 1) self.assertEqual(self._box.get_bytes(key), - self._nonascii_msg.encode('latin1')) + self._nonascii_msg.encode('latin-1')) def test_invalid_nonascii_header_as_string(self): subj = self._nonascii_msg.splitlines()[1] - key = self._box.add(subj.encode('latin1')) + key = self._box.add(subj.encode('latin-1')) self.assertEqual(self._box.get_string(key), 'Subject: =?unknown-8bit?b?RmFsaW5hcHThciBo4Xpob3pzeuFsbO104XNz' 'YWwuIE3hciByZW5kZWx06Ww/?=\n\n') diff --git a/Lib/test/test_metaclass.py b/Lib/test/test_metaclass.py index 219ab99..6862900 100644 --- a/Lib/test/test_metaclass.py +++ b/Lib/test/test_metaclass.py @@ -246,7 +246,13 @@ Test failures in looking up the __prepare__ method work. """ -__test__ = {'doctests' : doctests} +import sys + +# Trace function introduces __locals__ which causes various tests to fail. +if hasattr(sys, 'gettrace') and sys.gettrace(): + __test__ = {} +else: + __test__ = {'doctests' : doctests} def test_main(verbose=False): from test import support diff --git a/Lib/test/test_minidom.py b/Lib/test/test_minidom.py index 4c2b34a..acc4819 100644 --- a/Lib/test/test_minidom.py +++ b/Lib/test/test_minidom.py @@ -4,9 +4,7 @@ import pickle from test.support import verbose, run_unittest, findfile import unittest -import xml.dom import xml.dom.minidom -import xml.parsers.expat from xml.dom.minidom import parse, Node, Document, parseString from xml.dom.minidom import getDOMImplementation @@ -14,7 +12,6 @@ from xml.dom.minidom import getDOMImplementation tstfile = findfile("test.xml", subdir="xmltestdata") - # The tests of DocumentType importing use these helpers to construct # the documents to work with, since not all DOM builders actually # create the DocumentType nodes. @@ -1009,41 +1006,6 @@ class MinidomTest(unittest.TestCase): "test NodeList.item()") doc.unlink() - def testSAX2DOM(self): - from xml.dom import pulldom - - sax2dom = pulldom.SAX2DOM() - sax2dom.startDocument() - sax2dom.startElement("doc", {}) - sax2dom.characters("text") - sax2dom.startElement("subelm", {}) - sax2dom.characters("text") - sax2dom.endElement("subelm") - sax2dom.characters("text") - sax2dom.endElement("doc") - sax2dom.endDocument() - - doc = sax2dom.document - root = doc.documentElement - (text1, elm1, text2) = root.childNodes - text3 = elm1.childNodes[0] - - self.confirm(text1.previousSibling is None and - text1.nextSibling is elm1 and - elm1.previousSibling is text1 and - elm1.nextSibling is text2 and - text2.previousSibling is elm1 and - text2.nextSibling is None and - text3.previousSibling is None and - text3.nextSibling is None, "testSAX2DOM - siblings") - - self.confirm(root.parentNode is doc and - text1.parentNode is root and - elm1.parentNode is root and - text2.parentNode is root and - text3.parentNode is elm1, "testSAX2DOM - parents") - doc.unlink() - def testEncodings(self): doc = parseString('<foo>€</foo>') self.assertEqual(doc.toxml(), @@ -1490,6 +1452,7 @@ class MinidomTest(unittest.TestCase): doc.appendChild(doc.createComment("foo--bar")) self.assertRaises(ValueError, doc.toxml) + def testEmptyXMLNSValue(self): doc = parseString("<element xmlns=''>\n" "<foo/>\n</element>") diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py index 91e2cbb..139ab03 100644 --- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -163,6 +163,18 @@ class _TestProcess(BaseTestCase): self.assertEqual(current.ident, os.getpid()) self.assertEqual(current.exitcode, None) + def test_daemon_argument(self): + if self.TYPE == "threads": + return + + # By default uses the current process's daemon flag. + proc0 = self.Process(target=self._test) + self.assertEqual(proc0.daemon, self.current_process().daemon) + proc1 = self.Process(target=self._test, daemon=True) + self.assertTrue(proc1.daemon) + proc2 = self.Process(target=self._test, daemon=False) + self.assertFalse(proc2.daemon) + @classmethod def _test(cls, q, *args, **kwds): current = cls.current_process() diff --git a/Lib/test/test_nntplib.py b/Lib/test/test_nntplib.py index a387f61..ec790ad 100644 --- a/Lib/test/test_nntplib.py +++ b/Lib/test/test_nntplib.py @@ -1,10 +1,11 @@ import io +import socket import datetime import textwrap import unittest import functools import contextlib -import collections +import collections.abc from test import support from nntplib import NNTP, GroupInfo, _have_ssl import nntplib @@ -246,12 +247,32 @@ class NetworkedNNTPTestsMixin: if not name.startswith('test_'): continue meth = getattr(cls, name) - if not isinstance(meth, collections.Callable): + if not isinstance(meth, collections.abc.Callable): continue # Need to use a closure so that meth remains bound to its current # value setattr(cls, name, wrap_meth(meth)) + def test_with_statement(self): + def is_connected(): + if not hasattr(server, 'file'): + return False + try: + server.help() + except (socket.error, EOFError): + return False + return True + + with self.NNTP_CLASS(self.NNTP_HOST, timeout=TIMEOUT, usenetrc=False) as server: + self.assertTrue(is_connected()) + self.assertTrue(server.help()) + self.assertFalse(is_connected()) + + with self.NNTP_CLASS(self.NNTP_HOST, timeout=TIMEOUT, usenetrc=False) as server: + server.quit() + self.assertFalse(is_connected()) + + NetworkedNNTPTestsMixin.wrap_methods() @@ -813,7 +834,7 @@ class NNTPv1v2TestsMixin: def _check_article_body(self, lines): self.assertEqual(len(lines), 4) - self.assertEqual(lines[-1].decode('utf8'), "-- Signed by André.") + self.assertEqual(lines[-1].decode('utf-8'), "-- Signed by André.") self.assertEqual(lines[-2], b"") self.assertEqual(lines[-3], b".Here is a dot-starting line.") self.assertEqual(lines[-4], b"This is just a test article.") diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index 77fcbaf..e40f763 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -15,6 +15,13 @@ from test import support import contextlib import mmap import uuid +import asyncore +import asynchat +import socket +try: + import threading +except ImportError: + threading = None # Detect whether we're on a Linux system that uses the (now outdated # and unmaintained) linuxthreads threading library. There's an issue @@ -1261,6 +1268,287 @@ class LoginTests(unittest.TestCase): self.assertNotEqual(len(user_name), 0) +@unittest.skipUnless(hasattr(os, 'getpriority') and hasattr(os, 'setpriority'), + "needs os.getpriority and os.setpriority") +class ProgramPriorityTests(unittest.TestCase): + """Tests for os.getpriority() and os.setpriority().""" + + def test_set_get_priority(self): + + base = os.getpriority(os.PRIO_PROCESS, os.getpid()) + os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1) + try: + new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid()) + if base >= 19 and new_prio <= 19: + raise unittest.SkipTest( + "unable to reliably test setpriority at current nice level of %s" % base) + else: + self.assertEqual(new_prio, base + 1) + finally: + try: + os.setpriority(os.PRIO_PROCESS, os.getpid(), base) + except OSError as err: + if err.errno != errno.EACCES: + raise + + +class SendfileTestServer(asyncore.dispatcher, threading.Thread): + + class Handler(asynchat.async_chat): + + def __init__(self, conn): + asynchat.async_chat.__init__(self, conn) + self.in_buffer = [] + self.closed = False + self.push(b"220 ready\r\n") + + def handle_read(self): + data = self.recv(4096) + self.in_buffer.append(data) + + def get_data(self): + return b''.join(self.in_buffer) + + def handle_close(self): + self.close() + self.closed = True + + def handle_error(self): + raise + + def __init__(self, address): + threading.Thread.__init__(self) + asyncore.dispatcher.__init__(self) + self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.bind(address) + self.listen(5) + self.host, self.port = self.socket.getsockname()[:2] + self.handler_instance = None + self._active = False + self._active_lock = threading.Lock() + + # --- public API + + @property + def running(self): + return self._active + + def start(self): + assert not self.running + self.__flag = threading.Event() + threading.Thread.start(self) + self.__flag.wait() + + def stop(self): + assert self.running + self._active = False + self.join() + + def wait(self): + # wait for handler connection to be closed, then stop the server + while not getattr(self.handler_instance, "closed", False): + time.sleep(0.001) + self.stop() + + # --- internals + + def run(self): + self._active = True + self.__flag.set() + while self._active and asyncore.socket_map: + self._active_lock.acquire() + asyncore.loop(timeout=0.001, count=1) + self._active_lock.release() + asyncore.close_all() + + def handle_accept(self): + conn, addr = self.accept() + self.handler_instance = self.Handler(conn) + + def handle_connect(self): + self.close() + handle_read = handle_connect + + def writable(self): + return 0 + + def handle_error(self): + raise + + +@unittest.skipUnless(threading is not None, "test needs threading module") +@unittest.skipUnless(hasattr(os, 'sendfile'), "test needs os.sendfile()") +class TestSendfile(unittest.TestCase): + + DATA = b"12345abcde" * 16 * 1024 # 160 KB + SUPPORT_HEADERS_TRAILERS = not sys.platform.startswith("linux") and \ + not sys.platform.startswith("solaris") and \ + not sys.platform.startswith("sunos") + + @classmethod + def setUpClass(cls): + with open(support.TESTFN, "wb") as f: + f.write(cls.DATA) + + @classmethod + def tearDownClass(cls): + support.unlink(support.TESTFN) + + def setUp(self): + self.server = SendfileTestServer((support.HOST, 0)) + self.server.start() + self.client = socket.socket() + self.client.connect((self.server.host, self.server.port)) + self.client.settimeout(1) + # synchronize by waiting for "220 ready" response + self.client.recv(1024) + self.sockno = self.client.fileno() + self.file = open(support.TESTFN, 'rb') + self.fileno = self.file.fileno() + + def tearDown(self): + self.file.close() + self.client.close() + if self.server.running: + self.server.stop() + + def sendfile_wrapper(self, sock, file, offset, nbytes, headers=[], trailers=[]): + """A higher level wrapper representing how an application is + supposed to use sendfile(). + """ + while 1: + try: + if self.SUPPORT_HEADERS_TRAILERS: + return os.sendfile(sock, file, offset, nbytes, headers, + trailers) + else: + return os.sendfile(sock, file, offset, nbytes) + except OSError as err: + if err.errno == errno.ECONNRESET: + # disconnected + raise + elif err.errno in (errno.EAGAIN, errno.EBUSY): + # we have to retry send data + continue + else: + raise + + def test_send_whole_file(self): + # normal send + total_sent = 0 + offset = 0 + nbytes = 4096 + while total_sent < len(self.DATA): + sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes) + if sent == 0: + break + offset += sent + total_sent += sent + self.assertTrue(sent <= nbytes) + self.assertEqual(offset, total_sent) + + self.assertEqual(total_sent, len(self.DATA)) + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(len(data), len(self.DATA)) + self.assertEqual(data, self.DATA) + + def test_send_at_certain_offset(self): + # start sending a file at a certain offset + total_sent = 0 + offset = len(self.DATA) // 2 + must_send = len(self.DATA) - offset + nbytes = 4096 + while total_sent < must_send: + sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes) + if sent == 0: + break + offset += sent + total_sent += sent + self.assertTrue(sent <= nbytes) + + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + expected = self.DATA[len(self.DATA) // 2:] + self.assertEqual(total_sent, len(expected)) + self.assertEqual(len(data), len(expected)) + self.assertEqual(data, expected) + + def test_offset_overflow(self): + # specify an offset > file size + offset = len(self.DATA) + 4096 + try: + sent = os.sendfile(self.sockno, self.fileno, offset, 4096) + except OSError as e: + # Solaris can raise EINVAL if offset >= file length, ignore. + if e.errno != errno.EINVAL: + raise + else: + self.assertEqual(sent, 0) + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(data, b'') + + def test_invalid_offset(self): + with self.assertRaises(OSError) as cm: + os.sendfile(self.sockno, self.fileno, -1, 4096) + self.assertEqual(cm.exception.errno, errno.EINVAL) + + # --- headers / trailers tests + + if SUPPORT_HEADERS_TRAILERS: + + def test_headers(self): + total_sent = 0 + sent = os.sendfile(self.sockno, self.fileno, 0, 4096, + headers=[b"x" * 512]) + total_sent += sent + offset = 4096 + nbytes = 4096 + while 1: + sent = self.sendfile_wrapper(self.sockno, self.fileno, + offset, nbytes) + if sent == 0: + break + total_sent += sent + offset += sent + + expected_data = b"x" * 512 + self.DATA + self.assertEqual(total_sent, len(expected_data)) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(hash(data), hash(expected_data)) + + def test_trailers(self): + TESTFN2 = support.TESTFN + "2" + with open(TESTFN2, 'wb') as f: + f.write(b"abcde") + with open(TESTFN2, 'rb')as f: + self.addCleanup(os.remove, TESTFN2) + os.sendfile(self.sockno, f.fileno(), 0, 4096, + trailers=[b"12345"]) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(data, b"abcde12345") + + if hasattr(os, "SF_NODISKIO"): + def test_flags(self): + try: + os.sendfile(self.sockno, self.fileno, 0, 4096, + flags=os.SF_NODISKIO) + except OSError as err: + if err.errno not in (errno.EBUSY, errno.EAGAIN): + raise + + def test_main(): support.run_unittest( FileTests, @@ -1281,6 +1569,8 @@ def test_main(): PidTests, LoginTests, LinkTests, + TestSendfile, + ProgramPriorityTests, ) if __name__ == "__main__": diff --git a/Lib/test/test_osx_env.py b/Lib/test/test_osx_env.py index 8b3df37..24ec2b4 100644 --- a/Lib/test/test_osx_env.py +++ b/Lib/test/test_osx_env.py @@ -5,6 +5,7 @@ Test suite for OS X interpreter environment variables. from test.support import EnvironmentVarGuard, run_unittest import subprocess import sys +import sysconfig import unittest class OSXEnvironmentVariableTestCase(unittest.TestCase): @@ -27,8 +28,6 @@ class OSXEnvironmentVariableTestCase(unittest.TestCase): self._check_sys('PYTHONEXECUTABLE', '==', 'sys.executable') def test_main(): - from distutils import sysconfig - if sys.platform == 'darwin' and sysconfig.get_config_var('WITH_NEXT_FRAMEWORK'): run_unittest(OSXEnvironmentVariableTestCase) diff --git a/Lib/test/test_pdb.py b/Lib/test/test_pdb.py index d861df5..c197aff 100644 --- a/Lib/test/test_pdb.py +++ b/Lib/test/test_pdb.py @@ -20,9 +20,12 @@ class PdbTestInput(object): def __enter__(self): self.real_stdin = sys.stdin sys.stdin = _FakeInput(self.input) + self.orig_trace = sys.gettrace() if hasattr(sys, 'gettrace') else None def __exit__(self, *exc): sys.stdin = self.real_stdin + if self.orig_trace: + sys.settrace(self.orig_trace) def test_pdb_displayhook(): diff --git a/Lib/test/test_peepholer.py b/Lib/test/test_peepholer.py index b7d446f..f73565e 100644 --- a/Lib/test/test_peepholer.py +++ b/Lib/test/test_peepholer.py @@ -8,8 +8,10 @@ def disassemble(func): f = StringIO() tmp = sys.stdout sys.stdout = f - dis.dis(func) - sys.stdout = tmp + try: + dis.dis(func) + finally: + sys.stdout = tmp result = f.getvalue() f.close() return result @@ -17,6 +19,7 @@ def disassemble(func): def dis_single(line): return disassemble(compile(line, '', 'single')) + class TestTranforms(unittest.TestCase): def test_unot(self): @@ -99,6 +102,12 @@ class TestTranforms(unittest.TestCase): self.assertIn(elem, asm) self.assertNotIn('BUILD_TUPLE', asm) + # Long tuples should be folded too. + asm = dis_single(repr(tuple(range(10000)))) + # One LOAD_CONST for the tuple, one for the None return value + self.assertEqual(asm.count('LOAD_CONST'), 2) + self.assertNotIn('BUILD_TUPLE', asm) + # Bug 1053819: Tuple of constants misidentified when presented with: # . . . opcode_with_arg 100 unary_opcode BUILD_TUPLE 1 . . . # The following would segfault upon compilation @@ -267,6 +276,25 @@ class TestTranforms(unittest.TestCase): asm = disassemble(f) self.assertNotIn('BINARY_ADD', asm) + def test_constant_folding(self): + # Issue #11244: aggressive constant folding. + exprs = [ + "3 * -5", + "-3 * 5", + "2 * (3 * 4)", + "(2 * 3) * 4", + "(-1, 2, 3)", + "(1, -2, 3)", + "(1, 2, -3)", + "(1, 2, -3) * 6", + "lambda x: x in {(3 * -5) + (-1 - 6), (1, -2, 3) * 2, None}", + ] + for e in exprs: + asm = dis_single(e) + self.assertNotIn('UNARY_', asm, e) + self.assertNotIn('BINARY_', asm, e) + self.assertNotIn('BUILD_', asm, e) + class TestBuglets(unittest.TestCase): def test_bug_11510(self): diff --git a/Lib/test/test_pep292.py b/Lib/test/test_pep292.py index 119c7ea..a1e52e9 100644 --- a/Lib/test/test_pep292.py +++ b/Lib/test/test_pep292.py @@ -49,11 +49,11 @@ class TestTemplate(unittest.TestCase): (?P<invalid>) | (?P<escaped>%(delim)s) | @(?P<named>%(id)s) | - @{(?P<braced>%(id)s)} + @{(?P<braced>%(id)s)} ) """ s = MyPattern('$') - self.assertRaises(ValueError, s.substitute, dict()) + self.assertRaises(ValueError, s.substitute, dict()) def test_percents(self): eq = self.assertEqual diff --git a/Lib/test/test_pep3120.py b/Lib/test/test_pep3120.py index 09fedf0..496f8da 100644 --- a/Lib/test/test_pep3120.py +++ b/Lib/test/test_pep3120.py @@ -19,8 +19,8 @@ class PEP3120Test(unittest.TestCase): try: import test.badsyntax_pep3120 except SyntaxError as msg: - msg = str(msg) - self.assertTrue('UTF-8' in msg or 'utf8' in msg) + msg = str(msg).lower() + self.assertTrue('utf-8' in msg or 'utf8' in msg) else: self.fail("expected exception didn't occur") diff --git a/Lib/test/test_platform.py b/Lib/test/test_platform.py index ec134a1..b59f6e6 100644 --- a/Lib/test/test_platform.py +++ b/Lib/test/test_platform.py @@ -56,13 +56,11 @@ class PlatformTest(unittest.TestCase): def setUp(self): self.save_version = sys.version - self.save_subversion = sys.subversion self.save_mercurial = sys._mercurial self.save_platform = sys.platform def tearDown(self): sys.version = self.save_version - sys.subversion = self.save_subversion sys._mercurial = self.save_mercurial sys.platform = self.save_platform @@ -77,7 +75,7 @@ class PlatformTest(unittest.TestCase): ('IronPython', '1.0.0', '', '', '', '', '.NET 2.0.50727.42')), ): # branch and revision are not "parsed", but fetched - # from sys.subversion. Ignore them + # from sys._mercurial. Ignore them (name, version, branch, revision, buildno, builddate, compiler) \ = platform._sys_version(input) self.assertEqual( @@ -113,8 +111,6 @@ class PlatformTest(unittest.TestCase): if subversion is None: if hasattr(sys, "_mercurial"): del sys._mercurial - if hasattr(sys, "subversion"): - del sys.subversion else: sys._mercurial = subversion if sys_platform is not None: @@ -247,6 +243,34 @@ class PlatformTest(unittest.TestCase): ): self.assertEqual(platform._parse_release_file(input), output) + def test_popen(self): + mswindows = (sys.platform == "win32") + + if mswindows: + command = '"{}" -c "print(\'Hello\')"'.format(sys.executable) + else: + command = "'{}' -c 'print(\"Hello\")'".format(sys.executable) + with platform.popen(command) as stdout: + hello = stdout.read().strip() + stdout.close() + self.assertEqual(hello, "Hello") + + data = 'plop' + if mswindows: + command = '"{}" -c "import sys; data=sys.stdin.read(); exit(len(data))"' + else: + command = "'{}' -c 'import sys; data=sys.stdin.read(); exit(len(data))'" + command = command.format(sys.executable) + with platform.popen(command, 'w') as stdin: + stdout = stdin.write(data) + ret = stdin.close() + self.assertIsNotNone(ret) + if os.name == 'nt': + returncode = ret + else: + returncode = ret >> 8 + self.assertEqual(returncode, len(data)) + def test_main(): support.run_unittest( diff --git a/Lib/test/test_poplib.py b/Lib/test/test_poplib.py index 81af569..0a3adcc 100644 --- a/Lib/test/test_poplib.py +++ b/Lib/test/test_poplib.py @@ -108,6 +108,10 @@ class DummyPOP3Handler(asynchat.async_chat): def cmd_apop(self, arg): self.push('+OK done nothing.') + def cmd_quit(self, arg): + self.push('+OK closing.') + self.close_when_done() + class DummyPOP3Server(asyncore.dispatcher, threading.Thread): @@ -165,10 +169,10 @@ class TestPOP3Class(TestCase): def setUp(self): self.server = DummyPOP3Server((HOST, PORT)) self.server.start() - self.client = poplib.POP3(self.server.host, self.server.port) + self.client = poplib.POP3(self.server.host, self.server.port, timeout=3) def tearDown(self): - self.client.quit() + self.client.close() self.server.stop() def test_getwelcome(self): @@ -228,6 +232,12 @@ class TestPOP3Class(TestCase): self.client.uidl() self.client.uidl('foo') + def test_quit(self): + resp = self.client.quit() + self.assertTrue(resp) + self.assertIsNone(self.client.sock) + self.assertIsNone(self.client.file) + SUPPORTS_SSL = False if hasattr(poplib, 'POP3_SSL'): @@ -274,6 +284,7 @@ if hasattr(poplib, 'POP3_SSL'): else: DummyPOP3Handler.handle_read(self) + class TestPOP3_SSLClass(TestPOP3Class): # repeat previous tests by using poplib.POP3_SSL diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py index 45b3afc..72bc1de 100644 --- a/Lib/test/test_posix.py +++ b/Lib/test/test_posix.py @@ -37,7 +37,7 @@ class PosixTester(unittest.TestCase): NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdb", "uname", "times", "getloadavg", "getegid", "geteuid", "getgid", "getgroups", - "getpid", "getpgrp", "getppid", "getuid", + "getpid", "getpgrp", "getppid", "getuid", "sync", ] for name in NO_ARG_FUNCTIONS: @@ -132,6 +132,156 @@ class PosixTester(unittest.TestCase): finally: fp.close() + @unittest.skipUnless(hasattr(posix, 'truncate'), "test needs posix.truncate()") + def test_truncate(self): + with open(support.TESTFN, 'w') as fp: + fp.write('test') + fp.flush() + posix.truncate(support.TESTFN, 0) + + @unittest.skipUnless(hasattr(posix, 'fexecve'), "test needs posix.fexecve()") + @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") + @unittest.skipUnless(hasattr(os, 'wait'), "test needs os.wait()") + def test_fexecve(self): + fp = os.open(sys.executable, os.O_RDONLY) + try: + pid = os.fork() + if pid == 0: + os.chdir(os.path.split(sys.executable)[0]) + posix.fexecve(fp, [sys.executable, '-c', 'pass'], os.environ) + else: + self.assertEqual(os.wait(), (pid, 0)) + finally: + os.close(fp) + + @unittest.skipUnless(hasattr(posix, 'waitid'), "test needs posix.waitid()") + @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") + def test_waitid(self): + pid = os.fork() + if pid == 0: + os.chdir(os.path.split(sys.executable)[0]) + posix.execve(sys.executable, [sys.executable, '-c', 'pass'], os.environ) + else: + res = posix.waitid(posix.P_PID, pid, posix.WEXITED) + self.assertEqual(pid, res.si_pid) + + @unittest.skipUnless(hasattr(posix, 'lockf'), "test needs posix.lockf()") + def test_lockf(self): + fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) + try: + os.write(fd, b'test') + os.lseek(fd, 0, os.SEEK_SET) + posix.lockf(fd, posix.F_LOCK, 4) + # section is locked + posix.lockf(fd, posix.F_ULOCK, 4) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'pread'), "test needs posix.pread()") + def test_pread(self): + fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) + try: + os.write(fd, b'test') + os.lseek(fd, 0, os.SEEK_SET) + self.assertEqual(b'es', posix.pread(fd, 2, 1)) + # the first pread() shoudn't disturb the file offset + self.assertEqual(b'te', posix.read(fd, 2)) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'pwrite'), "test needs posix.pwrite()") + def test_pwrite(self): + fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) + try: + os.write(fd, b'test') + os.lseek(fd, 0, os.SEEK_SET) + posix.pwrite(fd, b'xx', 1) + self.assertEqual(b'txxt', posix.read(fd, 4)) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'posix_fallocate'), + "test needs posix.posix_fallocate()") + def test_posix_fallocate(self): + fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) + try: + posix.posix_fallocate(fd, 0, 10) + except OSError as inst: + # issue10812, ZFS doesn't appear to support posix_fallocate, + # so skip Solaris-based since they are likely to have ZFS. + if inst.errno != errno.EINVAL or not sys.platform.startswith("sunos"): + raise + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'posix_fadvise'), + "test needs posix.posix_fadvise()") + def test_posix_fadvise(self): + fd = os.open(support.TESTFN, os.O_RDONLY) + try: + posix.posix_fadvise(fd, 0, 0, posix.POSIX_FADV_WILLNEED) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'futimes'), "test needs posix.futimes()") + def test_futimes(self): + now = time.time() + fd = os.open(support.TESTFN, os.O_RDONLY) + try: + posix.futimes(fd, None) + self.assertRaises(TypeError, posix.futimes, fd, (None, None)) + self.assertRaises(TypeError, posix.futimes, fd, (now, None)) + self.assertRaises(TypeError, posix.futimes, fd, (None, now)) + posix.futimes(fd, (int(now), int(now))) + posix.futimes(fd, (now, now)) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'lutimes'), "test needs posix.lutimes()") + def test_lutimes(self): + now = time.time() + posix.lutimes(support.TESTFN, None) + self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (None, None)) + self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (now, None)) + self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (None, now)) + posix.lutimes(support.TESTFN, (int(now), int(now))) + posix.lutimes(support.TESTFN, (now, now)) + + @unittest.skipUnless(hasattr(posix, 'futimens'), "test needs posix.futimens()") + def test_futimens(self): + now = time.time() + fd = os.open(support.TESTFN, os.O_RDONLY) + try: + self.assertRaises(TypeError, posix.futimens, fd, (None, None), (None, None)) + self.assertRaises(TypeError, posix.futimens, fd, (now, 0), None) + self.assertRaises(TypeError, posix.futimens, fd, None, (now, 0)) + posix.futimens(fd, (int(now), int((now - int(now)) * 1e9)), + (int(now), int((now - int(now)) * 1e9))) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'writev'), "test needs posix.writev()") + def test_writev(self): + fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) + try: + os.writev(fd, (b'test1', b'tt2', b't3')) + os.lseek(fd, 0, os.SEEK_SET) + self.assertEqual(b'test1tt2t3', posix.read(fd, 10)) + finally: + os.close(fd) + + @unittest.skipUnless(hasattr(posix, 'readv'), "test needs posix.readv()") + def test_readv(self): + fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) + try: + os.write(fd, b'test1tt2t3') + os.lseek(fd, 0, os.SEEK_SET) + buf = [bytearray(i) for i in [5, 3, 2]] + self.assertEqual(posix.readv(fd, buf), 10) + self.assertEqual([b'test1', b'tt2', b't3'], [bytes(i) for i in buf]) + finally: + os.close(fd) + def test_dup(self): if hasattr(posix, 'dup'): fp = open(support.TESTFN) @@ -285,6 +435,18 @@ class PosixTester(unittest.TestCase): if hasattr(posix, 'listdir'): self.assertTrue(support.TESTFN in posix.listdir()) + @unittest.skipUnless(hasattr(posix, 'fdlistdir'), "test needs posix.fdlistdir()") + def test_fdlistdir(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + self.assertEqual( + sorted(posix.listdir('.')), + sorted(posix.fdlistdir(f)) + ) + # Check the fd was closed by fdlistdir + with self.assertRaises(OSError) as ctx: + posix.close(f) + self.assertEqual(ctx.exception.errno, errno.EBADF) + def test_access(self): if hasattr(posix, 'access'): self.assertTrue(posix.access(support.TESTFN, os.R_OK)) @@ -389,6 +551,198 @@ class PosixTester(unittest.TestCase): set([int(x) for x in groups.split()]), set(posix.getgroups() + [posix.getegid()])) + # tests for the posix *at functions follow + + @unittest.skipUnless(hasattr(posix, 'faccessat'), "test needs posix.faccessat()") + def test_faccessat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + self.assertTrue(posix.faccessat(f, support.TESTFN, os.R_OK)) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'fchmodat'), "test needs posix.fchmodat()") + def test_fchmodat(self): + os.chmod(support.TESTFN, stat.S_IRUSR) + + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.fchmodat(f, support.TESTFN, stat.S_IRUSR | stat.S_IWUSR) + + s = posix.stat(support.TESTFN) + self.assertEqual(s[0] & stat.S_IRWXU, stat.S_IRUSR | stat.S_IWUSR) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'fchownat'), "test needs posix.fchownat()") + def test_fchownat(self): + support.unlink(support.TESTFN) + open(support.TESTFN, 'w').close() + + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.fchownat(f, support.TESTFN, os.getuid(), os.getgid()) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'fstatat'), "test needs posix.fstatat()") + def test_fstatat(self): + support.unlink(support.TESTFN) + with open(support.TESTFN, 'w') as outfile: + outfile.write("testline\n") + + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + s1 = posix.stat(support.TESTFN) + s2 = posix.fstatat(f, support.TESTFN) + self.assertEqual(s1, s2) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'futimesat'), "test needs posix.futimesat()") + def test_futimesat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + now = time.time() + posix.futimesat(f, support.TESTFN, None) + self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (None, None)) + self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (now, None)) + self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (None, now)) + posix.futimesat(f, support.TESTFN, (int(now), int(now))) + posix.futimesat(f, support.TESTFN, (now, now)) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'linkat'), "test needs posix.linkat()") + def test_linkat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.linkat(f, support.TESTFN, f, support.TESTFN + 'link') + # should have same inodes + self.assertEqual(posix.stat(support.TESTFN)[1], + posix.stat(support.TESTFN + 'link')[1]) + finally: + posix.close(f) + support.unlink(support.TESTFN + 'link') + + @unittest.skipUnless(hasattr(posix, 'mkdirat'), "test needs posix.mkdirat()") + def test_mkdirat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.mkdirat(f, support.TESTFN + 'dir') + posix.stat(support.TESTFN + 'dir') # should not raise exception + finally: + posix.close(f) + support.rmtree(support.TESTFN + 'dir') + + @unittest.skipUnless(hasattr(posix, 'mknodat') and hasattr(stat, 'S_IFIFO'), + "don't have mknodat()/S_IFIFO") + def test_mknodat(self): + # Test using mknodat() to create a FIFO (the only use specified + # by POSIX). + support.unlink(support.TESTFN) + mode = stat.S_IFIFO | stat.S_IRUSR | stat.S_IWUSR + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.mknodat(f, support.TESTFN, mode, 0) + except OSError as e: + # Some old systems don't allow unprivileged users to use + # mknod(), or only support creating device nodes. + self.assertIn(e.errno, (errno.EPERM, errno.EINVAL)) + else: + self.assertTrue(stat.S_ISFIFO(posix.stat(support.TESTFN).st_mode)) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'openat'), "test needs posix.openat()") + def test_openat(self): + support.unlink(support.TESTFN) + with open(support.TESTFN, 'w') as outfile: + outfile.write("testline\n") + a = posix.open(posix.getcwd(), posix.O_RDONLY) + b = posix.openat(a, support.TESTFN, posix.O_RDONLY) + try: + res = posix.read(b, 9).decode(encoding="utf-8") + self.assertEqual("testline\n", res) + finally: + posix.close(a) + posix.close(b) + + @unittest.skipUnless(hasattr(posix, 'readlinkat'), "test needs posix.readlinkat()") + def test_readlinkat(self): + os.symlink(support.TESTFN, support.TESTFN + 'link') + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + self.assertEqual(posix.readlink(support.TESTFN + 'link'), + posix.readlinkat(f, support.TESTFN + 'link')) + finally: + support.unlink(support.TESTFN + 'link') + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'renameat'), "test needs posix.renameat()") + def test_renameat(self): + support.unlink(support.TESTFN) + open(support.TESTFN + 'ren', 'w').close() + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.renameat(f, support.TESTFN + 'ren', f, support.TESTFN) + except: + posix.rename(support.TESTFN + 'ren', support.TESTFN) + raise + else: + posix.stat(support.TESTFN) # should not throw exception + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'symlinkat'), "test needs posix.symlinkat()") + def test_symlinkat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.symlinkat(support.TESTFN, f, support.TESTFN + 'link') + self.assertEqual(posix.readlink(support.TESTFN + 'link'), support.TESTFN) + finally: + posix.close(f) + support.unlink(support.TESTFN + 'link') + + @unittest.skipUnless(hasattr(posix, 'unlinkat'), "test needs posix.unlinkat()") + def test_unlinkat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + open(support.TESTFN + 'del', 'w').close() + posix.stat(support.TESTFN + 'del') # should not throw exception + try: + posix.unlinkat(f, support.TESTFN + 'del') + except: + support.unlink(support.TESTFN + 'del') + raise + else: + self.assertRaises(OSError, posix.stat, support.TESTFN + 'link') + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'utimensat'), "test needs posix.utimensat()") + def test_utimensat(self): + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + now = time.time() + posix.utimensat(f, support.TESTFN, None, None) + self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, (None, None), (None, None)) + self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, (now, 0), None) + self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, None, (now, 0)) + posix.utimensat(f, support.TESTFN, (int(now), int((now - int(now)) * 1e9)), + (int(now), int((now - int(now)) * 1e9))) + finally: + posix.close(f) + + @unittest.skipUnless(hasattr(posix, 'mkfifoat'), "don't have mkfifoat()") + def test_mkfifoat(self): + support.unlink(support.TESTFN) + f = posix.open(posix.getcwd(), posix.O_RDONLY) + try: + posix.mkfifoat(f, support.TESTFN, stat.S_IRUSR | stat.S_IWUSR) + self.assertTrue(stat.S_ISFIFO(posix.stat(support.TESTFN).st_mode)) + finally: + posix.close(f) + class PosixGroupsTester(unittest.TestCase): def setUp(self): diff --git a/Lib/test/test_pulldom.py b/Lib/test/test_pulldom.py new file mode 100644 index 0000000..b81a595 --- /dev/null +++ b/Lib/test/test_pulldom.py @@ -0,0 +1,347 @@ +import io +import unittest +import sys +import xml.sax + +from xml.sax.xmlreader import AttributesImpl +from xml.dom import pulldom + +from test.support import run_unittest, findfile + + +tstfile = findfile("test.xml", subdir="xmltestdata") + +# A handy XML snippet, containing attributes, a namespace prefix, and a +# self-closing tag: +SMALL_SAMPLE = """<?xml version="1.0"?> +<html xmlns="http://www.w3.org/1999/xhtml" xmlns:xdc="http://www.xml.com/books"> +<!-- A comment --> +<title>Introduction to XSL</title> +<hr/> +<p><xdc:author xdc:attrib="prefixed attribute" attrib="other attrib">A. Namespace</xdc:author></p> +</html>""" + + +class PullDOMTestCase(unittest.TestCase): + + def test_parse(self): + """Minimal test of DOMEventStream.parse()""" + + # This just tests that parsing from a stream works. Actual parser + # semantics are tested using parseString with a more focused XML + # fragment. + + # Test with a filename: + handler = pulldom.parse(tstfile) + self.addCleanup(handler.stream.close) + list(handler) + + # Test with a file object: + with open(tstfile, "rb") as fin: + list(pulldom.parse(fin)) + + def test_parse_semantics(self): + """Test DOMEventStream parsing semantics.""" + + items = pulldom.parseString(SMALL_SAMPLE) + evt, node = next(items) + # Just check the node is a Document: + self.assertTrue(hasattr(node, "createElement")) + self.assertEqual(pulldom.START_DOCUMENT, evt) + evt, node = next(items) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("html", node.tagName) + self.assertEqual(2, len(node.attributes)) + self.assertEqual(node.attributes.getNamedItem("xmlns:xdc").value, + "http://www.xml.com/books") + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) # Line break + evt, node = next(items) + # XXX - A comment should be reported here! + # self.assertEqual(pulldom.COMMENT, evt) + # Line break after swallowed comment: + self.assertEqual(pulldom.CHARACTERS, evt) + evt, node = next(items) + self.assertEqual("title", node.tagName) + title_node = node + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) + self.assertEqual("Introduction to XSL", node.data) + evt, node = next(items) + self.assertEqual(pulldom.END_ELEMENT, evt) + self.assertEqual("title", node.tagName) + self.assertTrue(title_node is node) + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) + evt, node = next(items) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("hr", node.tagName) + evt, node = next(items) + self.assertEqual(pulldom.END_ELEMENT, evt) + self.assertEqual("hr", node.tagName) + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) + evt, node = next(items) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("p", node.tagName) + evt, node = next(items) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("xdc:author", node.tagName) + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) + evt, node = next(items) + self.assertEqual(pulldom.END_ELEMENT, evt) + self.assertEqual("xdc:author", node.tagName) + evt, node = next(items) + self.assertEqual(pulldom.END_ELEMENT, evt) + evt, node = next(items) + self.assertEqual(pulldom.CHARACTERS, evt) + evt, node = next(items) + self.assertEqual(pulldom.END_ELEMENT, evt) + # XXX No END_DOCUMENT item is ever obtained: + #evt, node = next(items) + #self.assertEqual(pulldom.END_DOCUMENT, evt) + + def test_expandItem(self): + """Ensure expandItem works as expected.""" + items = pulldom.parseString(SMALL_SAMPLE) + # Loop through the nodes until we get to a "title" start tag: + for evt, item in items: + if evt == pulldom.START_ELEMENT and item.tagName == "title": + items.expandNode(item) + self.assertEqual(1, len(item.childNodes)) + break + else: + self.fail("No \"title\" element detected in SMALL_SAMPLE!") + # Loop until we get to the next start-element: + for evt, node in items: + if evt == pulldom.START_ELEMENT: + break + self.assertEqual("hr", node.tagName, + "expandNode did not leave DOMEventStream in the correct state.") + # Attempt to expand a standalone element: + items.expandNode(node) + self.assertEqual(next(items)[0], pulldom.CHARACTERS) + evt, node = next(items) + self.assertEqual(node.tagName, "p") + items.expandNode(node) + next(items) # Skip character data + evt, node = next(items) + self.assertEqual(node.tagName, "html") + with self.assertRaises(StopIteration): + next(items) + items.clear() + self.assertIsNone(items.parser) + self.assertIsNone(items.stream) + + @unittest.expectedFailure + def test_comment(self): + """PullDOM does not receive "comment" events.""" + items = pulldom.parseString(SMALL_SAMPLE) + for evt, _ in items: + if evt == pulldom.COMMENT: + break + else: + self.fail("No comment was encountered") + + @unittest.expectedFailure + def test_end_document(self): + """PullDOM does not receive "end-document" events.""" + items = pulldom.parseString(SMALL_SAMPLE) + # Read all of the nodes up to and including </html>: + for evt, node in items: + if evt == pulldom.END_ELEMENT and node.tagName == "html": + break + try: + # Assert that the next node is END_DOCUMENT: + evt, node = next(items) + self.assertEqual(pulldom.END_DOCUMENT, evt) + except StopIteration: + self.fail( + "Ran out of events, but should have received END_DOCUMENT") + + +class ThoroughTestCase(unittest.TestCase): + """Test the hard-to-reach parts of pulldom.""" + + def test_thorough_parse(self): + """Test some of the hard-to-reach parts of PullDOM.""" + self._test_thorough(pulldom.parse(None, parser=SAXExerciser())) + + @unittest.expectedFailure + def test_sax2dom_fail(self): + """SAX2DOM can"t handle a PI before the root element.""" + pd = SAX2DOMTestHelper(None, SAXExerciser(), 12) + self._test_thorough(pd) + + def test_thorough_sax2dom(self): + """Test some of the hard-to-reach parts of SAX2DOM.""" + pd = SAX2DOMTestHelper(None, SAX2DOMExerciser(), 12) + self._test_thorough(pd, False) + + def _test_thorough(self, pd, before_root=True): + """Test some of the hard-to-reach parts of the parser, using a mock + parser.""" + + evt, node = next(pd) + self.assertEqual(pulldom.START_DOCUMENT, evt) + # Just check the node is a Document: + self.assertTrue(hasattr(node, "createElement")) + + if before_root: + evt, node = next(pd) + self.assertEqual(pulldom.COMMENT, evt) + self.assertEqual("a comment", node.data) + evt, node = next(pd) + self.assertEqual(pulldom.PROCESSING_INSTRUCTION, evt) + self.assertEqual("target", node.target) + self.assertEqual("data", node.data) + + evt, node = next(pd) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("html", node.tagName) + + evt, node = next(pd) + self.assertEqual(pulldom.COMMENT, evt) + self.assertEqual("a comment", node.data) + evt, node = next(pd) + self.assertEqual(pulldom.PROCESSING_INSTRUCTION, evt) + self.assertEqual("target", node.target) + self.assertEqual("data", node.data) + + evt, node = next(pd) + self.assertEqual(pulldom.START_ELEMENT, evt) + self.assertEqual("p", node.tagName) + + evt, node = next(pd) + self.assertEqual(pulldom.CHARACTERS, evt) + self.assertEqual("text", node.data) + evt, node = next(pd) + self.assertEqual(pulldom.END_ELEMENT, evt) + self.assertEqual("p", node.tagName) + evt, node = next(pd) + self.assertEqual(pulldom.END_ELEMENT, evt) + self.assertEqual("html", node.tagName) + evt, node = next(pd) + self.assertEqual(pulldom.END_DOCUMENT, evt) + + +class SAXExerciser(object): + """A fake sax parser that calls some of the harder-to-reach sax methods to + ensure it emits the correct events""" + + def setContentHandler(self, handler): + self._handler = handler + + def parse(self, _): + h = self._handler + h.startDocument() + + # The next two items ensure that items preceding the first + # start_element are properly stored and emitted: + h.comment("a comment") + h.processingInstruction("target", "data") + + h.startElement("html", AttributesImpl({})) + + h.comment("a comment") + h.processingInstruction("target", "data") + + h.startElement("p", AttributesImpl({"class": "paraclass"})) + h.characters("text") + h.endElement("p") + h.endElement("html") + h.endDocument() + + def stub(self, *args, **kwargs): + """Stub method. Does nothing.""" + pass + setProperty = stub + setFeature = stub + + +class SAX2DOMExerciser(SAXExerciser): + """The same as SAXExerciser, but without the processing instruction and + comment before the root element, because S2D can"t handle it""" + + def parse(self, _): + h = self._handler + h.startDocument() + h.startElement("html", AttributesImpl({})) + h.comment("a comment") + h.processingInstruction("target", "data") + h.startElement("p", AttributesImpl({"class": "paraclass"})) + h.characters("text") + h.endElement("p") + h.endElement("html") + h.endDocument() + + +class SAX2DOMTestHelper(pulldom.DOMEventStream): + """Allows us to drive SAX2DOM from a DOMEventStream.""" + + def reset(self): + self.pulldom = pulldom.SAX2DOM() + # This content handler relies on namespace support + self.parser.setFeature(xml.sax.handler.feature_namespaces, 1) + self.parser.setContentHandler(self.pulldom) + + +class SAX2DOMTestCase(unittest.TestCase): + + def confirm(self, test, testname="Test"): + self.assertTrue(test, testname) + + def test_basic(self): + """Ensure SAX2DOM can parse from a stream.""" + with io.StringIO(SMALL_SAMPLE) as fin: + sd = SAX2DOMTestHelper(fin, xml.sax.make_parser(), + len(SMALL_SAMPLE)) + for evt, node in sd: + if evt == pulldom.START_ELEMENT and node.tagName == "html": + break + # Because the buffer is the same length as the XML, all the + # nodes should have been parsed and added: + self.assertGreater(len(node.childNodes), 0) + + def testSAX2DOM(self): + """Ensure SAX2DOM expands nodes as expected.""" + sax2dom = pulldom.SAX2DOM() + sax2dom.startDocument() + sax2dom.startElement("doc", {}) + sax2dom.characters("text") + sax2dom.startElement("subelm", {}) + sax2dom.characters("text") + sax2dom.endElement("subelm") + sax2dom.characters("text") + sax2dom.endElement("doc") + sax2dom.endDocument() + + doc = sax2dom.document + root = doc.documentElement + (text1, elm1, text2) = root.childNodes + text3 = elm1.childNodes[0] + + self.assertIsNone(text1.previousSibling) + self.assertIs(text1.nextSibling, elm1) + self.assertIs(elm1.previousSibling, text1) + self.assertIs(elm1.nextSibling, text2) + self.assertIs(text2.previousSibling, elm1) + self.assertIsNone(text2.nextSibling) + self.assertIsNone(text3.previousSibling) + self.assertIsNone(text3.nextSibling) + + self.assertIs(root.parentNode, doc) + self.assertIs(text1.parentNode, root) + self.assertIs(elm1.parentNode, root) + self.assertIs(text2.parentNode, root) + self.assertIs(text3.parentNode, elm1) + doc.unlink() + + +def test_main(): + run_unittest(PullDOMTestCase, ThoroughTestCase, SAX2DOMTestCase) + + +if __name__ == "__main__": + test_main() diff --git a/Lib/test/test_pydoc.py b/Lib/test/test_pydoc.py index 1d575cd..fc4146c 100644 --- a/Lib/test/test_pydoc.py +++ b/Lib/test/test_pydoc.py @@ -248,6 +248,8 @@ class PydocDocTest(unittest.TestCase): @unittest.skipIf(sys.flags.optimize >= 2, "Docstrings are omitted with -O2 and above") + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __locals__ unexpectedly') def test_html_doc(self): result, doc_loc = get_pydoc_html(pydoc_mod) mod_file = inspect.getabsfile(pydoc_mod) @@ -263,6 +265,8 @@ class PydocDocTest(unittest.TestCase): @unittest.skipIf(sys.flags.optimize >= 2, "Docstrings are omitted with -O2 and above") + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __locals__ unexpectedly') def test_text_doc(self): result, doc_loc = get_pydoc_text(pydoc_mod) expected_text = expected_text_pattern % \ @@ -340,6 +344,8 @@ class PydocDocTest(unittest.TestCase): @unittest.skipIf(sys.flags.optimize >= 2, 'Docstrings are omitted with -O2 and above') + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'trace function introduces __locals__ unexpectedly') def test_help_output_redirect(self): # issue 940286, if output is set in Helper, then all output from # Helper.help should be redirected diff --git a/Lib/test/test_reprlib.py b/Lib/test/test_reprlib.py index b0dc4d7..e476941 100644 --- a/Lib/test/test_reprlib.py +++ b/Lib/test/test_reprlib.py @@ -234,7 +234,7 @@ class LongReprTest(unittest.TestCase): touch(os.path.join(self.subpkgname, self.pkgname + '.py')) from areallylongpackageandmodulenametotestreprtruncation.areallylongpackageandmodulenametotestreprtruncation import areallylongpackageandmodulenametotestreprtruncation eq(repr(areallylongpackageandmodulenametotestreprtruncation), - "<module '%s' from '%s'>" % (areallylongpackageandmodulenametotestreprtruncation.__name__, areallylongpackageandmodulenametotestreprtruncation.__file__)) + "<module %r from %r>" % (areallylongpackageandmodulenametotestreprtruncation.__name__, areallylongpackageandmodulenametotestreprtruncation.__file__)) eq(repr(sys), "<module 'sys' (built-in)>") def test_type(self): diff --git a/Lib/test/test_richcmp.py b/Lib/test/test_richcmp.py index f8f3717..0b629dc 100644 --- a/Lib/test/test_richcmp.py +++ b/Lib/test/test_richcmp.py @@ -220,6 +220,7 @@ class MiscTest(unittest.TestCase): for func in (do, operator.not_): self.assertRaises(Exc, func, Bad()) + @support.no_tracing def test_recursion(self): # Check that comparison for recursive objects fails gracefully from collections import UserList diff --git a/Lib/test/test_runpy.py b/Lib/test/test_runpy.py index ad3ab39..00f34b1 100644 --- a/Lib/test/test_runpy.py +++ b/Lib/test/test_runpy.py @@ -6,7 +6,8 @@ import sys import re import tempfile import py_compile -from test.support import forget, make_legacy_pyc, run_unittest, unload, verbose +from test.support import ( + forget, make_legacy_pyc, run_unittest, unload, verbose, no_tracing) from test.script_helper import ( make_pkg, make_script, make_zip_pkg, make_zip_script, temp_dir) @@ -395,6 +396,7 @@ argv0 = sys.argv[0] msg = "can't find '__main__' module in %r" % zip_name self._check_import_error(zip_name, msg) + @no_tracing def test_main_recursion_error(self): with temp_dir() as script_dir, temp_dir() as dummy_dir: mod_name = '__main__' diff --git a/Lib/test/test_sax.py b/Lib/test/test_sax.py index 0f6a1ca..bddb375 100644 --- a/Lib/test/test_sax.py +++ b/Lib/test/test_sax.py @@ -20,8 +20,8 @@ import unittest TEST_XMLFILE = findfile("test.xml", subdir="xmltestdata") TEST_XMLFILE_OUT = findfile("test.xml.out", subdir="xmltestdata") try: - TEST_XMLFILE.encode("utf8") - TEST_XMLFILE_OUT.encode("utf8") + TEST_XMLFILE.encode("utf-8") + TEST_XMLFILE_OUT.encode("utf-8") except UnicodeEncodeError: raise unittest.SkipTest("filename is not encodable to utf8") diff --git a/Lib/test/test_scope.py b/Lib/test/test_scope.py index fbc87aa..129a18a 100644 --- a/Lib/test/test_scope.py +++ b/Lib/test/test_scope.py @@ -1,5 +1,5 @@ import unittest -from test.support import check_syntax_error, run_unittest +from test.support import check_syntax_error, cpython_only, run_unittest class ScopeTests(unittest.TestCase): @@ -496,23 +496,22 @@ class ScopeTests(unittest.TestCase): self.assertNotIn("x", varnames) self.assertIn("y", varnames) + @cpython_only def testLocalsClass_WithTrace(self): # Issue23728: after the trace function returns, the locals() # dictionary is used to update all variables, this used to # include free variables. But in class statements, free # variables are not inserted... import sys + self.addCleanup(sys.settrace, sys.gettrace()) sys.settrace(lambda a,b,c:None) - try: - x = 12 + x = 12 - class C: - def f(self): - return x + class C: + def f(self): + return x - self.assertEqual(x, 12) # Used to raise UnboundLocalError - finally: - sys.settrace(None) + self.assertEqual(x, 12) # Used to raise UnboundLocalError def testBoundAndFree(self): # var is bound and free in class @@ -527,6 +526,7 @@ class ScopeTests(unittest.TestCase): inst = f(3)() self.assertEqual(inst.a, inst.m()) + @cpython_only def testInteractionWithTraceFunc(self): import sys @@ -543,6 +543,7 @@ class ScopeTests(unittest.TestCase): class TestClass: pass + self.addCleanup(sys.settrace, sys.gettrace()) sys.settrace(tracer) adaptgetter("foo", TestClass, (1, "")) sys.settrace(None) diff --git a/Lib/test/test_shelve.py b/Lib/test/test_shelve.py index 3e73f52..13c1265 100644 --- a/Lib/test/test_shelve.py +++ b/Lib/test/test_shelve.py @@ -2,7 +2,7 @@ import unittest import shelve import glob from test import support -from collections import MutableMapping +from collections.abc import MutableMapping from test.test_dbm import dbm_iterator def L1(s): @@ -129,8 +129,8 @@ class TestCase(unittest.TestCase): shelve.Shelf(d)[key] = [1] self.assertIn(key.encode('utf-8'), d) # but a different one can be given - shelve.Shelf(d, keyencoding='latin1')[key] = [1] - self.assertIn(key.encode('latin1'), d) + shelve.Shelf(d, keyencoding='latin-1')[key] = [1] + self.assertIn(key.encode('latin-1'), d) # with all consequences s = shelve.Shelf(d, keyencoding='ascii') self.assertRaises(UnicodeEncodeError, s.__setitem__, key, [1]) diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py index 4651f37..d973faa 100644 --- a/Lib/test/test_smtplib.py +++ b/Lib/test/test_smtplib.py @@ -424,6 +424,9 @@ sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'], # Simulated SMTP channel & server class SimSMTPChannel(smtpd.SMTPChannel): + # For testing failures in QUIT when using the context manager API. + quit_response = None + def __init__(self, extra_features, *args, **kw): self._extrafeatures = ''.join( [ "250-{0}\r\n".format(x) for x in extra_features ]) @@ -475,19 +478,31 @@ class SimSMTPChannel(smtpd.SMTPChannel): else: self.push('550 No access for you!') + def smtp_QUIT(self, arg): + # args is ignored + if self.quit_response is None: + super(SimSMTPChannel, self).smtp_QUIT(arg) + else: + self.push(self.quit_response) + self.close_when_done() + def handle_error(self): raise class SimSMTPServer(smtpd.SMTPServer): + # For testing failures in QUIT when using the context manager API. + quit_response = None + def __init__(self, *args, **kw): self._extra_features = [] smtpd.SMTPServer.__init__(self, *args, **kw) def handle_accepted(self, conn, addr): - self._SMTPchannel = SimSMTPChannel(self._extra_features, - self, conn, addr) + self._SMTPchannel = SimSMTPChannel( + self._extra_features, self, conn, addr) + self._SMTPchannel.quit_response = self.quit_response def process_message(self, peer, mailfrom, rcpttos, data): pass @@ -620,6 +635,25 @@ class SMTPSimTests(unittest.TestCase): self.assertIn(sim_auth_credentials['cram-md5'], str(err)) smtp.close() + def test_with_statement(self): + with smtplib.SMTP(HOST, self.port) as smtp: + code, message = smtp.noop() + self.assertEqual(code, 250) + self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo') + with smtplib.SMTP(HOST, self.port) as smtp: + smtp.close() + self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo') + + def test_with_statement_QUIT_failure(self): + self.serv.quit_response = '421 QUIT FAILED' + with self.assertRaises(smtplib.SMTPResponseException) as error: + with smtplib.SMTP(HOST, self.port) as smtp: + smtp.noop() + self.assertEqual(error.exception.smtp_code, 421) + self.assertEqual(error.exception.smtp_error, b'QUIT FAILED') + # We don't need to clean up self.serv.quit_response because a new + # server is always instantiated in the setUp(). + #TODO: add tests for correct AUTH method fallback now that the #test infrastructure can support it. diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index 9ba391e..d761a73 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -44,7 +44,7 @@ def linux_version(): return 0, 0, 0 HOST = support.HOST -MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf8') ## test unicode string and carriage return +MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8') ## test unicode string and carriage return SUPPORTS_IPV6 = socket.has_ipv6 and try_address('::1', family=socket.AF_INET6) try: @@ -325,6 +325,26 @@ class GeneralModuleTests(unittest.TestCase): if not fqhn in all_host_names: self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names))) + @unittest.skipUnless(hasattr(socket, 'sethostname'), "test needs socket.sethostname()") + @unittest.skipUnless(hasattr(socket, 'gethostname'), "test needs socket.gethostname()") + def test_sethostname(self): + oldhn = socket.gethostname() + try: + socket.sethostname('new') + except socket.error as e: + if e.errno == errno.EPERM: + self.skipTest("test should be run as root") + else: + raise + try: + # running test as root! + self.assertEqual(socket.gethostname(), 'new') + # Should work with bytes objects too + socket.sethostname(b'bar') + self.assertEqual(socket.gethostname(), 'bar') + finally: + socket.sethostname(oldhn) + def testRefCountGetNameInfo(self): # Testing reference count for getnameinfo if hasattr(sys, "getrefcount"): @@ -1065,7 +1085,7 @@ class FileObjectClassTestCase(SocketConnectedTest): """ bufsize = -1 # Use default buffer size - encoding = 'utf8' + encoding = 'utf-8' errors = 'strict' newline = None @@ -1286,7 +1306,7 @@ class FileObjectInterruptedTestCase(unittest.TestCase): data = b'' else: data = '' - expecting = expecting.decode('utf8') + expecting = expecting.decode('utf-8') while len(data) != len(expecting): part = fo.read(size) if not part: @@ -1444,7 +1464,7 @@ class UnicodeReadFileObjectClassTestCase(FileObjectClassTestCase): """Tests for socket.makefile() in text mode (rather than binary)""" read_mode = 'r' - read_msg = MSG.decode('utf8') + read_msg = MSG.decode('utf-8') write_mode = 'wb' write_msg = MSG newline = '' @@ -1456,7 +1476,7 @@ class UnicodeWriteFileObjectClassTestCase(FileObjectClassTestCase): read_mode = 'rb' read_msg = MSG write_mode = 'w' - write_msg = MSG.decode('utf8') + write_msg = MSG.decode('utf-8') newline = '' @@ -1464,9 +1484,9 @@ class UnicodeReadWriteFileObjectClassTestCase(FileObjectClassTestCase): """Tests for socket.makefile() in text mode (rather than binary)""" read_mode = 'r' - read_msg = MSG.decode('utf8') + read_msg = MSG.decode('utf-8') write_mode = 'w' - write_msg = MSG.decode('utf8') + write_msg = MSG.decode('utf-8') newline = '' diff --git a/Lib/test/test_strlit.py b/Lib/test/test_strlit.py index 9eb30e9..fb9cdbb 100644 --- a/Lib/test/test_strlit.py +++ b/Lib/test/test_strlit.py @@ -130,7 +130,7 @@ class TestLiterals(unittest.TestCase): self.assertRaises(SyntaxError, self.check_encoding, "utf-8", extra) def test_file_utf8(self): - self.check_encoding("utf8") + self.check_encoding("utf-8") def test_file_iso_8859_1(self): self.check_encoding("iso-8859-1") diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 01e670e..6523c8a 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -58,6 +58,8 @@ class BaseTestCase(unittest.TestCase): # shutdown time. That frustrates tests trying to check stderr produced # from a spawned Python process. actual = support.strip_python_stderr(stderr) + # strip_python_stderr also strips whitespace, so we do too. + expected = expected.strip() self.assertEqual(actual, expected, msg) @@ -69,6 +71,15 @@ class ProcessTestCase(BaseTestCase): "import sys; sys.exit(47)"]) self.assertEqual(rc, 47) + def test_call_timeout(self): + # call() function with timeout argument; we want to test that the child + # process gets killed when the timeout expires. If the child isn't + # killed, this call will deadlock since subprocess.call waits for the + # child. + self.assertRaises(subprocess.TimeoutExpired, subprocess.call, + [sys.executable, "-c", "while True: pass"], + timeout=0.1) + def test_check_call_zero(self): # check_call() function with zero return code rc = subprocess.check_call([sys.executable, "-c", @@ -111,6 +122,20 @@ class ProcessTestCase(BaseTestCase): self.fail("Expected ValueError when stdout arg supplied.") self.assertIn('stdout', c.exception.args[0]) + def test_check_output_timeout(self): + # check_output() function with timeout arg + with self.assertRaises(subprocess.TimeoutExpired) as c: + output = subprocess.check_output( + [sys.executable, "-c", + "import sys; sys.stdout.write('BDFL')\n" + "sys.stdout.flush()\n" + "while True: pass"], + # Some heavily loaded buildbots (sparc Debian 3.x) require + # this much time to start and print. + timeout=3) + self.fail("Expected TimeoutExpired.") + self.assertEqual(c.exception.output, b'BDFL') + def test_call_kwargs(self): # call() function with keyword args newenv = os.environ.copy() @@ -300,6 +325,31 @@ class ProcessTestCase(BaseTestCase): rc = subprocess.call([sys.executable, "-c", cmd], stdout=1) self.assertEqual(rc, 2) + def test_stdout_devnull(self): + p = subprocess.Popen([sys.executable, "-c", + 'for i in range(10240):' + 'print("x" * 1024)'], + stdout=subprocess.DEVNULL) + p.wait() + self.assertEqual(p.stdout, None) + + def test_stderr_devnull(self): + p = subprocess.Popen([sys.executable, "-c", + 'import sys\n' + 'for i in range(10240):' + 'sys.stderr.write("x" * 1024)'], + stderr=subprocess.DEVNULL) + p.wait() + self.assertEqual(p.stderr, None) + + def test_stdin_devnull(self): + p = subprocess.Popen([sys.executable, "-c", + 'import sys;' + 'sys.stdin.read(1)'], + stdin=subprocess.DEVNULL) + p.wait() + self.assertEqual(p.stdin, None) + def test_cwd(self): tmpdir = tempfile.gettempdir() # We cannot use os.path.realpath to canonicalize the path, @@ -368,6 +418,41 @@ class ProcessTestCase(BaseTestCase): self.assertEqual(stdout, b"banana") self.assertStderrEqual(stderr, b"pineapple") + def test_communicate_timeout(self): + p = subprocess.Popen([sys.executable, "-c", + 'import sys,os,time;' + 'sys.stderr.write("pineapple\\n");' + 'time.sleep(1);' + 'sys.stderr.write("pear\\n");' + 'sys.stdout.write(sys.stdin.read())'], + universal_newlines=True, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana", + timeout=0.3) + # Make sure we can keep waiting for it, and that we get the whole output + # after it completes. + (stdout, stderr) = p.communicate() + self.assertEqual(stdout, "banana") + self.assertStderrEqual(stderr.encode(), b"pineapple\npear\n") + + def test_communicate_timeout_large_ouput(self): + # Test a expring timeout while the child is outputting lots of data. + p = subprocess.Popen([sys.executable, "-c", + 'import sys,os,time;' + 'sys.stdout.write("a" * (64 * 1024));' + 'time.sleep(0.2);' + 'sys.stdout.write("a" * (64 * 1024));' + 'time.sleep(0.2);' + 'sys.stdout.write("a" * (64 * 1024));' + 'time.sleep(0.2);' + 'sys.stdout.write("a" * (64 * 1024));'], + stdout=subprocess.PIPE) + self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4) + (stdout, _) = p.communicate() + self.assertEqual(len(stdout), 4 * 64 * 1024) + # Test for the fd leak reported in http://bugs.python.org/issue2791. def test_communicate_pipe_fd_leak(self): for stdin_pipe in (False, True): @@ -564,6 +649,15 @@ class ProcessTestCase(BaseTestCase): # Subsequent invocations should just return the returncode self.assertEqual(p.wait(), 0) + def test_wait_timeout(self): + p = subprocess.Popen([sys.executable, + "-c", "import time; time.sleep(0.1)"]) + with self.assertRaises(subprocess.TimeoutExpired) as c: + p.wait(timeout=0.01) + self.assertIn("0.01", str(c.exception)) # For coverage of __str__. + # Some heavily loaded buildbots (sparc Debian 3.x) require this much + # time to start. + self.assertEqual(p.wait(timeout=3), 0) def test_invalid_bufsize(self): # an invalid type of the bufsize argument should raise @@ -1064,6 +1158,11 @@ class POSIXProcessTestCase(BaseTestCase): exitcode = subprocess.call([abs_program, "-c", "pass"]) self.assertEqual(exitcode, 0) + # absolute bytes path as a string + cmd = b"'" + abs_program + b"' -c pass" + exitcode = subprocess.call(cmd, shell=True) + self.assertEqual(exitcode, 0) + # bytes program, unicode PATH env = os.environ.copy() env["PATH"] = path @@ -1214,7 +1313,7 @@ class POSIXProcessTestCase(BaseTestCase): stdout, stderr = p.communicate() self.assertEqual(0, p.returncode, "sigchild_ignore.py exited" " non-zero with this error:\n%s" % - stderr.decode('utf8')) + stderr.decode('utf-8')) def test_select_unbuffered(self): # Issue #11459: bufsize=0 should really set the pipes as diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py index db27cdf..d412e67 100644 --- a/Lib/test/test_sys.py +++ b/Lib/test/test_sys.py @@ -303,6 +303,7 @@ class SysModuleTest(unittest.TestCase): self.assertEqual(sys.getdlopenflags(), oldflags+1) sys.setdlopenflags(oldflags) + @test.support.refcount_test def test_refcount(self): # n here must be a global in order for this test to pass while # tracing with a python function. Tracing calls PyFrame_FastToLocals diff --git a/Lib/test/test_sys_settrace.py b/Lib/test/test_sys_settrace.py index 43134e9..acb8e29 100644 --- a/Lib/test/test_sys_settrace.py +++ b/Lib/test/test_sys_settrace.py @@ -251,6 +251,7 @@ class TraceTestCase(unittest.TestCase): def setUp(self): self.using_gc = gc.isenabled() gc.disable() + self.addCleanup(sys.settrace, sys.gettrace()) def tearDown(self): if self.using_gc: @@ -389,6 +390,9 @@ class TraceTestCase(unittest.TestCase): class RaisingTraceFuncTestCase(unittest.TestCase): + def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) + def trace(self, frame, event, arg): """A trace function that raises an exception in response to a specific trace event.""" @@ -688,6 +692,10 @@ def no_jump_without_trace_function(): class JumpTestCase(unittest.TestCase): + def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) + sys.settrace(None) + def compare_jump_output(self, expected, received): if received != expected: self.fail( "Outputs don't match:\n" + @@ -739,6 +747,8 @@ class JumpTestCase(unittest.TestCase): def test_18_no_jump_to_non_integers(self): self.run_test(no_jump_to_non_integers) def test_19_no_jump_without_trace_function(self): + # Must set sys.settrace(None) in setUp(), else condition is not + # triggered. no_jump_without_trace_function() def test_20_large_function(self): diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index 68e094d..a645bf2 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -1289,7 +1289,7 @@ class UstarUnicodeTest(unittest.TestCase): self._test_unicode_filename("utf7") def test_utf8_filename(self): - self._test_unicode_filename("utf8") + self._test_unicode_filename("utf-8") def _test_unicode_filename(self, encoding): tar = tarfile.open(tmpname, "w", format=self.format, encoding=encoding, errors="strict") @@ -1368,7 +1368,7 @@ class GNUUnicodeTest(UstarUnicodeTest): def test_bad_pax_header(self): # Test for issue #8633. GNU tar <= 1.23 creates raw binary fields # without a hdrcharset=BINARY header. - for encoding, name in (("utf8", "pax/bad-pax-\udce4\udcf6\udcfc"), + for encoding, name in (("utf-8", "pax/bad-pax-\udce4\udcf6\udcfc"), ("iso8859-1", "pax/bad-pax-\xe4\xf6\xfc"),): with tarfile.open(tarname, encoding=encoding, errors="surrogateescape") as tar: try: @@ -1383,7 +1383,7 @@ class PAXUnicodeTest(UstarUnicodeTest): def test_binary_header(self): # Test a POSIX.1-2008 compatible header with a hdrcharset=BINARY field. - for encoding, name in (("utf8", "pax/hdrcharset-\udce4\udcf6\udcfc"), + for encoding, name in (("utf-8", "pax/hdrcharset-\udce4\udcf6\udcfc"), ("iso8859-1", "pax/hdrcharset-\xe4\xf6\xfc"),): with tarfile.open(tarname, encoding=encoding, errors="surrogateescape") as tar: try: diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index 5f99b2e..c107652 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -427,6 +427,14 @@ class ThreadTests(BaseTestCase): t.daemon = True self.assertTrue('daemon' in repr(t)) + def test_deamon_param(self): + t = threading.Thread() + self.assertFalse(t.daemon) + t = threading.Thread(daemon=False) + self.assertFalse(t.daemon) + t = threading.Thread(daemon=True) + self.assertTrue(t.daemon) + class ThreadJoinOnShutdown(BaseTestCase): @@ -677,6 +685,10 @@ class ThreadingExceptionTests(BaseTestCase): thread.start() self.assertRaises(RuntimeError, setattr, thread, "daemon", True) + def test_releasing_unacquired_lock(self): + lock = threading.Lock() + self.assertRaises(RuntimeError, lock.release) + class LockTests(lock_tests.LockTests): locktype = staticmethod(threading.Lock) diff --git a/Lib/test/test_trace.py b/Lib/test/test_trace.py index 461d1d8..d9bef38 100644 --- a/Lib/test/test_trace.py +++ b/Lib/test/test_trace.py @@ -102,6 +102,7 @@ class TracedClass(object): class TestLineCounts(unittest.TestCase): """White-box testing of line-counting, via runfunc""" def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) self.tracer = Trace(count=1, trace=0, countfuncs=0, countcallers=0) self.my_py_filename = fix_ext_py(__file__) @@ -192,6 +193,7 @@ class TestRunExecCounts(unittest.TestCase): """A simple sanity test of line-counting, via runctx (exec)""" def setUp(self): self.my_py_filename = fix_ext_py(__file__) + self.addCleanup(sys.settrace, sys.gettrace()) def test_exec_counts(self): self.tracer = Trace(count=1, trace=0, countfuncs=0, countcallers=0) @@ -218,6 +220,7 @@ class TestRunExecCounts(unittest.TestCase): class TestFuncs(unittest.TestCase): """White-box testing of funcs tracing""" def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) self.tracer = Trace(count=0, trace=0, countfuncs=1) self.filemod = my_file_and_modname() @@ -242,6 +245,8 @@ class TestFuncs(unittest.TestCase): } self.assertEqual(self.tracer.results().calledfuncs, expected) + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'pre-existing trace function throws off measurements') def test_inst_method_calling(self): obj = TracedClass(20) self.tracer.runfunc(obj.inst_method_calling, 1) @@ -257,9 +262,12 @@ class TestFuncs(unittest.TestCase): class TestCallers(unittest.TestCase): """White-box testing of callers tracing""" def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) self.tracer = Trace(count=0, trace=0, countcallers=1) self.filemod = my_file_and_modname() + @unittest.skipIf(hasattr(sys, 'gettrace') and sys.gettrace(), + 'pre-existing trace function throws off measurements') def test_loop_caller_importing(self): self.tracer.runfunc(traced_func_importing_caller, 1) @@ -280,6 +288,9 @@ class TestCallers(unittest.TestCase): # Created separately for issue #3821 class TestCoverage(unittest.TestCase): + def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) + def tearDown(self): rmtree(TESTFN) unlink(TESTFN) diff --git a/Lib/test/test_unicode.py b/Lib/test/test_unicode.py index 65b26c5..ace3736 100644 --- a/Lib/test/test_unicode.py +++ b/Lib/test/test_unicode.py @@ -614,7 +614,7 @@ class UnicodeTest(string_tests.CommonTest, self.assertEqual('{0!s}'.format(G('data')), 'string is data') msg = 'object.__format__ with a non-empty format string is deprecated' - with support.check_warnings((msg, PendingDeprecationWarning)): + with support.check_warnings((msg, DeprecationWarning)): self.assertEqual('{0:^10}'.format(E('data')), ' E(data) ') self.assertEqual('{0:^10s}'.format(E('data')), ' E(data) ') self.assertEqual('{0:>15s}'.format(G('data')), ' string is data') @@ -1182,11 +1182,14 @@ class UnicodeTest(string_tests.CommonTest, self.assertEqual('hello'.encode('ascii'), b'hello') self.assertEqual('hello'.encode('utf-7'), b'hello') self.assertEqual('hello'.encode('utf-8'), b'hello') - self.assertEqual('hello'.encode('utf8'), b'hello') + self.assertEqual('hello'.encode('utf-8'), b'hello') self.assertEqual('hello'.encode('utf-16-le'), b'h\000e\000l\000l\000o\000') self.assertEqual('hello'.encode('utf-16-be'), b'\000h\000e\000l\000l\000o') self.assertEqual('hello'.encode('latin-1'), b'hello') + # Default encoding is utf-8 + self.assertEqual('\u2603'.encode(), b'\xe2\x98\x83') + # Roundtrip safety for BMP (just the first 1024 chars) for c in range(1024): u = chr(c) @@ -1427,7 +1430,9 @@ class UnicodeTest(string_tests.CommonTest, # Test PyUnicode_FromFormat() def test_from_format(self): support.import_module('ctypes') - from ctypes import pythonapi, py_object, c_int + from ctypes import (pythonapi, py_object, + c_int, c_long, c_longlong, c_ssize_t, + c_uint, c_ulong, c_ulonglong, c_size_t) if sys.maxunicode == 65535: name = "PyUnicodeUCS2_FromFormat" else: @@ -1452,13 +1457,40 @@ class UnicodeTest(string_tests.CommonTest, 'string, got a non-ASCII byte: 0xe9$', PyUnicode_FromFormat, b'unicode\xe9=%s', 'ascii') + # test "%c" self.assertEqual(PyUnicode_FromFormat(b'%c', c_int(0xabcd)), '\uabcd') self.assertEqual(PyUnicode_FromFormat(b'%c', c_int(0x10ffff)), '\U0010ffff') - # other tests + # test "%" + self.assertEqual(PyUnicode_FromFormat(b'%'), '%') + self.assertEqual(PyUnicode_FromFormat(b'%%'), '%') + self.assertEqual(PyUnicode_FromFormat(b'%%s'), '%s') + self.assertEqual(PyUnicode_FromFormat(b'[%%]'), '[%]') + self.assertEqual(PyUnicode_FromFormat(b'%%%s', b'abc'), '%abc') + + # test integer formats (%i, %d, %u) + self.assertEqual(PyUnicode_FromFormat(b'%03i', c_int(10)), '010') + self.assertEqual(PyUnicode_FromFormat(b'%0.4i', c_int(10)), '0010') + self.assertEqual(PyUnicode_FromFormat(b'%i', c_int(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%li', c_long(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%lli', c_longlong(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%zi', c_ssize_t(-123)), '-123') + + self.assertEqual(PyUnicode_FromFormat(b'%d', c_int(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%ld', c_long(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%lld', c_longlong(-123)), '-123') + self.assertEqual(PyUnicode_FromFormat(b'%zd', c_ssize_t(-123)), '-123') + + self.assertEqual(PyUnicode_FromFormat(b'%u', c_uint(123)), '123') + self.assertEqual(PyUnicode_FromFormat(b'%lu', c_ulong(123)), '123') + self.assertEqual(PyUnicode_FromFormat(b'%llu', c_ulonglong(123)), '123') + self.assertEqual(PyUnicode_FromFormat(b'%zu', c_size_t(123)), '123') + + # test %A text = PyUnicode_FromFormat(b'%%A:%A', 'abc\xe9\uabcd\U0010ffff') self.assertEqual(text, r"%A:'abc\xe9\uabcd\U0010ffff'") + # test %V text = PyUnicode_FromFormat(b'repr=%V', 'abc', b'xyz') self.assertEqual(text, 'repr=abc') @@ -1472,6 +1504,13 @@ class UnicodeTest(string_tests.CommonTest, text = PyUnicode_FromFormat(b'repr=%V', None, b'abc\xff') self.assertEqual(text, 'repr=abc\ufffd') + # not supported: copy the raw format string. these tests are just here + # to check for crashs and should not be considered as specifications + self.assertEqual(PyUnicode_FromFormat(b'%1%s', b'abc'), '%s') + self.assertEqual(PyUnicode_FromFormat(b'%1abc'), '%1abc') + self.assertEqual(PyUnicode_FromFormat(b'%+i', c_int(10)), '%+i') + self.assertEqual(PyUnicode_FromFormat(b'%.%s', b'abc'), '%.%s') + # Test PyUnicode_AsWideChar() def test_aswidechar(self): from _testcapi import unicode_aswidechar diff --git a/Lib/test/test_urllib.py b/Lib/test/test_urllib.py index 074d833..d510e49 100644 --- a/Lib/test/test_urllib.py +++ b/Lib/test/test_urllib.py @@ -256,7 +256,7 @@ class urlretrieve_FileTests(unittest.TestCase): def constructLocalFileUrl(self, filePath): filePath = os.path.abspath(filePath) try: - filePath.encode("utf8") + filePath.encode("utf-8") except UnicodeEncodeError: raise unittest.SkipTest("filePath is not encodable to utf8") return "file://%s" % urllib.request.pathname2url(filePath) diff --git a/Lib/test/test_urllib2.py b/Lib/test/test_urllib2.py index 69bcfa2..1d0d98c 100644 --- a/Lib/test/test_urllib2.py +++ b/Lib/test/test_urllib2.py @@ -601,7 +601,7 @@ class OpenerDirectorTests(unittest.TestCase): def sanepathname2url(path): try: - path.encode("utf8") + path.encode("utf-8") except UnicodeEncodeError: raise unittest.SkipTest("path is not encodable to utf8") urlpath = urllib.request.pathname2url(path) diff --git a/Lib/test/test_uuid.py b/Lib/test/test_uuid.py index 43fa656..7bc59ed 100644 --- a/Lib/test/test_uuid.py +++ b/Lib/test/test_uuid.py @@ -471,14 +471,14 @@ class TestUUID(TestCase): if pid == 0: os.close(fds[0]) value = uuid.uuid4() - os.write(fds[1], value.hex.encode('latin1')) + os.write(fds[1], value.hex.encode('latin-1')) os._exit(0) else: os.close(fds[1]) parent_value = uuid.uuid4().hex os.waitpid(pid, 0) - child_value = os.read(fds[0], 100).decode('latin1') + child_value = os.read(fds[0], 100).decode('latin-1') self.assertNotEqual(parent_value, child_value) diff --git a/Lib/test/test_xml_etree.py b/Lib/test/test_xml_etree.py index 22fafa9..40c2291 100644 --- a/Lib/test/test_xml_etree.py +++ b/Lib/test/test_xml_etree.py @@ -22,7 +22,7 @@ from xml.etree import ElementTree as ET SIMPLE_XMLFILE = findfile("simple.xml", subdir="xmltestdata") try: - SIMPLE_XMLFILE.encode("utf8") + SIMPLE_XMLFILE.encode("utf-8") except UnicodeEncodeError: raise unittest.SkipTest("filename is not encodable to utf8") SIMPLE_NS_XMLFILE = findfile("simple-ns.xml", subdir="xmltestdata") @@ -1255,8 +1255,8 @@ def processinginstruction(): >>> ET.tostring(ET.PI('test', '<testing&>')) b'<?test <testing&>?>' - >>> ET.tostring(ET.PI('test', '<testing&>\xe3'), 'latin1') - b"<?xml version='1.0' encoding='latin1'?>\\n<?test <testing&>\\xe3?>" + >>> ET.tostring(ET.PI('test', '<testing&>\xe3'), 'latin-1') + b"<?xml version='1.0' encoding='latin-1'?>\\n<?test <testing&>\\xe3?>" """ # diff --git a/Lib/test/test_xmlrpc_net.py b/Lib/test/test_xmlrpc_net.py index 5df79f0..b9853ed 100644 --- a/Lib/test/test_xmlrpc_net.py +++ b/Lib/test/test_xmlrpc_net.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -import collections +import collections.abc import errno import socket import sys @@ -48,7 +48,7 @@ class CurrentTimeTest(unittest.TestCase): # Perform a minimal sanity check on the result, just to be sure # the request means what we think it means. - self.assertIsInstance(builders, collections.Sequence) + self.assertIsInstance(builders, collections.abc.Sequence) self.assertTrue([x for x in builders if "3.x" in x], builders) diff --git a/Lib/test/test_zipimport_support.py b/Lib/test/test_zipimport_support.py index 610dfa0..f0e2fbc 100644 --- a/Lib/test/test_zipimport_support.py +++ b/Lib/test/test_zipimport_support.py @@ -163,20 +163,19 @@ class ZipSupportTests(unittest.TestCase): test_zipped_doctest.test_DocTestRunner.verbose_flag, test_zipped_doctest.test_Example, test_zipped_doctest.test_debug, - test_zipped_doctest.test_pdb_set_trace, - test_zipped_doctest.test_pdb_set_trace_nested, test_zipped_doctest.test_testsource, test_zipped_doctest.test_trailing_space_in_test, test_zipped_doctest.test_DocTestSuite, test_zipped_doctest.test_DocTestFinder, ] - # These remaining tests are the ones which need access + # These tests are the ones which need access # to the data files, so we don't run them fail_due_to_missing_data_files = [ test_zipped_doctest.test_DocFileSuite, test_zipped_doctest.test_testfile, test_zipped_doctest.test_unittest_reportflags, ] + for obj in known_good_tests: _run_object_doctest(obj, test_zipped_doctest) finally: |