diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/test_external_inspection.py | 122 | ||||
| -rw-r--r-- | Lib/test/test_iter.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_listcomps.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_math.py | 15 | ||||
| -rw-r--r-- | Lib/test/test_platform.py | 16 |
5 files changed, 152 insertions, 5 deletions
diff --git a/Lib/test/test_external_inspection.py b/Lib/test/test_external_inspection.py index 90214e814f2..0f31c225e68 100644 --- a/Lib/test/test_external_inspection.py +++ b/Lib/test/test_external_inspection.py @@ -7,7 +7,7 @@ import socket import threading from asyncio import staggered, taskgroups, base_events, tasks from unittest.mock import ANY -from test.support import os_helper, SHORT_TIMEOUT, busy_retry +from test.support import os_helper, SHORT_TIMEOUT, busy_retry, requires_gil_enabled from test.support.script_helper import make_script from test.support.socket_helper import find_unused_port @@ -876,6 +876,126 @@ class TestGetStackTrace(unittest.TestCase): ], ) + @skip_if_not_supported + @unittest.skipIf( + sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED, + "Test only runs on Linux with process_vm_readv support", + ) + @requires_gil_enabled("Free threaded builds don't have an 'active thread'") + def test_only_active_thread(self): + # Test that only_active_thread parameter works correctly + port = find_unused_port() + script = textwrap.dedent( + f"""\ + import time, sys, socket, threading + + # Connect to the test process + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect(('localhost', {port})) + + def worker_thread(name, barrier, ready_event): + barrier.wait() # Synchronize thread start + ready_event.wait() # Wait for main thread signal + # Sleep to keep thread alive + time.sleep(10_000) + + def main_work(): + # Do busy work to hold the GIL + sock.sendall(b"working\\n") + count = 0 + while count < 100000000: + count += 1 + if count % 10000000 == 0: + pass # Keep main thread busy + sock.sendall(b"done\\n") + + # Create synchronization primitives + num_threads = 3 + barrier = threading.Barrier(num_threads + 1) # +1 for main thread + ready_event = threading.Event() + + # Start worker threads + threads = [] + for i in range(num_threads): + t = threading.Thread(target=worker_thread, args=(f"Worker-{{i}}", barrier, ready_event)) + t.start() + threads.append(t) + + # Wait for all threads to be ready + barrier.wait() + + # Signal ready to parent process + sock.sendall(b"ready\\n") + + # Signal threads to start waiting + ready_event.set() + + # Give threads time to start sleeping + time.sleep(0.1) + + # Now do busy work to hold the GIL + main_work() + """ + ) + + with os_helper.temp_dir() as work_dir: + script_dir = os.path.join(work_dir, "script_pkg") + os.mkdir(script_dir) + + # Create a socket server to communicate with the target process + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(("localhost", port)) + server_socket.settimeout(SHORT_TIMEOUT) + server_socket.listen(1) + + script_name = _make_test_script(script_dir, "script", script) + client_socket = None + try: + p = subprocess.Popen([sys.executable, script_name]) + client_socket, _ = server_socket.accept() + server_socket.close() + + # Wait for ready signal + response = b"" + while b"ready" not in response: + response += client_socket.recv(1024) + + # Wait for the main thread to start its busy work + while b"working" not in response: + response += client_socket.recv(1024) + + # Get stack trace with all threads + unwinder_all = RemoteUnwinder(p.pid, all_threads=True) + all_traces = unwinder_all.get_stack_trace() + + # Get stack trace with only GIL holder + unwinder_gil = RemoteUnwinder(p.pid, only_active_thread=True) + gil_traces = unwinder_gil.get_stack_trace() + + except PermissionError: + self.skipTest( + "Insufficient permissions to read the stack trace" + ) + finally: + if client_socket is not None: + client_socket.close() + p.kill() + p.terminate() + p.wait(timeout=SHORT_TIMEOUT) + + # Verify we got multiple threads in all_traces + self.assertGreater(len(all_traces), 1, "Should have multiple threads") + + # Verify we got exactly one thread in gil_traces + self.assertEqual(len(gil_traces), 1, "Should have exactly one GIL holder") + + # The GIL holder should be in the all_traces list + gil_thread_id = gil_traces[0][0] + all_thread_ids = [trace[0] for trace in all_traces] + self.assertIn(gil_thread_id, all_thread_ids, + "GIL holder should be among all threads") + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_iter.py b/Lib/test/test_iter.py index 1b9f3cf7624..18e4b676c53 100644 --- a/Lib/test/test_iter.py +++ b/Lib/test/test_iter.py @@ -1147,7 +1147,7 @@ class TestCase(unittest.TestCase): def test_exception_locations(self): # The location of an exception raised from __init__ or - # __next__ should should be the iterator expression + # __next__ should be the iterator expression def init_raises(): try: diff --git a/Lib/test/test_listcomps.py b/Lib/test/test_listcomps.py index cffdeeacc5d..70148dc30fc 100644 --- a/Lib/test/test_listcomps.py +++ b/Lib/test/test_listcomps.py @@ -716,7 +716,7 @@ class ListComprehensionTest(unittest.TestCase): def test_exception_locations(self): # The location of an exception raised from __init__ or - # __next__ should should be the iterator expression + # __next__ should be the iterator expression def init_raises(): try: diff --git a/Lib/test/test_math.py b/Lib/test/test_math.py index 384ad5c828d..46cb54647b1 100644 --- a/Lib/test/test_math.py +++ b/Lib/test/test_math.py @@ -475,6 +475,19 @@ class MathTests(unittest.TestCase): # similarly, copysign(2., NAN) could be 2. or -2. self.assertEqual(abs(math.copysign(2., NAN)), 2.) + def test_signbit(self): + self.assertRaises(TypeError, math.signbit) + self.assertRaises(TypeError, math.signbit, '1.0') + + # C11, ยง7.12.3.6 requires signbit() to return a nonzero value + # if and only if the sign of its argument value is negative, + # but in practice, we are only interested in a boolean value. + self.assertIsInstance(math.signbit(1.0), bool) + + for arg in [0., 1., INF, NAN]: + self.assertFalse(math.signbit(arg)) + self.assertTrue(math.signbit(-arg)) + def testCos(self): self.assertRaises(TypeError, math.cos) self.ftest('cos(-pi/2)', math.cos(-math.pi/2), 0, abs_tol=math.ulp(1)) @@ -1387,7 +1400,6 @@ class MathTests(unittest.TestCase): args = ((-5, -5, 10), (1.5, 4611686018427387904, 2305843009213693952)) self.assertEqual(sumprod(*args), 0.0) - @requires_IEEE_754 @unittest.skipIf(HAVE_DOUBLE_ROUNDING, "sumprod() accuracy not guaranteed on machines with double rounding") @@ -2486,7 +2498,6 @@ class MathTests(unittest.TestCase): with self.assertRaises(ValueError): math.nextafter(1.0, INF, steps=-1) - @requires_IEEE_754 def test_ulp(self): self.assertEqual(math.ulp(1.0), sys.float_info.epsilon) diff --git a/Lib/test/test_platform.py b/Lib/test/test_platform.py index 3688cc4267b..479649053ab 100644 --- a/Lib/test/test_platform.py +++ b/Lib/test/test_platform.py @@ -133,6 +133,22 @@ class PlatformTest(unittest.TestCase): for terse in (False, True): res = platform.platform(aliased, terse) + def test__platform(self): + for src, res in [ + ('foo bar', 'foo_bar'), + ( + '1/2\\3:4;5"6(7)8(7)6"5;4:3\\2/1', + '1-2-3-4-5-6-7-8-7-6-5-4-3-2-1' + ), + ('--', ''), + ('-f', '-f'), + ('-foo----', '-foo'), + ('--foo---', '-foo'), + ('---foo--', '-foo'), + ]: + with self.subTest(src=src): + self.assertEqual(platform._platform(src), res) + def test_system(self): res = platform.system() |