diff --git a/Lib/test/test_android.py b/Lib/test/test_android.py index 076190f75..31daafbc3 100644 --- a/Lib/test/test_android.py +++ b/Lib/test/test_android.py @@ -1,5 +1,4 @@ import io -import platform import queue import re import subprocess @@ -17,17 +16,11 @@ from unittest.mock import patch if sys.platform != "android": raise unittest.SkipTest("Android-specific") -api_level = platform.android_ver().api_level - # (name, level, fileno) STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)] # Test redirection of stdout and stderr to the Android log. -@unittest.skipIf( - api_level < 23 and platform.machine() == "aarch64", - "SELinux blocks reading logs on older ARM64 emulators" -) class TestAndroidOutput(unittest.TestCase): maxDiff = None @@ -42,31 +35,41 @@ class TestAndroidOutput(unittest.TestCase): for line in self.logcat_process.stdout: self.logcat_queue.put(line.rstrip("\n")) self.logcat_process.stdout.close() + self.logcat_thread = Thread(target=logcat_thread) self.logcat_thread.start() - from ctypes import CDLL, c_char_p, c_int - android_log_write = getattr(CDLL("liblog.so"), "__android_log_write") - android_log_write.argtypes = (c_int, c_char_p, c_char_p) - ANDROID_LOG_INFO = 4 + try: + from ctypes import CDLL, c_char_p, c_int + android_log_write = getattr(CDLL("liblog.so"), "__android_log_write") + android_log_write.argtypes = (c_int, c_char_p, c_char_p) + ANDROID_LOG_INFO = 4 - # Separate tests using a marker line with a different tag. - tag, message = "python.test", f"{self.id()} {time()}" - android_log_write( - ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8")) - self.assert_log("I", tag, message, skip=True, timeout=5) + # Separate tests using a marker line with a different tag. + tag, message = "python.test", f"{self.id()} {time()}" + android_log_write( + ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8")) + self.assert_log("I", tag, message, skip=True) + except: + # If setUp throws an exception, tearDown is not automatically + # called. Avoid leaving a dangling thread which would keep the + # Python process alive indefinitely. + self.tearDown() + raise def assert_logs(self, level, tag, expected, **kwargs): for line in expected: self.assert_log(level, tag, line, **kwargs) - def assert_log(self, level, tag, expected, *, skip=False, timeout=0.5): - deadline = time() + timeout + def assert_log(self, level, tag, expected, *, skip=False): + deadline = time() + LOOPBACK_TIMEOUT while True: try: line = self.logcat_queue.get(timeout=(deadline - time())) except queue.Empty: - self.fail(f"line not found: {expected!r}") + raise self.failureException( + f"line not found: {expected!r}" + ) from None if match := re.fullmatch(fr"(.)/{tag}: (.*)", line): try: self.assertEqual(level, match[1]) @@ -81,35 +84,42 @@ class TestAndroidOutput(unittest.TestCase): self.logcat_process.wait(LOOPBACK_TIMEOUT) self.logcat_thread.join(LOOPBACK_TIMEOUT) + # Avoid an irrelevant warning about threading._dangling. + self.logcat_thread = None + @contextmanager - def unbuffered(self, stream): - stream.reconfigure(write_through=True) + def reconfigure(self, stream, **settings): + original_settings = {key: getattr(stream, key, None) for key in settings.keys()} + stream.reconfigure(**settings) try: yield finally: - stream.reconfigure(write_through=False) + stream.reconfigure(**original_settings) - # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't - # test them directly. Detect this mode and use some temporary streams with - # the same properties. def stream_context(self, stream_name, level): - # https://developer.android.com/ndk/reference/group/logging - prio = {"I": 4, "W": 5}[level] - stack = ExitStack() stack.enter_context(self.subTest(stream_name)) + + # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't + # test them directly. Detect this mode and use some temporary streams with + # the same properties. stream = getattr(sys, stream_name) native_stream = getattr(sys, f"__{stream_name}__") if isinstance(stream, io.StringIO): + # https://developer.android.com/ndk/reference/group/logging + prio = {"I": 4, "W": 5}[level] stack.enter_context( patch( f"sys.{stream_name}", - TextLogStream( - prio, f"python.{stream_name}", native_stream.fileno(), - errors="backslashreplace" + stream := TextLogStream( + prio, f"python.{stream_name}", native_stream, ), ) ) + + # The tests assume the stream is initially buffered. + stack.enter_context(self.reconfigure(stream, write_through=False)) + return stack def test_str(self): @@ -136,7 +146,7 @@ class TestAndroidOutput(unittest.TestCase): self.assert_logs(level, tag, lines) # Single-line messages, - with self.unbuffered(stream): + with self.reconfigure(stream, write_through=True): write("", []) write("a") @@ -166,14 +176,18 @@ class TestAndroidOutput(unittest.TestCase): # Multi-line messages. Avoid identical consecutive lines, as # they may activate "chatty" filtering and break the tests. - write("\nx", [""]) + # + # Additional spaces will appear in the output where necessary to + # protect leading newlines. + write("\nx", [" "]) write("\na\n", ["x", "a"]) - write("\n", [""]) + write("\n", [" "]) + write("\n\n", [" ", " "]) write("b\n", ["b"]) - write("c\n\n", ["c", ""]) + write("c\n\n", ["c", " "]) write("d\ne", ["d"]) write("xx", []) - write("f\n\ng", ["exxf", ""]) + write("f\n\ng", ["exxf", " "]) write("\n", ["g"]) # Since this is a line-based logging system, line buffering @@ -183,16 +197,17 @@ class TestAndroidOutput(unittest.TestCase): # However, buffering can be turned off completely if you want a # flush after every write. - with self.unbuffered(stream): - write("\nx", ["", "x"]) - write("\na\n", ["", "a"]) - write("\n", [""]) + with self.reconfigure(stream, write_through=True): + write("\nx", [" ", "x"]) + write("\na\n", [" ", "a"]) + write("\n", [" "]) + write("\n\n", [" ", " "]) write("b\n", ["b"]) - write("c\n\n", ["c", ""]) + write("c\n\n", ["c", " "]) write("d\ne", ["d", "e"]) write("xx", ["xx"]) - write("f\n\ng", ["f", "", "g"]) - write("\n", [""]) + write("f\n\ng", ["f", " ", "g"]) + write("\n", [" "]) # "\r\n" should be translated into "\n". write("hello\r\n", ["hello"]) @@ -312,19 +327,16 @@ class TestAndroidOutput(unittest.TestCase): # currently use `logcat -v tag`, which shows each line as if it # was a separate log entry, but strips a single trailing # newline. - # - # On newer versions of Android, all three of the above tools (or - # maybe Logcat itself) will also strip any number of leading - # newlines. - write(b"\nx", ["", "x"] if api_level < 30 else ["x"]) - write(b"\na\n", ["", "a"] if api_level < 30 else ["a"]) - write(b"\n", [""]) + write(b"\nx", [" ", "x"]) + write(b"\na\n", [" ", "a"]) + write(b"\n", [" "]) + write(b"\n\n", [" ", ""]) write(b"b\n", ["b"]) write(b"c\n\n", ["c", ""]) write(b"d\ne", ["d", "e"]) write(b"xx", ["xx"]) write(b"f\n\ng", ["f", "", "g"]) - write(b"\n", [""]) + write(b"\n", [" "]) # "\r\n" should be translated into "\n". write(b"hello\r\n", ["hello"]) diff --git a/Lib/test/test_binascii.py b/Lib/test/test_binascii.py index b8d350a89..cd8acb4c2 100644 --- a/Lib/test/test_binascii.py +++ b/Lib/test/test_binascii.py @@ -222,6 +222,7 @@ class BinASCIITest(unittest.TestCase): assertInvalidLength(b'a' * (4 * 87 + 1)) assertInvalidLength(b'A\tB\nC ??DE') # only 5 valid characters + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: Error not raised by a2b_uu def test_uu(self): MAX_UU = 45 for backtick in (True, False): @@ -242,6 +243,10 @@ class BinASCIITest(unittest.TestCase): self.assertEqual(binascii.a2b_uu(b"\xff"), b"\x00"*31) self.assertRaises(binascii.Error, binascii.a2b_uu, b"\xff\x00") self.assertRaises(binascii.Error, binascii.a2b_uu, b"!!!!") + self.assertRaises(binascii.Error, binascii.a2b_uu, + self.type2test(b"")) + self.assertRaises(binascii.Error, binascii.a2b_uu, + self.type2test(b"#86)C")[:0]) self.assertRaises(binascii.Error, binascii.b2a_uu, 46*b"!") # Issue #7701 (crash on a pydebug build) @@ -440,6 +445,7 @@ class BinASCIITest(unittest.TestCase): self.assertConversion(binary, converted, restored, quotetabs=quotetabs, istext=istext, header=header) + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: Error not raised by a2b_uu def test_empty_string(self): # A test for SF bug #1022953. Make sure SystemError is not raised. empty = self.type2test(b'') @@ -449,6 +455,9 @@ class BinASCIITest(unittest.TestCase): binascii.crc_hqx(empty, 0) continue f = getattr(binascii, func) + if func == 'a2b_uu': + self.assertRaises(binascii.Error, f, empty) + continue try: f(empty) except Exception as err: diff --git a/Lib/test/test_builtin.py b/Lib/test/test_builtin.py index 13b0d4a2a..163ebcfb5 100644 --- a/Lib/test/test_builtin.py +++ b/Lib/test/test_builtin.py @@ -293,6 +293,27 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase): self.assertEqual(overridden_outputs, ['all', 'any', 'tuple']) + def test_builtin_call_async_genexpr_no_crash(self): + async def f_all(): + return all(await 2 for _ in []) + + async def f_any(): + return any(await 2 for _ in []) + + async def f_tuple(): + return tuple(await 2 for _ in []) + + async def f_list(): + return list(await 2 for _ in []) + + async def f_set(): + return set(await 2 for _ in []) + + for f in (f_all, f_any, f_tuple, f_list, f_set): + with self.subTest(func=f.__name__): + with self.assertRaises(TypeError): + run_yielding_async_fn(f) + def test_ascii(self): self.assertEqual(ascii(''), '\'\'') self.assertEqual(ascii(0), '0') @@ -606,7 +627,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase): exec(co, glob) self.assertEqual(type(glob['ticker']()), AsyncGeneratorType) - @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: <_ast.Name object at 0xb40000731e3d1360> is not an instance of + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: <_ast.Name object at 0xb40000731e3d1360> is not an instance of def test_compile_ast(self): args = ("a*__debug__", "f.py", "exec") raw = compile(*args, flags = ast.PyCF_ONLY_AST).body[0] @@ -970,7 +991,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase): self.assertRaisesRegex(NameError, "name 'superglobal' is not defined", eval, code, ns) - @unittest.expectedFailure # TODO: RUSTPYTHON; wrong error message + @unittest.expectedFailure # TODO: RUSTPYTHON; wrong error message def test_exec_builtins_mapping_import(self): code = compile("import foo.bar", "test", "exec") ns = {'__builtins__': types.MappingProxyType({})} @@ -979,7 +1000,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase): exec(code, ns) self.assertEqual(ns['foo'], ('foo.bar', ns, ns, None, 0)) - @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: AttributeError not raised by eval + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: AttributeError not raised by eval def test_eval_builtins_mapping_reduce(self): # list_iterator.__reduce__() calls _PyEval_GetBuiltin("iter") code = compile("x.__reduce__()", "test", "eval") diff --git a/Lib/test/test_defaultdict.py b/Lib/test/test_defaultdict.py index fbd7354a9..a193eb10f 100644 --- a/Lib/test/test_defaultdict.py +++ b/Lib/test/test_defaultdict.py @@ -204,5 +204,20 @@ class TestDefaultDict(unittest.TestCase): self.assertEqual(test_dict[key], 2) self.assertEqual(count, 2) + def test_repr_recursive_factory(self): + # gh-145492: defaultdict.__repr__ should not cause infinite recursion + # when the factory's __repr__ calls repr() on the defaultdict. + class ProblematicFactory: + def __call__(self): + return {} + def __repr__(self): + repr(dd) + return f"ProblematicFactory for {dd}" + + dd = defaultdict(ProblematicFactory()) + # Should not raise RecursionError + r = repr(dd) + self.assertIn("ProblematicFactory for", r) + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_mmap.py b/Lib/test/test_mmap.py index ed96be53c..6b67c9e50 100644 --- a/Lib/test/test_mmap.py +++ b/Lib/test/test_mmap.py @@ -1,9 +1,12 @@ from test.support import ( - requires, _2G, _4G, gc_collect, cpython_only, is_emscripten + requires, _2G, _4G, gc_collect, cpython_only, is_emscripten, is_apple, + in_systemd_nspawn_sync_suppressed, ) from test.support.import_helper import import_module from test.support.os_helper import TESTFN, unlink +from test.support.script_helper import assert_python_ok import unittest +import errno import os import re import itertools @@ -11,6 +14,7 @@ import random import socket import string import sys +import textwrap import weakref # Skip test if we can't import mmap. @@ -32,6 +36,10 @@ if is_emscripten: class MmapTests(unittest.TestCase): def setUp(self): + # TODO: RUSTPYTHON; Remove this once windows doesn't get errored on setup:/ + if os.name == "nt": + raise unittest.SkipTest("TODO: RUSTPYTHON; Error during class setUp") + if os.path.exists(TESTFN): os.unlink(TESTFN) @@ -41,6 +49,7 @@ class MmapTests(unittest.TestCase): except OSError: pass + @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: 'mmap' object has no attribute 'seekable' def test_basic(self): # Test mmap module on Unix systems and Windows @@ -93,11 +102,12 @@ class MmapTests(unittest.TestCase): self.assertEqual(end, PAGESIZE + 6) # test seeking around (try to overflow the seek implementation) - m.seek(0,0) + self.assertTrue(m.seekable()) + self.assertEqual(m.seek(0, 0), 0) self.assertEqual(m.tell(), 0) - m.seek(42,1) + self.assertEqual(m.seek(42, 1), 42) self.assertEqual(m.tell(), 42) - m.seek(0,2) + self.assertEqual(m.seek(0, 2), len(m)) self.assertEqual(m.tell(), len(m)) # Try to seek to negative position... @@ -162,7 +172,7 @@ class MmapTests(unittest.TestCase): # Ensuring that readonly mmap can't be write() to try: - m.seek(0,0) + m.seek(0, 0) m.write(b'abc') except TypeError: pass @@ -171,7 +181,7 @@ class MmapTests(unittest.TestCase): # Ensuring that readonly mmap can't be write_byte() to try: - m.seek(0,0) + m.seek(0, 0) m.write_byte(b'd') except TypeError: pass @@ -255,15 +265,79 @@ class MmapTests(unittest.TestCase): # Try writing with PROT_EXEC and without PROT_WRITE prot = mmap.PROT_READ | getattr(mmap, 'PROT_EXEC', 0) with open(TESTFN, "r+b") as f: - m = mmap.mmap(f.fileno(), mapsize, prot=prot) - self.assertRaises(TypeError, m.write, b"abcdef") - self.assertRaises(TypeError, m.write_byte, 0) - m.close() + try: + m = mmap.mmap(f.fileno(), mapsize, prot=prot) + except PermissionError: + # on macOS 14, PROT_READ | PROT_EXEC is not allowed + pass + else: + self.assertRaises(TypeError, m.write, b"abcdef") + self.assertRaises(TypeError, m.write_byte, 0) + m.close() + + @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows') + def test_trackfd_parameter(self): + size = 64 + with open(TESTFN, "wb") as f: + f.write(b"a"*size) + for close_original_fd in True, False: + with self.subTest(close_original_fd=close_original_fd): + with open(TESTFN, "r+b") as f: + with mmap.mmap(f.fileno(), size, trackfd=False) as m: + if close_original_fd: + f.close() + self.assertEqual(len(m), size) + with self.assertRaises(OSError) as err_cm: + m.size() + self.assertEqual(err_cm.exception.errno, errno.EBADF) + with self.assertRaises(ValueError): + m.resize(size * 2) + with self.assertRaises(ValueError): + m.resize(size // 2) + self.assertEqual(m.closed, False) + + # Smoke-test other API + m.write_byte(ord('X')) + m[2] = ord('Y') + m.flush() + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(4), b'XaYa') + self.assertEqual(m.tell(), 1) + m.seek(0) + self.assertEqual(m.tell(), 0) + self.assertEqual(m.read_byte(), ord('X')) + + self.assertEqual(m.closed, True) + self.assertEqual(os.stat(TESTFN).st_size, size) + + @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows') + def test_trackfd_neg1(self): + size = 64 + with mmap.mmap(-1, size, trackfd=False) as m: + with self.assertRaises(OSError): + m.size() + with self.assertRaises(ValueError): + m.resize(size // 2) + self.assertEqual(len(m), size) + m[0] = ord('a') + assert m[0] == ord('a') + + @unittest.skipIf(os.name != 'nt', 'trackfd only fails on Windows') + def test_no_trackfd_parameter_on_windows(self): + # 'trackffd' is an invalid keyword argument for this function + size = 64 + with self.assertRaises(TypeError): + mmap.mmap(-1, size, trackfd=True) + with self.assertRaises(TypeError): + mmap.mmap(-1, size, trackfd=False) def test_bad_file_desc(self): # Try opening a bad file descriptor... self.assertRaises(OSError, mmap.mmap, -2, 4096) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_tougher_find(self): # Do a tougher .find() test. SF bug 515943 pointed out that, in 2.2, # searching for data with embedded \0 bytes didn't work. @@ -282,6 +356,7 @@ class MmapTests(unittest.TestCase): self.assertEqual(m.find(slice + b'x'), -1) m.close() + @unittest.skip("TODO: RUSTPYTHON; panic") def test_find_end(self): # test the new 'end' parameter works as expected with open(TESTFN, 'wb+') as f: @@ -299,7 +374,29 @@ class MmapTests(unittest.TestCase): self.assertEqual(m.find(b'one', 1, -2), -1) self.assertEqual(m.find(bytearray(b'one')), 0) + for i in range(-n-1, n+1): + for j in range(-n-1, n+1): + for p in [b"o", b"on", b"two", b"ones", b"s"]: + expected = data.find(p, i, j) + self.assertEqual(m.find(p, i, j), expected, (p, i, j)) + def test_find_does_not_access_beyond_buffer(self): + try: + flags = mmap.MAP_PRIVATE | mmap.MAP_ANONYMOUS + PAGESIZE = mmap.PAGESIZE + PROT_NONE = 0 + PROT_READ = mmap.PROT_READ + except AttributeError as e: + raise unittest.SkipTest("mmap flags unavailable") from e + for i in range(0, 2049): + with mmap.mmap(-1, PAGESIZE * (i + 1), + flags=flags, prot=PROT_NONE) as guard: + with mmap.mmap(-1, PAGESIZE * (i + 2048), + flags=flags, prot=PROT_READ) as fm: + fm.find(b"fo", -2) + + + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_rfind(self): # test the new 'end' parameter works as expected with open(TESTFN, 'wb+') as f: @@ -360,6 +457,7 @@ class MmapTests(unittest.TestCase): self.assertRaises(ValueError, mmap.mmap, f.fileno(), 0, offset=2147418112) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_move(self): # make move works everywhere (64-bit format problem earlier) with open(TESTFN, 'wb+') as f: @@ -407,7 +505,6 @@ class MmapTests(unittest.TestCase): m.move(0, 0, 1) m.move(0, 0, 0) - def test_anonymous(self): # anonymous mmap.mmap(-1, PAGE) m = mmap.mmap(-1, PAGESIZE) @@ -420,6 +517,7 @@ class MmapTests(unittest.TestCase): m[x] = b self.assertEqual(m[x], b) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_read_all(self): m = mmap.mmap(-1, 16) self.addCleanup(m.close) @@ -441,6 +539,7 @@ class MmapTests(unittest.TestCase): m.seek(9) self.assertEqual(m.read(-42), bytes(range(9, 16))) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_read_invalid_arg(self): m = mmap.mmap(-1, 16) self.addCleanup(m.close) @@ -557,6 +656,7 @@ class MmapTests(unittest.TestCase): except OSError: pass + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_subclass(self): class anon_mmap(mmap.mmap): def __new__(klass, *args, **kwargs): @@ -609,6 +709,7 @@ class MmapTests(unittest.TestCase): self.assertEqual(m[:], b"012barbaz9") self.assertRaises(ValueError, m.write, b"ba") + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_non_ascii_byte(self): for b in (129, 200, 255): # > 128 m = mmap.mmap(-1, 1) @@ -618,6 +719,7 @@ class MmapTests(unittest.TestCase): self.assertEqual(m.read_byte(), b) m.close() + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") @unittest.skipUnless(os.name == 'nt', 'requires Windows') def test_tagname(self): data1 = b"0123456789" @@ -646,14 +748,16 @@ class MmapTests(unittest.TestCase): m2.close() m1.close() + with self.assertRaisesRegex(TypeError, 'tagname'): + mmap.mmap(-1, 8, tagname=1) + @cpython_only @unittest.skipUnless(os.name == 'nt', 'requires Windows') def test_sizeof(self): m1 = mmap.mmap(-1, 100) tagname = random_tagname() m2 = mmap.mmap(-1, 100, tagname=tagname) - self.assertEqual(sys.getsizeof(m2), - sys.getsizeof(m1) + len(tagname) + 1) + self.assertGreater(sys.getsizeof(m2), sys.getsizeof(m1)) @unittest.skipUnless(os.name == 'nt', 'requires Windows') def test_crasher_on_windows(self): @@ -708,6 +812,7 @@ class MmapTests(unittest.TestCase): "wrong exception raised in context manager") self.assertTrue(m.closed, "context manager failed") + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_weakref(self): # Check mmap objects are weakrefable mm = mmap.mmap(-1, 16) @@ -717,6 +822,7 @@ class MmapTests(unittest.TestCase): gc_collect() self.assertIs(wr(), None) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_write_returning_the_number_of_bytes_written(self): mm = mmap.mmap(-1, 16) self.assertEqual(mm.write(b""), 0) @@ -724,6 +830,7 @@ class MmapTests(unittest.TestCase): self.assertEqual(mm.write(b"yz"), 2) self.assertEqual(mm.write(b"python"), 6) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_resize_past_pos(self): m = mmap.mmap(-1, 8192) self.addCleanup(m.close) @@ -744,7 +851,7 @@ class MmapTests(unittest.TestCase): with self.assertRaises(TypeError): m * 2 - @unittest.skipIf(sys.platform.startswith("linux"), "TODO: RUSTPYTHON, memmap2 doesn't throw OSError when offset is not a multiple of mmap.PAGESIZE on Linux") + @unittest.skipIf(sys.platform.startswith("linux"), "TODO: RUSTPYTHON; memmap2 doesn't throw OSError when offset is not a multiple of mmap.PAGESIZE on Linux") def test_flush_return_value(self): # mm.flush() should return None on success, raise an # exception on error under all platforms. @@ -753,11 +860,13 @@ class MmapTests(unittest.TestCase): mm.write(b'python') result = mm.flush() self.assertIsNone(result) - if sys.platform.startswith('linux'): + if (sys.platform.startswith(('linux', 'android')) + and not in_systemd_nspawn_sync_suppressed()): # 'offset' must be a multiple of mmap.PAGESIZE on Linux. # See bpo-34754 for details. self.assertRaises(OSError, mm.flush, 1, len(b'python')) + @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON") def test_repr(self): open_mmap_repr_pat = re.compile( r" None: + self.access = access + + def __enter__(self): + self.f = open(TESTFN, "w+b") + self.f.write(random.randbytes(100)) + self.f.flush() + + m = mmap.mmap(self.f.fileno(), 100, access=self.access) + + class X: + def __index__(self): + m.close() + return 10 + + return (m, X) + + def __exit__(self, exc_type, exc_value, traceback): + self.f.close() + + read_access_modes = [ + mmap.ACCESS_READ, + mmap.ACCESS_WRITE, + mmap.ACCESS_COPY, + mmap.ACCESS_DEFAULT, + ] + + write_access_modes = [ + mmap.ACCESS_WRITE, + mmap.ACCESS_COPY, + mmap.ACCESS_DEFAULT, + ] + + for access in read_access_modes: + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[X()] + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[X() : 20] + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[X() : 20 : 2] + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[20 : X() : -2] + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m.read(X()) + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m.find(b"1", 1, X()) + + for access in write_access_modes: + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[X() : 20] = b"1" * 10 + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[X() : 20 : 2] = b"1" * 5 + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m[20 : X() : -2] = b"1" * 5 + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m.move(1, 2, X()) + + with MmapClosedByIntContext(access) as (m, X): + with self.assertRaisesRegex(ValueError, "mmap closed or invalid"): + m.write_byte(X()) + + @unittest.skipUnless(os.name == 'nt', 'requires Windows') + @unittest.skipUnless(hasattr(mmap.mmap, '_protect'), 'test needs debug build') + def test_access_violations(self): + from test.support.os_helper import TESTFN + + code = textwrap.dedent(""" + import faulthandler + import mmap + import os + import sys + from contextlib import suppress + + # Prevent logging access violations to stderr. + faulthandler.disable() + + PAGESIZE = mmap.PAGESIZE + PAGE_NOACCESS = 0x01 + + with open(sys.argv[1], 'bw+') as f: + f.write(b'A'* PAGESIZE) + f.flush() + + m = mmap.mmap(f.fileno(), PAGESIZE) + m._protect(PAGE_NOACCESS, 0, PAGESIZE) + with suppress(OSError): + m.read(PAGESIZE) + assert False, 'mmap.read() did not raise' + with suppress(OSError): + m.read_byte() + assert False, 'mmap.read_byte() did not raise' + with suppress(OSError): + m.readline() + assert False, 'mmap.readline() did not raise' + with suppress(OSError): + m.write(b'A'* PAGESIZE) + assert False, 'mmap.write() did not raise' + with suppress(OSError): + m.write_byte(0) + assert False, 'mmap.write_byte() did not raise' + with suppress(OSError): + m[0] # test mmap_subscript + assert False, 'mmap.__getitem__() did not raise' + with suppress(OSError): + m[0:10] # test mmap_subscript + assert False, 'mmap.__getitem__() did not raise' + with suppress(OSError): + m[0:10:2] # test mmap_subscript + assert False, 'mmap.__getitem__() did not raise' + with suppress(OSError): + m[0] = 1 + assert False, 'mmap.__setitem__() did not raise' + with suppress(OSError): + m[0:10] = b'A'* 10 + assert False, 'mmap.__setitem__() did not raise' + with suppress(OSError): + m[0:10:2] = b'A'* 5 + assert False, 'mmap.__setitem__() did not raise' + with suppress(OSError): + m.move(0, 10, 1) + assert False, 'mmap.move() did not raise' + with suppress(OSError): + list(m) # test mmap_item + assert False, 'mmap.__getitem__() did not raise' + with suppress(OSError): + m.find(b'A') + assert False, 'mmap.find() did not raise' + with suppress(OSError): + m.rfind(b'A') + assert False, 'mmap.rfind() did not raise' + """) + rt, stdout, stderr = assert_python_ok("-c", code, TESTFN) + self.assertEqual(stdout.strip(), b'') + self.assertEqual(stderr.strip(), b'') + + class LargeMmapTests(unittest.TestCase): def setUp(self): @@ -897,7 +1216,7 @@ class LargeMmapTests(unittest.TestCase): unlink(TESTFN) def _make_test_file(self, num_zeroes, tail): - if sys.platform[:3] == 'win' or sys.platform == 'darwin': + if sys.platform[:3] == 'win' or is_apple: requires('largefile', 'test requires %s bytes and a long time to run' % str(0x180000000)) f = open(TESTFN, 'w+b') diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py index eb0e4d25d..d7e1c00f1 100644 --- a/Lib/test/test_posix.py +++ b/Lib/test/test_posix.py @@ -1934,6 +1934,11 @@ class _PosixSpawnMixin: ) support.wait_process(pid, exitcode=0) + def test_setpgroup_allow_none(self): + path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM + pid = self.spawn_func(path, args, os.environ, setpgroup=None) + support.wait_process(pid, exitcode=0) + def test_setpgroup_wrong_type(self): with self.assertRaises(TypeError): self.spawn_func(sys.executable, @@ -2034,6 +2039,21 @@ class _PosixSpawnMixin: [sys.executable, "-c", "pass"], os.environ, setsigdef=[signal.NSIG, signal.NSIG+1]) + def test_scheduler_allow_none(self): + path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM + pid = self.spawn_func(path, args, os.environ, scheduler=None) + support.wait_process(pid, exitcode=0) + + @unittest.expectedFailure # TODO: RUSTPYTHON; Wrong error message + @support.subTests("scheduler", [object(), 1, [1, 2]]) + def test_scheduler_wrong_type(self, scheduler): + path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM + with self.assertRaisesRegex( + TypeError, + "scheduler must be a tuple or None", + ): + self.spawn_func(path, args, os.environ, scheduler=scheduler) + @unittest.expectedFailureIf(sys.platform in ("darwin", "linux"), "TODO: RUSTPYTHON; NotImplementedError: scheduler parameter is not yet implemented") @requires_sched @unittest.skipIf(sys.platform.startswith(('freebsd', 'netbsd')), diff --git a/Lib/test/test_uuid.py b/Lib/test/test_uuid.py index 0e1a723ce..c8400da3a 100644 --- a/Lib/test/test_uuid.py +++ b/Lib/test/test_uuid.py @@ -1176,6 +1176,47 @@ class CommandLineTestCases: self.assertEqual(cm.exception.code, 2) self.assertIn("error: Incorrect number of arguments", mock_err.getvalue()) + @mock.patch.object(sys, "argv", + ["", "-u", "uuid3", "-n", "@dns", "-N", "python.org"]) + def test_cli_uuid3_outputted_with_valid_namespace_and_name(self): + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self.uuid.main() + + output = stdout.getvalue().strip() + uuid_output = self.uuid.UUID(output) + + # Output should be in the form of uuid3 + self.assertEqual(output, str(uuid_output)) + self.assertEqual(uuid_output.version, 3) + + @mock.patch.object(sys, "argv", + ["", "-u", "uuid3", "-n", + "0d6a16cc-34a7-47d8-b660-214d0ae184d2", + "-N", "some.user"]) + def test_cli_uuid3_outputted_with_custom_namespace_and_name(self): + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self.uuid.main() + + output = stdout.getvalue().strip() + uuid_output = self.uuid.UUID(output) + + # Output should be in the form of uuid3 + self.assertEqual(output, str(uuid_output)) + self.assertEqual(uuid_output.version, 3) + + @mock.patch.object(sys, "argv", + ["", "-u", "uuid3", "-n", "any UUID", "-N", "python.org"]) + @mock.patch('sys.stderr', new_callable=io.StringIO) + def test_cli_uuid3_with_invalid_namespace(self, mock_err): + with self.assertRaises(SystemExit) as cm: + self.uuid.main() + # Check that exception code is the same as argparse.ArgumentParser.error + self.assertEqual(cm.exception.code, 2) + self.assertIn("error: badly formed hexadecimal UUID string", + mock_err.getvalue()) + @mock.patch.object(sys, "argv", [""]) def test_cli_uuid4_outputted_with_no_args(self): stdout = io.StringIO() @@ -1203,23 +1244,9 @@ class CommandLineTestCases: uuid_output = self.uuid.UUID(o) self.assertEqual(uuid_output.version, 4) - @mock.patch.object(sys, "argv", - ["", "-u", "uuid3", "-n", "@dns", "-N", "python.org"]) - def test_cli_uuid3_ouputted_with_valid_namespace_and_name(self): - stdout = io.StringIO() - with contextlib.redirect_stdout(stdout): - self.uuid.main() - - output = stdout.getvalue().strip() - uuid_output = self.uuid.UUID(output) - - # Output should be in the form of uuid5 - self.assertEqual(output, str(uuid_output)) - self.assertEqual(uuid_output.version, 3) - @mock.patch.object(sys, "argv", ["", "-u", "uuid5", "-n", "@dns", "-N", "python.org"]) - def test_cli_uuid5_ouputted_with_valid_namespace_and_name(self): + def test_cli_uuid5_outputted_with_valid_namespace_and_name(self): stdout = io.StringIO() with contextlib.redirect_stdout(stdout): self.uuid.main() @@ -1231,6 +1258,33 @@ class CommandLineTestCases: self.assertEqual(output, str(uuid_output)) self.assertEqual(uuid_output.version, 5) + @mock.patch.object(sys, "argv", + ["", "-u", "uuid5", "-n", + "0d6a16cc-34a7-47d8-b660-214d0ae184d2", + "-N", "some.user"]) + def test_cli_uuid5_ouputted_with_custom_namespace_and_name(self): + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self.uuid.main() + + output = stdout.getvalue().strip() + uuid_output = self.uuid.UUID(output) + + # Output should be in the form of uuid5 + self.assertEqual(output, str(uuid_output)) + self.assertEqual(uuid_output.version, 5) + + @mock.patch.object(sys, "argv", + ["", "-u", "uuid5", "-n", "any UUID", "-N", "python.org"]) + @mock.patch('sys.stderr', new_callable=io.StringIO) + def test_cli_uuid5_with_invalid_namespace(self, mock_err): + with self.assertRaises(SystemExit) as cm: + self.uuid.main() + # Check that exception code is the same as argparse.ArgumentParser.error + self.assertEqual(cm.exception.code, 2) + self.assertIn("error: badly formed hexadecimal UUID string", + mock_err.getvalue()) + @mock.patch.object(sys, "argv", ["", "-u", "uuid6"]) def test_cli_uuid6(self): self.do_test_standalone_uuid(6) diff --git a/Lib/uuid.py b/Lib/uuid.py index 313f2fc46..043ad7bcb 100644 --- a/Lib/uuid.py +++ b/Lib/uuid.py @@ -961,7 +961,7 @@ def main(): default="uuid4", help="function to generate the UUID") parser.add_argument("-n", "--namespace", - choices=["any UUID", *namespaces.keys()], + metavar=f"{{any UUID,{','.join(namespaces)}}}", help="uuid3/uuid5 only: " "a UUID, or a well-known predefined UUID addressed " "by namespace name") @@ -983,7 +983,13 @@ def main(): f"{args.uuid} requires a namespace and a name. " "Run 'python -m uuid -h' for more information." ) - namespace = namespaces[namespace] if namespace in namespaces else UUID(namespace) + if namespace in namespaces: + namespace = namespaces[namespace] + else: + try: + namespace = UUID(namespace) + except ValueError as exc: + parser.error(f"{exc}: {args.namespace!r}") for _ in range(args.count): print(uuid_func(namespace, name)) else: