aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test')
-rw-r--r--Lib/test/support/__init__.py4
-rw-r--r--Lib/test/support/hashlib_helper.py221
-rw-r--r--Lib/test/test_asyncio/test_futures.py58
-rw-r--r--Lib/test/test_concurrent_futures/test_future.py57
-rw-r--r--Lib/test/test_faulthandler.py23
-rw-r--r--Lib/test/test_fractions.py208
-rw-r--r--Lib/test/test_free_threading/test_functools.py75
-rw-r--r--Lib/test/test_genericalias.py6
-rw-r--r--Lib/test/test_hmac.py154
-rw-r--r--Lib/test/test_httpservers.py497
-rw-r--r--Lib/test/test_importlib/import_/test_relative_imports.py15
-rw-r--r--Lib/test/test_interpreters/test_api.py8
-rw-r--r--Lib/test/test_logging.py23
-rw-r--r--Lib/test/test_platform.py9
-rw-r--r--Lib/test/test_string/test_templatelib.py7
-rw-r--r--Lib/test/test_syntax.py17
-rw-r--r--Lib/test/test_sys.py7
-rw-r--r--Lib/test/test_tarfile.py49
-rw-r--r--Lib/test/test_typing.py30
-rw-r--r--Lib/test/test_wave.py26
20 files changed, 872 insertions, 622 deletions
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index c74c3a31909..9b6e80fdad9 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -696,9 +696,11 @@ def sortdict(dict):
return "{%s}" % withcommas
-def run_code(code: str) -> dict[str, object]:
+def run_code(code: str, extra_names: dict[str, object] | None = None) -> dict[str, object]:
"""Run a piece of code after dedenting it, and return its global namespace."""
ns = {}
+ if extra_names:
+ ns.update(extra_names)
exec(textwrap.dedent(code), ns)
return ns
diff --git a/Lib/test/support/hashlib_helper.py b/Lib/test/support/hashlib_helper.py
index 5043f08dd93..7032257b068 100644
--- a/Lib/test/support/hashlib_helper.py
+++ b/Lib/test/support/hashlib_helper.py
@@ -23,6 +23,22 @@ def requires_builtin_hmac():
return unittest.skipIf(_hmac is None, "requires _hmac")
+def _missing_hash(digestname, implementation=None, *, exc=None):
+ parts = ["missing", implementation, f"hash algorithm: {digestname!r}"]
+ msg = " ".join(filter(None, parts))
+ raise unittest.SkipTest(msg) from exc
+
+
+def _openssl_availabillity(digestname, *, usedforsecurity):
+ try:
+ _hashlib.new(digestname, usedforsecurity=usedforsecurity)
+ except AttributeError:
+ assert _hashlib is None
+ _missing_hash(digestname, "OpenSSL")
+ except ValueError as exc:
+ _missing_hash(digestname, "OpenSSL", exc=exc)
+
+
def _decorate_func_or_class(func_or_class, decorator_func):
if not isinstance(func_or_class, type):
return decorator_func(func_or_class)
@@ -71,8 +87,7 @@ def requires_hashdigest(digestname, openssl=None, usedforsecurity=True):
try:
test_availability()
except ValueError as exc:
- msg = f"missing hash algorithm: {digestname!r}"
- raise unittest.SkipTest(msg) from exc
+ _missing_hash(digestname, exc=exc)
return func(*args, **kwargs)
return wrapper
@@ -87,14 +102,44 @@ def requires_openssl_hashdigest(digestname, *, usedforsecurity=True):
The hashing algorithm may be missing or blocked by a strict crypto policy.
"""
def decorator_func(func):
- @requires_hashlib()
+ @requires_hashlib() # avoid checking at each call
@functools.wraps(func)
def wrapper(*args, **kwargs):
+ _openssl_availabillity(digestname, usedforsecurity=usedforsecurity)
+ return func(*args, **kwargs)
+ return wrapper
+
+ def decorator(func_or_class):
+ return _decorate_func_or_class(func_or_class, decorator_func)
+ return decorator
+
+
+def find_openssl_hashdigest_constructor(digestname, *, usedforsecurity=True):
+ """Find the OpenSSL hash function constructor by its name."""
+ assert isinstance(digestname, str), digestname
+ _openssl_availabillity(digestname, usedforsecurity=usedforsecurity)
+ # This returns a function of the form _hashlib.openssl_<name> and
+ # not a lambda function as it is rejected by _hashlib.hmac_new().
+ return getattr(_hashlib, f"openssl_{digestname}")
+
+
+def requires_builtin_hashdigest(
+ module_name, digestname, *, usedforsecurity=True
+):
+ """Decorator raising SkipTest if a HACL* hashing algorithm is missing.
+
+ - The *module_name* is the C extension module name based on HACL*.
+ - The *digestname* is one of its member, e.g., 'md5'.
+ """
+ def decorator_func(func):
+ @functools.wraps(func)
+ def wrapper(*args, **kwargs):
+ module = import_module(module_name)
try:
- _hashlib.new(digestname, usedforsecurity=usedforsecurity)
- except ValueError:
- msg = f"missing OpenSSL hash algorithm: {digestname!r}"
- raise unittest.SkipTest(msg)
+ getattr(module, digestname)
+ except AttributeError:
+ fullname = f'{module_name}.{digestname}'
+ _missing_hash(fullname, implementation="HACL")
return func(*args, **kwargs)
return wrapper
@@ -103,6 +148,168 @@ def requires_openssl_hashdigest(digestname, *, usedforsecurity=True):
return decorator
+def find_builtin_hashdigest_constructor(
+ module_name, digestname, *, usedforsecurity=True
+):
+ """Find the HACL* hash function constructor.
+
+ - The *module_name* is the C extension module name based on HACL*.
+ - The *digestname* is one of its member, e.g., 'md5'.
+ """
+ module = import_module(module_name)
+ try:
+ constructor = getattr(module, digestname)
+ constructor(b'', usedforsecurity=usedforsecurity)
+ except (AttributeError, TypeError, ValueError):
+ _missing_hash(f'{module_name}.{digestname}', implementation="HACL")
+ return constructor
+
+
+class HashFunctionsTrait:
+ """Mixin trait class containing hash functions.
+
+ This class is assumed to have all unitest.TestCase methods but should
+ not directly inherit from it to prevent the test suite being run on it.
+
+ Subclasses should implement the hash functions by returning an object
+ that can be recognized as a valid digestmod parameter for both hashlib
+ and HMAC. In particular, it cannot be a lambda function as it will not
+ be recognized by hashlib (it will still be accepted by the pure Python
+ implementation of HMAC).
+ """
+
+ ALGORITHMS = [
+ 'md5', 'sha1',
+ 'sha224', 'sha256', 'sha384', 'sha512',
+ 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512',
+ ]
+
+ # Default 'usedforsecurity' to use when looking up a hash function.
+ usedforsecurity = True
+
+ def _find_constructor(self, name):
+ # By default, a missing algorithm skips the test that uses it.
+ self.assertIn(name, self.ALGORITHMS)
+ self.skipTest(f"missing hash function: {name}")
+
+ @property
+ def md5(self):
+ return self._find_constructor("md5")
+
+ @property
+ def sha1(self):
+ return self._find_constructor("sha1")
+
+ @property
+ def sha224(self):
+ return self._find_constructor("sha224")
+
+ @property
+ def sha256(self):
+ return self._find_constructor("sha256")
+
+ @property
+ def sha384(self):
+ return self._find_constructor("sha384")
+
+ @property
+ def sha512(self):
+ return self._find_constructor("sha512")
+
+ @property
+ def sha3_224(self):
+ return self._find_constructor("sha3_224")
+
+ @property
+ def sha3_256(self):
+ return self._find_constructor("sha3_256")
+
+ @property
+ def sha3_384(self):
+ return self._find_constructor("sha3_384")
+
+ @property
+ def sha3_512(self):
+ return self._find_constructor("sha3_512")
+
+
+class NamedHashFunctionsTrait(HashFunctionsTrait):
+ """Trait containing named hash functions.
+
+ Hash functions are available if and only if they are available in hashlib.
+ """
+
+ def _find_constructor(self, name):
+ self.assertIn(name, self.ALGORITHMS)
+ return name
+
+
+class OpenSSLHashFunctionsTrait(HashFunctionsTrait):
+ """Trait containing OpenSSL hash functions.
+
+ Hash functions are available if and only if they are available in _hashlib.
+ """
+
+ def _find_constructor(self, name):
+ self.assertIn(name, self.ALGORITHMS)
+ return find_openssl_hashdigest_constructor(
+ name, usedforsecurity=self.usedforsecurity
+ )
+
+
+class BuiltinHashFunctionsTrait(HashFunctionsTrait):
+ """Trait containing HACL* hash functions.
+
+ Hash functions are available if and only if they are available in C.
+ In particular, HACL* HMAC-MD5 may be available even though HACL* md5
+ is not since the former is unconditionally built.
+ """
+
+ def _find_constructor_in(self, module, name):
+ self.assertIn(name, self.ALGORITHMS)
+ return find_builtin_hashdigest_constructor(module, name)
+
+ @property
+ def md5(self):
+ return self._find_constructor_in("_md5", "md5")
+
+ @property
+ def sha1(self):
+ return self._find_constructor_in("_sha1", "sha1")
+
+ @property
+ def sha224(self):
+ return self._find_constructor_in("_sha2", "sha224")
+
+ @property
+ def sha256(self):
+ return self._find_constructor_in("_sha2", "sha256")
+
+ @property
+ def sha384(self):
+ return self._find_constructor_in("_sha2", "sha384")
+
+ @property
+ def sha512(self):
+ return self._find_constructor_in("_sha2", "sha512")
+
+ @property
+ def sha3_224(self):
+ return self._find_constructor_in("_sha3", "sha3_224")
+
+ @property
+ def sha3_256(self):
+ return self._find_constructor_in("_sha3","sha3_256")
+
+ @property
+ def sha3_384(self):
+ return self._find_constructor_in("_sha3","sha3_384")
+
+ @property
+ def sha3_512(self):
+ return self._find_constructor_in("_sha3","sha3_512")
+
+
def find_gil_minsize(modules_names, default=2048):
"""Get the largest GIL_MINSIZE value for the given cryptographic modules.
diff --git a/Lib/test/test_asyncio/test_futures.py b/Lib/test/test_asyncio/test_futures.py
index 8b51522278a..39bef465bdb 100644
--- a/Lib/test/test_asyncio/test_futures.py
+++ b/Lib/test/test_asyncio/test_futures.py
@@ -413,7 +413,7 @@ class BaseFutureTests:
def test_copy_state(self):
from asyncio.futures import _copy_future_state
- f = self._new_future(loop=self.loop)
+ f = concurrent.futures.Future()
f.set_result(10)
newf = self._new_future(loop=self.loop)
@@ -421,7 +421,7 @@ class BaseFutureTests:
self.assertTrue(newf.done())
self.assertEqual(newf.result(), 10)
- f_exception = self._new_future(loop=self.loop)
+ f_exception = concurrent.futures.Future()
f_exception.set_exception(RuntimeError())
newf_exception = self._new_future(loop=self.loop)
@@ -429,7 +429,7 @@ class BaseFutureTests:
self.assertTrue(newf_exception.done())
self.assertRaises(RuntimeError, newf_exception.result)
- f_cancelled = self._new_future(loop=self.loop)
+ f_cancelled = concurrent.futures.Future()
f_cancelled.cancel()
newf_cancelled = self._new_future(loop=self.loop)
@@ -441,7 +441,7 @@ class BaseFutureTests:
except BaseException as e:
f_exc = e
- f_conexc = self._new_future(loop=self.loop)
+ f_conexc = concurrent.futures.Future()
f_conexc.set_exception(f_exc)
newf_conexc = self._new_future(loop=self.loop)
@@ -454,6 +454,56 @@ class BaseFutureTests:
newf_tb = ''.join(traceback.format_tb(newf_exc.__traceback__))
self.assertEqual(newf_tb.count('raise concurrent.futures.InvalidStateError'), 1)
+ def test_copy_state_from_concurrent_futures(self):
+ """Test _copy_future_state from concurrent.futures.Future.
+
+ This tests the optimized path using _get_snapshot when available.
+ """
+ from asyncio.futures import _copy_future_state
+
+ # Test with a result
+ f_concurrent = concurrent.futures.Future()
+ f_concurrent.set_result(42)
+ f_asyncio = self._new_future(loop=self.loop)
+ _copy_future_state(f_concurrent, f_asyncio)
+ self.assertTrue(f_asyncio.done())
+ self.assertEqual(f_asyncio.result(), 42)
+
+ # Test with an exception
+ f_concurrent_exc = concurrent.futures.Future()
+ f_concurrent_exc.set_exception(ValueError("test exception"))
+ f_asyncio_exc = self._new_future(loop=self.loop)
+ _copy_future_state(f_concurrent_exc, f_asyncio_exc)
+ self.assertTrue(f_asyncio_exc.done())
+ with self.assertRaises(ValueError) as cm:
+ f_asyncio_exc.result()
+ self.assertEqual(str(cm.exception), "test exception")
+
+ # Test with cancelled state
+ f_concurrent_cancelled = concurrent.futures.Future()
+ f_concurrent_cancelled.cancel()
+ f_asyncio_cancelled = self._new_future(loop=self.loop)
+ _copy_future_state(f_concurrent_cancelled, f_asyncio_cancelled)
+ self.assertTrue(f_asyncio_cancelled.cancelled())
+
+ # Test that destination already cancelled prevents copy
+ f_concurrent_result = concurrent.futures.Future()
+ f_concurrent_result.set_result(10)
+ f_asyncio_precancelled = self._new_future(loop=self.loop)
+ f_asyncio_precancelled.cancel()
+ _copy_future_state(f_concurrent_result, f_asyncio_precancelled)
+ self.assertTrue(f_asyncio_precancelled.cancelled())
+
+ # Test exception type conversion
+ f_concurrent_invalid = concurrent.futures.Future()
+ f_concurrent_invalid.set_exception(concurrent.futures.InvalidStateError("invalid"))
+ f_asyncio_invalid = self._new_future(loop=self.loop)
+ _copy_future_state(f_concurrent_invalid, f_asyncio_invalid)
+ self.assertTrue(f_asyncio_invalid.done())
+ with self.assertRaises(asyncio.exceptions.InvalidStateError) as cm:
+ f_asyncio_invalid.result()
+ self.assertEqual(str(cm.exception), "invalid")
+
def test_iter(self):
fut = self._new_future(loop=self.loop)
diff --git a/Lib/test/test_concurrent_futures/test_future.py b/Lib/test/test_concurrent_futures/test_future.py
index 4066ea1ee4b..06b11a3bacf 100644
--- a/Lib/test/test_concurrent_futures/test_future.py
+++ b/Lib/test/test_concurrent_futures/test_future.py
@@ -6,6 +6,7 @@ from concurrent.futures._base import (
PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future)
from test import support
+from test.support import threading_helper
from .util import (
PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE,
@@ -282,6 +283,62 @@ class FutureTests(BaseTestCase):
self.assertEqual(f.exception(), e)
+ def test_get_snapshot(self):
+ """Test the _get_snapshot method for atomic state retrieval."""
+ # Test with a pending future
+ f = Future()
+ done, cancelled, result, exception = f._get_snapshot()
+ self.assertFalse(done)
+ self.assertFalse(cancelled)
+ self.assertIsNone(result)
+ self.assertIsNone(exception)
+
+ # Test with a finished future (successful result)
+ f = Future()
+ f.set_result(42)
+ done, cancelled, result, exception = f._get_snapshot()
+ self.assertTrue(done)
+ self.assertFalse(cancelled)
+ self.assertEqual(result, 42)
+ self.assertIsNone(exception)
+
+ # Test with a finished future (exception)
+ f = Future()
+ exc = ValueError("test error")
+ f.set_exception(exc)
+ done, cancelled, result, exception = f._get_snapshot()
+ self.assertTrue(done)
+ self.assertFalse(cancelled)
+ self.assertIsNone(result)
+ self.assertIs(exception, exc)
+
+ # Test with a cancelled future
+ f = Future()
+ f.cancel()
+ done, cancelled, result, exception = f._get_snapshot()
+ self.assertTrue(done)
+ self.assertTrue(cancelled)
+ self.assertIsNone(result)
+ self.assertIsNone(exception)
+
+ # Test concurrent access (basic thread safety check)
+ f = Future()
+ f.set_result(100)
+ results = []
+
+ def get_snapshot():
+ for _ in range(1000):
+ snapshot = f._get_snapshot()
+ results.append(snapshot)
+
+ threads = [threading.Thread(target=get_snapshot) for _ in range(4)]
+ with threading_helper.start_threads(threads):
+ pass
+ # All snapshots should be identical for a finished future
+ expected = (True, False, 100, None)
+ for result in results:
+ self.assertEqual(result, expected)
+
def setUpModule():
setup_module()
diff --git a/Lib/test/test_faulthandler.py b/Lib/test/test_faulthandler.py
index 371c63adce9..2fb963f52e5 100644
--- a/Lib/test/test_faulthandler.py
+++ b/Lib/test/test_faulthandler.py
@@ -166,29 +166,6 @@ class FaultHandlerTests(unittest.TestCase):
fatal_error = 'Windows fatal exception: %s' % name_regex
self.check_error(code, line_number, fatal_error, **kw)
- @unittest.skipIf(sys.platform.startswith('aix'),
- "the first page of memory is a mapped read-only on AIX")
- def test_read_null(self):
- if not MS_WINDOWS:
- self.check_fatal_error("""
- import faulthandler
- faulthandler.enable()
- faulthandler._read_null()
- """,
- 3,
- # Issue #12700: Read NULL raises SIGILL on Mac OS X Lion
- '(?:Segmentation fault'
- '|Bus error'
- '|Illegal instruction)')
- else:
- self.check_windows_exception("""
- import faulthandler
- faulthandler.enable()
- faulthandler._read_null()
- """,
- 3,
- 'access violation')
-
@skip_segfault_on_android
def test_sigsegv(self):
self.check_fatal_error("""
diff --git a/Lib/test/test_fractions.py b/Lib/test/test_fractions.py
index 84faa636064..96b3f305194 100644
--- a/Lib/test/test_fractions.py
+++ b/Lib/test/test_fractions.py
@@ -1,7 +1,7 @@
"""Tests for Lib/fractions.py."""
from decimal import Decimal
-from test.support import requires_IEEE_754
+from test.support import requires_IEEE_754, adjust_int_max_str_digits
import math
import numbers
import operator
@@ -395,12 +395,14 @@ class FractionTest(unittest.TestCase):
def testFromString(self):
self.assertEqual((5, 1), _components(F("5")))
+ self.assertEqual((5, 1), _components(F("005")))
self.assertEqual((3, 2), _components(F("3/2")))
self.assertEqual((3, 2), _components(F("3 / 2")))
self.assertEqual((3, 2), _components(F(" \n +3/2")))
self.assertEqual((-3, 2), _components(F("-3/2 ")))
- self.assertEqual((13, 2), _components(F(" 013/02 \n ")))
+ self.assertEqual((13, 2), _components(F(" 0013/002 \n ")))
self.assertEqual((16, 5), _components(F(" 3.2 ")))
+ self.assertEqual((16, 5), _components(F("003.2")))
self.assertEqual((-16, 5), _components(F(" -3.2 ")))
self.assertEqual((-3, 1), _components(F(" -3. ")))
self.assertEqual((3, 5), _components(F(" .6 ")))
@@ -419,116 +421,102 @@ class FractionTest(unittest.TestCase):
self.assertRaisesMessage(
ZeroDivisionError, "Fraction(3, 0)",
F, "3/0")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '3/'",
- F, "3/")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '/2'",
- F, "/2")
- self.assertRaisesMessage(
- # Denominators don't need a sign.
- ValueError, "Invalid literal for Fraction: '3/+2'",
- F, "3/+2")
- self.assertRaisesMessage(
- # Imitate float's parsing.
- ValueError, "Invalid literal for Fraction: '+ 3/2'",
- F, "+ 3/2")
- self.assertRaisesMessage(
- # Avoid treating '.' as a regex special character.
- ValueError, "Invalid literal for Fraction: '3a2'",
- F, "3a2")
- self.assertRaisesMessage(
- # Don't accept combinations of decimals and rationals.
- ValueError, "Invalid literal for Fraction: '3/7.2'",
- F, "3/7.2")
- self.assertRaisesMessage(
- # Don't accept combinations of decimals and rationals.
- ValueError, "Invalid literal for Fraction: '3.2/7'",
- F, "3.2/7")
- self.assertRaisesMessage(
- # Allow 3. and .3, but not .
- ValueError, "Invalid literal for Fraction: '.'",
- F, ".")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '_'",
- F, "_")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '_1'",
- F, "_1")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1__2'",
- F, "1__2")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '/_'",
- F, "/_")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1_/'",
- F, "1_/")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '_1/'",
- F, "_1/")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1__2/'",
- F, "1__2/")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/_'",
- F, "1/_")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/_1'",
- F, "1/_1")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/1__2'",
- F, "1/1__2")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1._111'",
- F, "1._111")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1.1__1'",
- F, "1.1__1")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1.1e+_1'",
- F, "1.1e+_1")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1.1e+1__1'",
- F, "1.1e+1__1")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '123.dd'",
- F, "123.dd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '123.5_dd'",
- F, "123.5_dd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: 'dd.5'",
- F, "dd.5")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '7_dd'",
- F, "7_dd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/dd'",
- F, "1/dd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/123_dd'",
- F, "1/123_dd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '789edd'",
- F, "789edd")
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '789e2_dd'",
- F, "789e2_dd")
+
+ def check_invalid(s):
+ msg = "Invalid literal for Fraction: " + repr(s)
+ self.assertRaisesMessage(ValueError, msg, F, s)
+
+ check_invalid("3/")
+ check_invalid("/2")
+ # Denominators don't need a sign.
+ check_invalid("3/+2")
+ check_invalid("3/-2")
+ # Imitate float's parsing.
+ check_invalid("+ 3/2")
+ check_invalid("- 3/2")
+ # Avoid treating '.' as a regex special character.
+ check_invalid("3a2")
+ # Don't accept combinations of decimals and rationals.
+ check_invalid("3/7.2")
+ check_invalid("3.2/7")
+ # No space around dot.
+ check_invalid("3 .2")
+ check_invalid("3. 2")
+ # No space around e.
+ check_invalid("3.2 e1")
+ check_invalid("3.2e 1")
+ # Fractional part don't need a sign.
+ check_invalid("3.+2")
+ check_invalid("3.-2")
+ # Only accept base 10.
+ check_invalid("0x10")
+ check_invalid("0x10/1")
+ check_invalid("1/0x10")
+ check_invalid("0x10.")
+ check_invalid("0x10.1")
+ check_invalid("1.0x10")
+ check_invalid("1.0e0x10")
+ # Only accept decimal digits.
+ check_invalid("³")
+ check_invalid("³/2")
+ check_invalid("3/²")
+ check_invalid("³.2")
+ check_invalid("3.²")
+ check_invalid("3.2e²")
+ check_invalid("¼")
+ # Allow 3. and .3, but not .
+ check_invalid(".")
+ check_invalid("_")
+ check_invalid("_1")
+ check_invalid("1__2")
+ check_invalid("/_")
+ check_invalid("1_/")
+ check_invalid("_1/")
+ check_invalid("1__2/")
+ check_invalid("1/_")
+ check_invalid("1/_1")
+ check_invalid("1/1__2")
+ check_invalid("1._111")
+ check_invalid("1.1__1")
+ check_invalid("1.1e+_1")
+ check_invalid("1.1e+1__1")
+ check_invalid("123.dd")
+ check_invalid("123.5_dd")
+ check_invalid("dd.5")
+ check_invalid("7_dd")
+ check_invalid("1/dd")
+ check_invalid("1/123_dd")
+ check_invalid("789edd")
+ check_invalid("789e2_dd")
# Test catastrophic backtracking.
val = "9"*50 + "_"
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '" + val + "'",
- F, val)
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1/" + val + "'",
- F, "1/" + val)
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1." + val + "'",
- F, "1." + val)
- self.assertRaisesMessage(
- ValueError, "Invalid literal for Fraction: '1.1+e" + val + "'",
- F, "1.1+e" + val)
+ check_invalid(val)
+ check_invalid("1/" + val)
+ check_invalid("1." + val)
+ check_invalid("." + val)
+ check_invalid("1.1+e" + val)
+ check_invalid("1.1e" + val)
+
+ def test_limit_int(self):
+ maxdigits = 5000
+ with adjust_int_max_str_digits(maxdigits):
+ msg = 'Exceeds the limit'
+ val = '1' * maxdigits
+ num = (10**maxdigits - 1)//9
+ self.assertEqual((num, 1), _components(F(val)))
+ self.assertRaisesRegex(ValueError, msg, F, val + '1')
+ self.assertEqual((num, 2), _components(F(val + '/2')))
+ self.assertRaisesRegex(ValueError, msg, F, val + '1/2')
+ self.assertEqual((1, num), _components(F('1/' + val)))
+ self.assertRaisesRegex(ValueError, msg, F, '1/1' + val)
+ self.assertEqual(((10**(maxdigits+1) - 1)//9, 10**maxdigits),
+ _components(F('1.' + val)))
+ self.assertRaisesRegex(ValueError, msg, F, '1.1' + val)
+ self.assertEqual((num, 10**maxdigits), _components(F('.' + val)))
+ self.assertRaisesRegex(ValueError, msg, F, '.1' + val)
+ self.assertRaisesRegex(ValueError, msg, F, '1.1e1' + val)
+ self.assertEqual((11, 10), _components(F('1.1e' + '0' * maxdigits)))
+ self.assertRaisesRegex(ValueError, msg, F, '1.1e' + '0' * (maxdigits+1))
def testImmutable(self):
r = F(7, 3)
diff --git a/Lib/test/test_free_threading/test_functools.py b/Lib/test/test_free_threading/test_functools.py
new file mode 100644
index 00000000000..a442fe056ce
--- /dev/null
+++ b/Lib/test/test_free_threading/test_functools.py
@@ -0,0 +1,75 @@
+import random
+import unittest
+
+from functools import lru_cache
+from threading import Barrier, Thread
+
+from test.support import threading_helper
+
+@threading_helper.requires_working_threading()
+class TestLRUCache(unittest.TestCase):
+
+ def _test_concurrent_operations(self, maxsize):
+ num_threads = 10
+ b = Barrier(num_threads)
+ @lru_cache(maxsize=maxsize)
+ def func(arg=0):
+ return object()
+
+
+ def thread_func():
+ b.wait()
+ for i in range(1000):
+ r = random.randint(0, 1000)
+ if i < 800:
+ func(i)
+ elif i < 900:
+ func.cache_info()
+ else:
+ func.cache_clear()
+
+ threads = []
+ for i in range(num_threads):
+ t = Thread(target=thread_func)
+ threads.append(t)
+
+ with threading_helper.start_threads(threads):
+ pass
+
+ def test_concurrent_operations_unbounded(self):
+ self._test_concurrent_operations(maxsize=None)
+
+ def test_concurrent_operations_bounded(self):
+ self._test_concurrent_operations(maxsize=128)
+
+ def _test_reentrant_cache_clear(self, maxsize):
+ num_threads = 10
+ b = Barrier(num_threads)
+ @lru_cache(maxsize=maxsize)
+ def func(arg=0):
+ func.cache_clear()
+ return object()
+
+
+ def thread_func():
+ b.wait()
+ for i in range(1000):
+ func(random.randint(0, 10000))
+
+ threads = []
+ for i in range(num_threads):
+ t = Thread(target=thread_func)
+ threads.append(t)
+
+ with threading_helper.start_threads(threads):
+ pass
+
+ def test_reentrant_cache_clear_unbounded(self):
+ self._test_reentrant_cache_clear(maxsize=None)
+
+ def test_reentrant_cache_clear_bounded(self):
+ self._test_reentrant_cache_clear(maxsize=128)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/Lib/test/test_genericalias.py b/Lib/test/test_genericalias.py
index 8d21ded4501..ea0dc241e39 100644
--- a/Lib/test/test_genericalias.py
+++ b/Lib/test/test_genericalias.py
@@ -61,6 +61,7 @@ try:
from tkinter import Event
except ImportError:
Event = None
+from string.templatelib import Template, Interpolation
from typing import TypeVar
T = TypeVar('T')
@@ -139,7 +140,10 @@ class BaseTest(unittest.TestCase):
DictReader, DictWriter,
array,
staticmethod,
- classmethod]
+ classmethod,
+ Template,
+ Interpolation,
+ ]
if ctypes is not None:
generic_types.extend((ctypes.Array, ctypes.LibraryLoader, ctypes.py_object))
if ValueProxy is not None:
diff --git a/Lib/test/test_hmac.py b/Lib/test/test_hmac.py
index 70c79437722..e898644dd8a 100644
--- a/Lib/test/test_hmac.py
+++ b/Lib/test/test_hmac.py
@@ -1,8 +1,27 @@
+"""Test suite for HMAC.
+
+Python provides three different implementations of HMAC:
+
+- OpenSSL HMAC using OpenSSL hash functions.
+- HACL* HMAC using HACL* hash functions.
+- Generic Python HMAC using user-defined hash functions.
+
+The generic Python HMAC implementation is able to use OpenSSL
+callables or names, HACL* named hash functions or arbitrary
+objects implementing PEP 247 interface.
+
+In the two first cases, Python HMAC wraps a C HMAC object (either OpenSSL
+or HACL*-based). As a last resort, HMAC is re-implemented in pure Python.
+It is however interesting to test the pure Python implementation against
+the OpenSSL and HACL* hash functions.
+"""
+
import binascii
import functools
import hmac
import hashlib
import random
+import test.support
import test.support.hashlib_helper as hashlib_helper
import types
import unittest
@@ -10,6 +29,12 @@ import unittest.mock as mock
import warnings
from _operator import _compare_digest as operator_compare_digest
from test.support import check_disallow_instantiation
+from test.support.hashlib_helper import (
+ BuiltinHashFunctionsTrait,
+ HashFunctionsTrait,
+ NamedHashFunctionsTrait,
+ OpenSSLHashFunctionsTrait,
+)
from test.support.import_helper import import_fresh_module, import_module
try:
@@ -382,50 +407,7 @@ class BuiltinAssertersMixin(ThroughBuiltinAPIMixin, AssertersMixin):
pass
-class HashFunctionsTrait:
- """Trait class for 'hashfunc' in hmac_new() and hmac_digest()."""
-
- ALGORITHMS = [
- 'md5', 'sha1',
- 'sha224', 'sha256', 'sha384', 'sha512',
- 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512',
- ]
-
- # By default, a missing algorithm skips the test that uses it.
- _ = property(lambda self: self.skipTest("missing hash function"))
- md5 = sha1 = _
- sha224 = sha256 = sha384 = sha512 = _
- sha3_224 = sha3_256 = sha3_384 = sha3_512 = _
- del _
-
-
-class WithOpenSSLHashFunctions(HashFunctionsTrait):
- """Test a HMAC implementation with an OpenSSL-based callable 'hashfunc'."""
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- for name in cls.ALGORITHMS:
- @property
- @hashlib_helper.requires_openssl_hashdigest(name)
- def func(self, *, __name=name): # __name needed to bind 'name'
- return getattr(_hashlib, f'openssl_{__name}')
- setattr(cls, name, func)
-
-
-class WithNamedHashFunctions(HashFunctionsTrait):
- """Test a HMAC implementation with a named 'hashfunc'."""
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- for name in cls.ALGORITHMS:
- setattr(cls, name, name)
-
-
-class RFCTestCaseMixin(AssertersMixin, HashFunctionsTrait):
+class RFCTestCaseMixin(HashFunctionsTrait, AssertersMixin):
"""Test HMAC implementations against RFC 2202/4231 and NIST test vectors.
- Test vectors for MD5 and SHA-1 are taken from RFC 2202.
@@ -739,26 +721,83 @@ class RFCTestCaseMixin(AssertersMixin, HashFunctionsTrait):
)
-class PyRFCTestCase(ThroughObjectMixin, PyAssertersMixin,
- WithOpenSSLHashFunctions, RFCTestCaseMixin,
- unittest.TestCase):
+class PurePythonInitHMAC(PyModuleMixin, HashFunctionsTrait):
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ for meth in ['_init_openssl_hmac', '_init_builtin_hmac']:
+ fn = getattr(cls.hmac.HMAC, meth)
+ cm = mock.patch.object(cls.hmac.HMAC, meth, autospec=True, wraps=fn)
+ cls.enterClassContext(cm)
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.hmac.HMAC._init_openssl_hmac.assert_not_called()
+ cls.hmac.HMAC._init_builtin_hmac.assert_not_called()
+ # Do not assert that HMAC._init_old() has been called as it's tricky
+ # to determine whether a test for a specific hash function has been
+ # executed or not. On regular builds, it will be called but if a
+ # hash function is not available, it's hard to detect for which
+ # test we should checj HMAC._init_old() or not.
+ super().tearDownClass()
+
+
+class PyRFCOpenSSLTestCase(ThroughObjectMixin,
+ PyAssertersMixin,
+ OpenSSLHashFunctionsTrait,
+ RFCTestCaseMixin,
+ PurePythonInitHMAC,
+ unittest.TestCase):
"""Python implementation of HMAC using hmac.HMAC().
- The underlying hash functions are OpenSSL-based.
+ The underlying hash functions are OpenSSL-based but
+ _init_old() is used instead of _init_openssl_hmac().
"""
-class PyDotNewRFCTestCase(ThroughModuleAPIMixin, PyAssertersMixin,
- WithOpenSSLHashFunctions, RFCTestCaseMixin,
- unittest.TestCase):
+class PyRFCBuiltinTestCase(ThroughObjectMixin,
+ PyAssertersMixin,
+ BuiltinHashFunctionsTrait,
+ RFCTestCaseMixin,
+ PurePythonInitHMAC,
+ unittest.TestCase):
+ """Python implementation of HMAC using hmac.HMAC().
+
+ The underlying hash functions are HACL*-based but
+ _init_old() is used instead of _init_builtin_hmac().
+ """
+
+
+class PyDotNewOpenSSLRFCTestCase(ThroughModuleAPIMixin,
+ PyAssertersMixin,
+ OpenSSLHashFunctionsTrait,
+ RFCTestCaseMixin,
+ PurePythonInitHMAC,
+ unittest.TestCase):
+ """Python implementation of HMAC using hmac.new().
+
+ The underlying hash functions are OpenSSL-based but
+ _init_old() is used instead of _init_openssl_hmac().
+ """
+
+
+class PyDotNewBuiltinRFCTestCase(ThroughModuleAPIMixin,
+ PyAssertersMixin,
+ BuiltinHashFunctionsTrait,
+ RFCTestCaseMixin,
+ PurePythonInitHMAC,
+ unittest.TestCase):
"""Python implementation of HMAC using hmac.new().
- The underlying hash functions are OpenSSL-based.
+ The underlying hash functions are HACL-based but
+ _init_old() is used instead of _init_openssl_hmac().
"""
class OpenSSLRFCTestCase(OpenSSLAssertersMixin,
- WithOpenSSLHashFunctions, RFCTestCaseMixin,
+ OpenSSLHashFunctionsTrait,
+ RFCTestCaseMixin,
unittest.TestCase):
"""OpenSSL implementation of HMAC.
@@ -767,7 +806,8 @@ class OpenSSLRFCTestCase(OpenSSLAssertersMixin,
class BuiltinRFCTestCase(BuiltinAssertersMixin,
- WithNamedHashFunctions, RFCTestCaseMixin,
+ NamedHashFunctionsTrait,
+ RFCTestCaseMixin,
unittest.TestCase):
"""Built-in HACL* implementation of HMAC.
@@ -784,12 +824,6 @@ class BuiltinRFCTestCase(BuiltinAssertersMixin,
self.check_hmac_hexdigest(key, msg, hexdigest, digest_size, func)
-# TODO(picnixz): once we have a HACL* HMAC, we should also test the Python
-# implementation of HMAC with a HACL*-based hash function. For now, we only
-# test it partially via the '_sha2' module, but for completeness we could
-# also test the RFC test vectors against all possible implementations.
-
-
class DigestModTestCaseMixin(CreatorMixin, DigestMixin):
"""Tests for the 'digestmod' parameter for hmac_new() and hmac_digest()."""
diff --git a/Lib/test/test_httpservers.py b/Lib/test/test_httpservers.py
index 2cafa4e45a1..df5d2a7bedc 100644
--- a/Lib/test/test_httpservers.py
+++ b/Lib/test/test_httpservers.py
@@ -3,16 +3,15 @@
Written by Cody A.W. Somerville <cody-somerville@ubuntu.com>,
Josip Dzolonga, and Michael Otteneder for the 2007/08 GHOP contest.
"""
-from collections import OrderedDict
+
from http.server import BaseHTTPRequestHandler, HTTPServer, HTTPSServer, \
- SimpleHTTPRequestHandler, CGIHTTPRequestHandler
+ SimpleHTTPRequestHandler
from http import server, HTTPStatus
import os
import socket
import sys
import re
-import base64
import ntpath
import pathlib
import shutil
@@ -31,7 +30,7 @@ from io import BytesIO, StringIO
import unittest
from test import support
from test.support import (
- is_apple, import_helper, os_helper, requires_subprocess, threading_helper
+ is_apple, import_helper, os_helper, threading_helper
)
try:
@@ -522,42 +521,120 @@ class SimpleHTTPServerTestCase(BaseTestCase):
reader.close()
return body
+ def check_list_dir_dirname(self, dirname, quotedname=None):
+ fullpath = os.path.join(self.tempdir, dirname)
+ try:
+ os.mkdir(os.path.join(self.tempdir, dirname))
+ except (OSError, UnicodeEncodeError):
+ self.skipTest(f'Can not create directory {dirname!a} '
+ f'on current file system')
+
+ if quotedname is None:
+ quotedname = urllib.parse.quote(dirname, errors='surrogatepass')
+ response = self.request(self.base_url + '/' + quotedname + '/')
+ body = self.check_status_and_reason(response, HTTPStatus.OK)
+ displaypath = html.escape(f'{self.base_url}/{dirname}/', quote=False)
+ enc = sys.getfilesystemencoding()
+ prefix = f'listing for {displaypath}</'.encode(enc, 'surrogateescape')
+ self.assertIn(prefix + b'title>', body)
+ self.assertIn(prefix + b'h1>', body)
+
+ def check_list_dir_filename(self, filename):
+ fullpath = os.path.join(self.tempdir, filename)
+ content = ascii(fullpath).encode() + (os_helper.TESTFN_UNDECODABLE or b'\xff')
+ try:
+ with open(fullpath, 'wb') as f:
+ f.write(content)
+ except OSError:
+ self.skipTest(f'Can not create file {filename!a} '
+ f'on current file system')
+
+ response = self.request(self.base_url + '/')
+ body = self.check_status_and_reason(response, HTTPStatus.OK)
+ quotedname = urllib.parse.quote(filename, errors='surrogatepass')
+ enc = response.headers.get_content_charset()
+ self.assertIsNotNone(enc)
+ self.assertIn((f'href="{quotedname}"').encode('ascii'), body)
+ displayname = html.escape(filename, quote=False)
+ self.assertIn(f'>{displayname}<'.encode(enc, 'surrogateescape'), body)
+
+ response = self.request(self.base_url + '/' + quotedname)
+ self.check_status_and_reason(response, HTTPStatus.OK, data=content)
+
+ @unittest.skipUnless(os_helper.TESTFN_NONASCII,
+ 'need os_helper.TESTFN_NONASCII')
+ def test_list_dir_nonascii_dirname(self):
+ dirname = os_helper.TESTFN_NONASCII + '.dir'
+ self.check_list_dir_dirname(dirname)
+
+ @unittest.skipUnless(os_helper.TESTFN_NONASCII,
+ 'need os_helper.TESTFN_NONASCII')
+ def test_list_dir_nonascii_filename(self):
+ filename = os_helper.TESTFN_NONASCII + '.txt'
+ self.check_list_dir_filename(filename)
+
@unittest.skipIf(is_apple,
'undecodable name cannot always be decoded on Apple platforms')
@unittest.skipIf(sys.platform == 'win32',
'undecodable name cannot be decoded on win32')
@unittest.skipUnless(os_helper.TESTFN_UNDECODABLE,
'need os_helper.TESTFN_UNDECODABLE')
- def test_undecodable_filename(self):
- enc = sys.getfilesystemencoding()
- filename = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.txt'
- with open(os.path.join(self.tempdir, filename), 'wb') as f:
- f.write(os_helper.TESTFN_UNDECODABLE)
- response = self.request(self.base_url + '/')
- if is_apple:
- # On Apple platforms the HFS+ filesystem replaces bytes that
- # aren't valid UTF-8 into a percent-encoded value.
- for name in os.listdir(self.tempdir):
- if name != 'test': # Ignore a filename created in setUp().
- filename = name
- break
- body = self.check_status_and_reason(response, HTTPStatus.OK)
- quotedname = urllib.parse.quote(filename, errors='surrogatepass')
- self.assertIn(('href="%s"' % quotedname)
- .encode(enc, 'surrogateescape'), body)
- self.assertIn(('>%s<' % html.escape(filename, quote=False))
- .encode(enc, 'surrogateescape'), body)
- response = self.request(self.base_url + '/' + quotedname)
- self.check_status_and_reason(response, HTTPStatus.OK,
- data=os_helper.TESTFN_UNDECODABLE)
+ def test_list_dir_undecodable_dirname(self):
+ dirname = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.dir'
+ self.check_list_dir_dirname(dirname)
- def test_undecodable_parameter(self):
- # sanity check using a valid parameter
+ @unittest.skipIf(is_apple,
+ 'undecodable name cannot always be decoded on Apple platforms')
+ @unittest.skipIf(sys.platform == 'win32',
+ 'undecodable name cannot be decoded on win32')
+ @unittest.skipUnless(os_helper.TESTFN_UNDECODABLE,
+ 'need os_helper.TESTFN_UNDECODABLE')
+ def test_list_dir_undecodable_filename(self):
+ filename = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.txt'
+ self.check_list_dir_filename(filename)
+
+ def test_list_dir_undecodable_dirname2(self):
+ dirname = '\ufffd.dir'
+ self.check_list_dir_dirname(dirname, quotedname='%ff.dir')
+
+ @unittest.skipUnless(os_helper.TESTFN_UNENCODABLE,
+ 'need os_helper.TESTFN_UNENCODABLE')
+ def test_list_dir_unencodable_dirname(self):
+ dirname = os_helper.TESTFN_UNENCODABLE + '.dir'
+ self.check_list_dir_dirname(dirname)
+
+ @unittest.skipUnless(os_helper.TESTFN_UNENCODABLE,
+ 'need os_helper.TESTFN_UNENCODABLE')
+ def test_list_dir_unencodable_filename(self):
+ filename = os_helper.TESTFN_UNENCODABLE + '.txt'
+ self.check_list_dir_filename(filename)
+
+ def test_list_dir_escape_dirname(self):
+ # Characters that need special treating in URL or HTML.
+ for name in ('q?', 'f#', '&amp;', '&amp', '<i>', '"dq"', "'sq'",
+ '%A4', '%E2%82%AC'):
+ with self.subTest(name=name):
+ dirname = name + '.dir'
+ self.check_list_dir_dirname(dirname,
+ quotedname=urllib.parse.quote(dirname, safe='&<>\'"'))
+
+ def test_list_dir_escape_filename(self):
+ # Characters that need special treating in URL or HTML.
+ for name in ('q?', 'f#', '&amp;', '&amp', '<i>', '"dq"', "'sq'",
+ '%A4', '%E2%82%AC'):
+ with self.subTest(name=name):
+ filename = name + '.txt'
+ self.check_list_dir_filename(filename)
+ os_helper.unlink(os.path.join(self.tempdir, filename))
+
+ def test_list_dir_with_query_and_fragment(self):
+ prefix = f'listing for {self.base_url}/</'.encode('latin1')
+ response = self.request(self.base_url + '/#123').read()
+ self.assertIn(prefix + b'title>', response)
+ self.assertIn(prefix + b'h1>', response)
response = self.request(self.base_url + '/?x=123').read()
- self.assertRegex(response, rf'listing for {self.base_url}/\?x=123'.encode('latin1'))
- # now the bogus encoding
- response = self.request(self.base_url + '/?x=%bb').read()
- self.assertRegex(response, rf'listing for {self.base_url}/\?x=\xef\xbf\xbd'.encode('latin1'))
+ self.assertIn(prefix + b'title>', response)
+ self.assertIn(prefix + b'h1>', response)
def test_get_dir_redirect_location_domain_injection_bug(self):
"""Ensure //evil.co/..%2f../../X does not put //evil.co/ in Location.
@@ -615,10 +692,19 @@ class SimpleHTTPServerTestCase(BaseTestCase):
# check for trailing "/" which should return 404. See Issue17324
response = self.request(self.base_url + '/test/')
self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
+ response = self.request(self.base_url + '/test%2f')
+ self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
+ response = self.request(self.base_url + '/test%2F')
+ self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
response = self.request(self.base_url + '/')
self.check_status_and_reason(response, HTTPStatus.OK)
+ response = self.request(self.base_url + '%2f')
+ self.check_status_and_reason(response, HTTPStatus.OK)
+ response = self.request(self.base_url + '%2F')
+ self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.base_url)
self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
+ self.assertEqual(response.getheader("Location"), self.base_url + "/")
self.assertEqual(response.getheader("Content-Length"), "0")
response = self.request(self.base_url + '/?hi=2')
self.check_status_and_reason(response, HTTPStatus.OK)
@@ -724,6 +810,8 @@ class SimpleHTTPServerTestCase(BaseTestCase):
self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.tempdir_name)
self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
+ self.assertEqual(response.getheader("Location"),
+ self.tempdir_name + "/")
response = self.request(self.tempdir_name + '/?hi=2')
self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.tempdir_name + '?hi=1')
@@ -731,350 +819,6 @@ class SimpleHTTPServerTestCase(BaseTestCase):
self.assertEqual(response.getheader("Location"),
self.tempdir_name + "/?hi=1")
- def test_html_escape_filename(self):
- filename = '<test&>.txt'
- fullpath = os.path.join(self.tempdir, filename)
-
- try:
- open(fullpath, 'wb').close()
- except OSError:
- raise unittest.SkipTest('Can not create file %s on current file '
- 'system' % filename)
-
- try:
- response = self.request(self.base_url + '/')
- body = self.check_status_and_reason(response, HTTPStatus.OK)
- enc = response.headers.get_content_charset()
- finally:
- os.unlink(fullpath) # avoid affecting test_undecodable_filename
-
- self.assertIsNotNone(enc)
- html_text = '>%s<' % html.escape(filename, quote=False)
- self.assertIn(html_text.encode(enc), body)
-
-
-cgi_file1 = """\
-#!%s
-
-print("Content-type: text/html")
-print()
-print("Hello World")
-"""
-
-cgi_file2 = """\
-#!%s
-import os
-import sys
-import urllib.parse
-
-print("Content-type: text/html")
-print()
-
-content_length = int(os.environ["CONTENT_LENGTH"])
-query_string = sys.stdin.buffer.read(content_length)
-params = {key.decode("utf-8"): val.decode("utf-8")
- for key, val in urllib.parse.parse_qsl(query_string)}
-
-print("%%s, %%s, %%s" %% (params["spam"], params["eggs"], params["bacon"]))
-"""
-
-cgi_file4 = """\
-#!%s
-import os
-
-print("Content-type: text/html")
-print()
-
-print(os.environ["%s"])
-"""
-
-cgi_file6 = """\
-#!%s
-import os
-
-print("X-ambv: was here")
-print("Content-type: text/html")
-print()
-print("<pre>")
-for k, v in os.environ.items():
- try:
- k.encode('ascii')
- v.encode('ascii')
- except UnicodeEncodeError:
- continue # see: BPO-44647
- print(f"{k}={v}")
-print("</pre>")
-"""
-
-
-@unittest.skipIf(hasattr(os, 'geteuid') and os.geteuid() == 0,
- "This test can't be run reliably as root (issue #13308).")
-@requires_subprocess()
-class CGIHTTPServerTestCase(BaseTestCase):
- class request_handler(NoLogRequestHandler, CGIHTTPRequestHandler):
- _test_case_self = None # populated by each setUp() method call.
-
- def __init__(self, *args, **kwargs):
- with self._test_case_self.assertWarnsRegex(
- DeprecationWarning,
- r'http\.server\.CGIHTTPRequestHandler'):
- # This context also happens to catch and silence the
- # threading DeprecationWarning from os.fork().
- super().__init__(*args, **kwargs)
-
- linesep = os.linesep.encode('ascii')
-
- def setUp(self):
- self.request_handler._test_case_self = self # practical, but yuck.
- BaseTestCase.setUp(self)
- self.cwd = os.getcwd()
- self.parent_dir = tempfile.mkdtemp()
- self.cgi_dir = os.path.join(self.parent_dir, 'cgi-bin')
- self.cgi_child_dir = os.path.join(self.cgi_dir, 'child-dir')
- self.sub_dir_1 = os.path.join(self.parent_dir, 'sub')
- self.sub_dir_2 = os.path.join(self.sub_dir_1, 'dir')
- self.cgi_dir_in_sub_dir = os.path.join(self.sub_dir_2, 'cgi-bin')
- os.mkdir(self.cgi_dir)
- os.mkdir(self.cgi_child_dir)
- os.mkdir(self.sub_dir_1)
- os.mkdir(self.sub_dir_2)
- os.mkdir(self.cgi_dir_in_sub_dir)
- self.nocgi_path = None
- self.file1_path = None
- self.file2_path = None
- self.file3_path = None
- self.file4_path = None
- self.file5_path = None
-
- # The shebang line should be pure ASCII: use symlink if possible.
- # See issue #7668.
- self._pythonexe_symlink = None
- if os_helper.can_symlink():
- self.pythonexe = os.path.join(self.parent_dir, 'python')
- self._pythonexe_symlink = support.PythonSymlink(self.pythonexe).__enter__()
- else:
- self.pythonexe = sys.executable
-
- try:
- # The python executable path is written as the first line of the
- # CGI Python script. The encoding cookie cannot be used, and so the
- # path should be encodable to the default script encoding (utf-8)
- self.pythonexe.encode('utf-8')
- except UnicodeEncodeError:
- self.tearDown()
- self.skipTest("Python executable path is not encodable to utf-8")
-
- self.nocgi_path = os.path.join(self.parent_dir, 'nocgi.py')
- with open(self.nocgi_path, 'w', encoding='utf-8') as fp:
- fp.write(cgi_file1 % self.pythonexe)
- os.chmod(self.nocgi_path, 0o777)
-
- self.file1_path = os.path.join(self.cgi_dir, 'file1.py')
- with open(self.file1_path, 'w', encoding='utf-8') as file1:
- file1.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file1_path, 0o777)
-
- self.file2_path = os.path.join(self.cgi_dir, 'file2.py')
- with open(self.file2_path, 'w', encoding='utf-8') as file2:
- file2.write(cgi_file2 % self.pythonexe)
- os.chmod(self.file2_path, 0o777)
-
- self.file3_path = os.path.join(self.cgi_child_dir, 'file3.py')
- with open(self.file3_path, 'w', encoding='utf-8') as file3:
- file3.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file3_path, 0o777)
-
- self.file4_path = os.path.join(self.cgi_dir, 'file4.py')
- with open(self.file4_path, 'w', encoding='utf-8') as file4:
- file4.write(cgi_file4 % (self.pythonexe, 'QUERY_STRING'))
- os.chmod(self.file4_path, 0o777)
-
- self.file5_path = os.path.join(self.cgi_dir_in_sub_dir, 'file5.py')
- with open(self.file5_path, 'w', encoding='utf-8') as file5:
- file5.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file5_path, 0o777)
-
- self.file6_path = os.path.join(self.cgi_dir, 'file6.py')
- with open(self.file6_path, 'w', encoding='utf-8') as file6:
- file6.write(cgi_file6 % self.pythonexe)
- os.chmod(self.file6_path, 0o777)
-
- os.chdir(self.parent_dir)
-
- def tearDown(self):
- self.request_handler._test_case_self = None
- try:
- os.chdir(self.cwd)
- if self._pythonexe_symlink:
- self._pythonexe_symlink.__exit__(None, None, None)
- if self.nocgi_path:
- os.remove(self.nocgi_path)
- if self.file1_path:
- os.remove(self.file1_path)
- if self.file2_path:
- os.remove(self.file2_path)
- if self.file3_path:
- os.remove(self.file3_path)
- if self.file4_path:
- os.remove(self.file4_path)
- if self.file5_path:
- os.remove(self.file5_path)
- if self.file6_path:
- os.remove(self.file6_path)
- os.rmdir(self.cgi_child_dir)
- os.rmdir(self.cgi_dir)
- os.rmdir(self.cgi_dir_in_sub_dir)
- os.rmdir(self.sub_dir_2)
- os.rmdir(self.sub_dir_1)
- # The 'gmon.out' file can be written in the current working
- # directory if C-level code profiling with gprof is enabled.
- os_helper.unlink(os.path.join(self.parent_dir, 'gmon.out'))
- os.rmdir(self.parent_dir)
- finally:
- BaseTestCase.tearDown(self)
-
- def test_url_collapse_path(self):
- # verify tail is the last portion and head is the rest on proper urls
- test_vectors = {
- '': '//',
- '..': IndexError,
- '/.//..': IndexError,
- '/': '//',
- '//': '//',
- '/\\': '//\\',
- '/.//': '//',
- 'cgi-bin/file1.py': '/cgi-bin/file1.py',
- '/cgi-bin/file1.py': '/cgi-bin/file1.py',
- 'a': '//a',
- '/a': '//a',
- '//a': '//a',
- './a': '//a',
- './C:/': '/C:/',
- '/a/b': '/a/b',
- '/a/b/': '/a/b/',
- '/a/b/.': '/a/b/',
- '/a/b/c/..': '/a/b/',
- '/a/b/c/../d': '/a/b/d',
- '/a/b/c/../d/e/../f': '/a/b/d/f',
- '/a/b/c/../d/e/../../f': '/a/b/f',
- '/a/b/c/../d/e/.././././..//f': '/a/b/f',
- '../a/b/c/../d/e/.././././..//f': IndexError,
- '/a/b/c/../d/e/../../../f': '/a/f',
- '/a/b/c/../d/e/../../../../f': '//f',
- '/a/b/c/../d/e/../../../../../f': IndexError,
- '/a/b/c/../d/e/../../../../f/..': '//',
- '/a/b/c/../d/e/../../../../f/../.': '//',
- }
- for path, expected in test_vectors.items():
- if isinstance(expected, type) and issubclass(expected, Exception):
- self.assertRaises(expected,
- server._url_collapse_path, path)
- else:
- actual = server._url_collapse_path(path)
- self.assertEqual(expected, actual,
- msg='path = %r\nGot: %r\nWanted: %r' %
- (path, actual, expected))
-
- def test_headers_and_content(self):
- res = self.request('/cgi-bin/file1.py')
- self.assertEqual(
- (res.read(), res.getheader('Content-type'), res.status),
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK))
-
- def test_issue19435(self):
- res = self.request('///////////nocgi.py/../cgi-bin/nothere.sh')
- self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
-
- def test_post(self):
- params = urllib.parse.urlencode(
- {'spam' : 1, 'eggs' : 'python', 'bacon' : 123456})
- headers = {'Content-type' : 'application/x-www-form-urlencoded'}
- res = self.request('/cgi-bin/file2.py', 'POST', params, headers)
-
- self.assertEqual(res.read(), b'1, python, 123456' + self.linesep)
-
- def test_invaliduri(self):
- res = self.request('/cgi-bin/invalid')
- res.read()
- self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
-
- def test_authorization(self):
- headers = {b'Authorization' : b'Basic ' +
- base64.b64encode(b'username:pass')}
- res = self.request('/cgi-bin/file1.py', 'GET', headers=headers)
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_no_leading_slash(self):
- # http://bugs.python.org/issue2254
- res = self.request('cgi-bin/file1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_os_environ_is_not_altered(self):
- signature = "Test CGI Server"
- os.environ['SERVER_SOFTWARE'] = signature
- res = self.request('/cgi-bin/file1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
- self.assertEqual(os.environ['SERVER_SOFTWARE'], signature)
-
- def test_urlquote_decoding_in_cgi_check(self):
- res = self.request('/cgi-bin%2ffile1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_nested_cgi_path_issue21323(self):
- res = self.request('/cgi-bin/child-dir/file3.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_query_with_multiple_question_mark(self):
- res = self.request('/cgi-bin/file4.py?a=b?c=d')
- self.assertEqual(
- (b'a=b?c=d' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_query_with_continuous_slashes(self):
- res = self.request('/cgi-bin/file4.py?k=aa%2F%2Fbb&//q//p//=//a//b//')
- self.assertEqual(
- (b'k=aa%2F%2Fbb&//q//p//=//a//b//' + self.linesep,
- 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_cgi_path_in_sub_directories(self):
- try:
- CGIHTTPRequestHandler.cgi_directories.append('/sub/dir/cgi-bin')
- res = self.request('/sub/dir/cgi-bin/file5.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
- finally:
- CGIHTTPRequestHandler.cgi_directories.remove('/sub/dir/cgi-bin')
-
- def test_accept(self):
- browser_accept = \
- 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
- tests = (
- ((('Accept', browser_accept),), browser_accept),
- ((), ''),
- # Hack case to get two values for the one header
- ((('Accept', 'text/html'), ('ACCEPT', 'text/plain')),
- 'text/html,text/plain'),
- )
- for headers, expected in tests:
- headers = OrderedDict(headers)
- with self.subTest(headers):
- res = self.request('/cgi-bin/file6.py', 'GET', headers=headers)
- self.assertEqual(http.HTTPStatus.OK, res.status)
- expected = f"HTTP_ACCEPT={expected}".encode('ascii')
- self.assertIn(expected, res.read())
-
class SocketlessRequestHandler(SimpleHTTPRequestHandler):
def __init__(self, directory=None):
@@ -1095,6 +839,7 @@ class SocketlessRequestHandler(SimpleHTTPRequestHandler):
def log_message(self, format, *args):
pass
+
class RejectingSocketlessRequestHandler(SocketlessRequestHandler):
def handle_expect_100(self):
self.send_error(HTTPStatus.EXPECTATION_FAILED)
diff --git a/Lib/test/test_importlib/import_/test_relative_imports.py b/Lib/test/test_importlib/import_/test_relative_imports.py
index e535d119763..1549cbe96ce 100644
--- a/Lib/test/test_importlib/import_/test_relative_imports.py
+++ b/Lib/test/test_importlib/import_/test_relative_imports.py
@@ -223,6 +223,21 @@ class RelativeImports:
self.__import__('sys', {'__package__': '', '__spec__': None},
level=1)
+ def test_malicious_relative_import(self):
+ # https://github.com/python/cpython/issues/134100
+ # Test to make sure UAF bug with error msg doesn't come back to life
+ import sys
+ loooong = "".ljust(0x23000, "b")
+ name = f"a.{loooong}.c"
+
+ with util.uncache(name):
+ sys.modules[name] = {}
+ with self.assertRaisesRegex(
+ KeyError,
+ r"'a\.b+' not in sys\.modules as expected"
+ ):
+ __import__(f"{loooong}.c", {"__package__": "a"}, level=1)
+
(Frozen_RelativeImports,
Source_RelativeImports
diff --git a/Lib/test/test_interpreters/test_api.py b/Lib/test/test_interpreters/test_api.py
index 66c7afce88f..1e2d572b1cb 100644
--- a/Lib/test/test_interpreters/test_api.py
+++ b/Lib/test/test_interpreters/test_api.py
@@ -1452,6 +1452,14 @@ class LowLevelTests(TestBase):
self.assertFalse(
self.interp_exists(interpid))
+ with self.subTest('basic C-API'):
+ interpid = _testinternalcapi.create_interpreter()
+ self.assertTrue(
+ self.interp_exists(interpid))
+ _testinternalcapi.destroy_interpreter(interpid, basic=True)
+ self.assertFalse(
+ self.interp_exists(interpid))
+
def test_get_config(self):
# This test overlaps with
# test.test_capi.test_misc.InterpreterConfigTests.
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
index 1e5adcc8db1..fa5b1e43816 100644
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -61,7 +61,7 @@ import warnings
import weakref
from http.server import HTTPServer, BaseHTTPRequestHandler
-from unittest.mock import patch
+from unittest.mock import call, Mock, patch
from urllib.parse import urlparse, parse_qs
from socketserver import (ThreadingUDPServer, DatagramRequestHandler,
ThreadingTCPServer, StreamRequestHandler)
@@ -5655,12 +5655,19 @@ class BasicConfigTest(unittest.TestCase):
assertRaises = self.assertRaises
handlers = [logging.StreamHandler()]
stream = sys.stderr
+ formatter = logging.Formatter()
assertRaises(ValueError, logging.basicConfig, filename='test.log',
stream=stream)
assertRaises(ValueError, logging.basicConfig, filename='test.log',
handlers=handlers)
assertRaises(ValueError, logging.basicConfig, stream=stream,
handlers=handlers)
+ assertRaises(ValueError, logging.basicConfig, formatter=formatter,
+ format='%(message)s')
+ assertRaises(ValueError, logging.basicConfig, formatter=formatter,
+ datefmt='%H:%M:%S')
+ assertRaises(ValueError, logging.basicConfig, formatter=formatter,
+ style='%')
# Issue 23207: test for invalid kwargs
assertRaises(ValueError, logging.basicConfig, loglevel=logging.INFO)
# Should pop both filename and filemode even if filename is None
@@ -5795,6 +5802,20 @@ class BasicConfigTest(unittest.TestCase):
# didn't write anything due to the encoding error
self.assertEqual(data, r'')
+ def test_formatter_given(self):
+ mock_formatter = Mock()
+ mock_handler = Mock(formatter=None)
+ with patch("logging.Formatter") as mock_formatter_init:
+ logging.basicConfig(formatter=mock_formatter, handlers=[mock_handler])
+ self.assertEqual(mock_handler.setFormatter.call_args_list, [call(mock_formatter)])
+ self.assertEqual(mock_formatter_init.call_count, 0)
+
+ def test_formatter_not_given(self):
+ mock_handler = Mock(formatter=None)
+ with patch("logging.Formatter") as mock_formatter_init:
+ logging.basicConfig(handlers=[mock_handler])
+ self.assertEqual(mock_formatter_init.call_count, 1)
+
@support.requires_working_socket()
def test_log_taskName(self):
async def log_record():
diff --git a/Lib/test/test_platform.py b/Lib/test/test_platform.py
index 818e807dd3a..3b673a47c8c 100644
--- a/Lib/test/test_platform.py
+++ b/Lib/test/test_platform.py
@@ -383,15 +383,6 @@ class PlatformTest(unittest.TestCase):
finally:
platform._uname_cache = None
- def test_java_ver(self):
- import re
- msg = re.escape(
- "'java_ver' is deprecated and slated for removal in Python 3.15"
- )
- with self.assertWarnsRegex(DeprecationWarning, msg):
- res = platform.java_ver()
- self.assertEqual(len(res), 4)
-
@unittest.skipUnless(support.MS_WINDOWS, 'This test only makes sense on Windows')
def test_win32_ver(self):
release1, version1, csd1, ptype1 = 'a', 'b', 'c', 'd'
diff --git a/Lib/test/test_string/test_templatelib.py b/Lib/test/test_string/test_templatelib.py
index 5b9490c2be6..85fcff486d6 100644
--- a/Lib/test/test_string/test_templatelib.py
+++ b/Lib/test/test_string/test_templatelib.py
@@ -148,6 +148,13 @@ class TemplateIterTests(unittest.TestCase):
self.assertEqual(res[1].format_spec, '')
self.assertEqual(res[2], ' yz')
+ def test_exhausted(self):
+ # See https://github.com/python/cpython/issues/134119.
+ template_iter = iter(t"{1}")
+ self.assertIsInstance(next(template_iter), Interpolation)
+ self.assertRaises(StopIteration, next, template_iter)
+ self.assertRaises(StopIteration, next, template_iter)
+
if __name__ == '__main__':
unittest.main()
diff --git a/Lib/test/test_syntax.py b/Lib/test/test_syntax.py
index 0eccf03a1a9..c7ac7914158 100644
--- a/Lib/test/test_syntax.py
+++ b/Lib/test/test_syntax.py
@@ -1431,6 +1431,23 @@ Better error message for using `except as` with not a name:
Traceback (most recent call last):
SyntaxError: cannot use except* statement with literal
+Regression tests for gh-133999:
+
+ >>> try: pass
+ ... except TypeError as name: raise from None
+ Traceback (most recent call last):
+ SyntaxError: invalid syntax
+
+ >>> try: pass
+ ... except* TypeError as name: raise from None
+ Traceback (most recent call last):
+ SyntaxError: invalid syntax
+
+ >>> match 1:
+ ... case 1 | 2 as abc: raise from None
+ Traceback (most recent call last):
+ SyntaxError: invalid syntax
+
Ensure that early = are not matched by the parser as invalid comparisons
>>> f(2, 4, x=34); 1 $ 2
Traceback (most recent call last):
diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py
index 8af2e3488b4..fb1c8492a64 100644
--- a/Lib/test/test_sys.py
+++ b/Lib/test/test_sys.py
@@ -2176,6 +2176,13 @@ raise Exception("Remote script exception")
with self.assertRaises(OSError):
sys.remote_exec(99999, "print('should not run')")
+ def test_remote_exec_invalid_script(self):
+ """Test remote exec with invalid script type"""
+ with self.assertRaises(TypeError):
+ sys.remote_exec(0, None)
+ with self.assertRaises(TypeError):
+ sys.remote_exec(0, 123)
+
def test_remote_exec_syntax_error(self):
"""Test remote exec with syntax error in script"""
script = '''
diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py
index 2d9649237a9..2018a20afd1 100644
--- a/Lib/test/test_tarfile.py
+++ b/Lib/test/test_tarfile.py
@@ -3490,11 +3490,12 @@ class ArchiveMaker:
with t.open() as tar:
... # `tar` is now a TarFile with 'filename' in it!
"""
- def __init__(self):
+ def __init__(self, **kwargs):
self.bio = io.BytesIO()
+ self.tar_kwargs = dict(kwargs)
def __enter__(self):
- self.tar_w = tarfile.TarFile(mode='w', fileobj=self.bio)
+ self.tar_w = tarfile.TarFile(mode='w', fileobj=self.bio, **self.tar_kwargs)
return self
def __exit__(self, *exc):
@@ -4073,7 +4074,10 @@ class TestExtractionFilters(unittest.TestCase):
# that in the test archive.)
with tarfile.TarFile.open(tarname) as tar:
for tarinfo in tar.getmembers():
- filtered = tarfile.tar_filter(tarinfo, '')
+ try:
+ filtered = tarfile.tar_filter(tarinfo, '')
+ except UnicodeEncodeError:
+ continue
self.assertIs(filtered.name, tarinfo.name)
self.assertIs(filtered.type, tarinfo.type)
@@ -4084,11 +4088,48 @@ class TestExtractionFilters(unittest.TestCase):
for tarinfo in tar.getmembers():
try:
filtered = tarfile.data_filter(tarinfo, '')
- except tarfile.FilterError:
+ except (tarfile.FilterError, UnicodeEncodeError):
continue
self.assertIs(filtered.name, tarinfo.name)
self.assertIs(filtered.type, tarinfo.type)
+ @unittest.skipIf(sys.platform == 'win32', 'requires native bytes paths')
+ def test_filter_unencodable(self):
+ # Sanity check using a valid path.
+ tarinfo = tarfile.TarInfo(os_helper.TESTFN)
+ filtered = tarfile.tar_filter(tarinfo, '')
+ self.assertIs(filtered.name, tarinfo.name)
+ filtered = tarfile.data_filter(tarinfo, '')
+ self.assertIs(filtered.name, tarinfo.name)
+
+ tarinfo = tarfile.TarInfo('test\x00')
+ self.assertRaises(ValueError, tarfile.tar_filter, tarinfo, '')
+ self.assertRaises(ValueError, tarfile.data_filter, tarinfo, '')
+ tarinfo = tarfile.TarInfo('\ud800')
+ self.assertRaises(UnicodeEncodeError, tarfile.tar_filter, tarinfo, '')
+ self.assertRaises(UnicodeEncodeError, tarfile.data_filter, tarinfo, '')
+
+ @unittest.skipIf(sys.platform == 'win32', 'requires native bytes paths')
+ def test_extract_unencodable(self):
+ # Create a member with name \xed\xa0\x80 which is UTF-8 encoded
+ # lone surrogate \ud800.
+ with ArchiveMaker(encoding='ascii', errors='surrogateescape') as arc:
+ arc.add('\udced\udca0\udc80')
+ with os_helper.temp_cwd() as tmp:
+ tar = arc.open(encoding='utf-8', errors='surrogatepass',
+ errorlevel=1)
+ self.assertEqual(tar.getnames(), ['\ud800'])
+ with self.assertRaises(UnicodeEncodeError):
+ tar.extractall()
+ self.assertEqual(os.listdir(), [])
+
+ tar = arc.open(encoding='utf-8', errors='surrogatepass',
+ errorlevel=0, debug=1)
+ with support.captured_stderr() as stderr:
+ tar.extractall()
+ self.assertEqual(os.listdir(), [])
+ self.assertIn('tarfile: UnicodeEncodeError ', stderr.getvalue())
+
def test_change_default_filter_on_instance(self):
tar = tarfile.TarFile(tarname, 'r')
def strict_filter(tarinfo, path):
diff --git a/Lib/test/test_typing.py b/Lib/test/test_typing.py
index 6ef633e4545..246be22a0d8 100644
--- a/Lib/test/test_typing.py
+++ b/Lib/test/test_typing.py
@@ -8487,6 +8487,36 @@ class TypedDictTests(BaseTestCase):
self.assertEqual(Child.__required_keys__, frozenset(['a']))
self.assertEqual(Child.__optional_keys__, frozenset())
+ def test_inheritance_pep563(self):
+ def _make_td(future, class_name, annos, base, extra_names=None):
+ lines = []
+ if future:
+ lines.append('from __future__ import annotations')
+ lines.append('from typing import TypedDict')
+ lines.append(f'class {class_name}({base}):')
+ for name, anno in annos.items():
+ lines.append(f' {name}: {anno}')
+ code = '\n'.join(lines)
+ ns = run_code(code, extra_names)
+ return ns[class_name]
+
+ for base_future in (True, False):
+ for child_future in (True, False):
+ with self.subTest(base_future=base_future, child_future=child_future):
+ base = _make_td(
+ base_future, "Base", {"base": "int"}, "TypedDict"
+ )
+ self.assertIsNotNone(base.__annotate__)
+ child = _make_td(
+ child_future, "Child", {"child": "int"}, "Base", {"Base": base}
+ )
+ base_anno = ForwardRef("int", module="builtins") if base_future else int
+ child_anno = ForwardRef("int", module="builtins") if child_future else int
+ self.assertEqual(base.__annotations__, {'base': base_anno})
+ self.assertEqual(
+ child.__annotations__, {'child': child_anno, 'base': base_anno}
+ )
+
def test_required_notrequired_keys(self):
self.assertEqual(NontotalMovie.__required_keys__,
frozenset({"title"}))
diff --git a/Lib/test/test_wave.py b/Lib/test/test_wave.py
index 5e771c8de96..6c3362857fc 100644
--- a/Lib/test/test_wave.py
+++ b/Lib/test/test_wave.py
@@ -136,32 +136,6 @@ class MiscTestCase(unittest.TestCase):
not_exported = {'WAVE_FORMAT_PCM', 'WAVE_FORMAT_EXTENSIBLE', 'KSDATAFORMAT_SUBTYPE_PCM'}
support.check__all__(self, wave, not_exported=not_exported)
- def test_read_deprecations(self):
- filename = support.findfile('pluck-pcm8.wav', subdir='audiodata')
- with wave.open(filename) as reader:
- with self.assertWarns(DeprecationWarning):
- with self.assertRaises(wave.Error):
- reader.getmark('mark')
- with self.assertWarns(DeprecationWarning):
- self.assertIsNone(reader.getmarkers())
-
- def test_write_deprecations(self):
- with io.BytesIO(b'') as tmpfile:
- with wave.open(tmpfile, 'wb') as writer:
- writer.setnchannels(1)
- writer.setsampwidth(1)
- writer.setframerate(1)
- writer.setcomptype('NONE', 'not compressed')
-
- with self.assertWarns(DeprecationWarning):
- with self.assertRaises(wave.Error):
- writer.setmark(0, 0, 'mark')
- with self.assertWarns(DeprecationWarning):
- with self.assertRaises(wave.Error):
- writer.getmark('mark')
- with self.assertWarns(DeprecationWarning):
- self.assertIsNone(writer.getmarkers())
-
class WaveLowLevelTest(unittest.TestCase):