gh-99377: Add audit events for thread creation and clear (GH-99378)

2 files changed, 67 insertions, 0 deletions

diff --git a/Lib/test/audit-tests.py b/Lib/test/audit-tests.py
index a411072ba7d..bf56cea541d 100644
--- a/Lib/test/audit-tests.py
+++ b/Lib/test/audit-tests.py
@@ -419,6 +419,48 @@ def test_sys_getframe():
     sys._getframe()
 
 
+def test_threading():
+    import _thread
+
+    def hook(event, args):
+        if event.startswith(("_thread.", "cpython.PyThreadState", "test.")):
+            print(event, args)
+
+    sys.addaudithook(hook)
+
+    lock = _thread.allocate_lock()
+    lock.acquire()
+
+    class test_func:
+        def __repr__(self): return "<test_func>"
+        def __call__(self):
+            sys.audit("test.test_func")
+            lock.release()
+
+    i = _thread.start_new_thread(test_func(), ())
+    lock.acquire()
+
+
+def test_threading_abort():
+    # Ensures that aborting PyThreadState_New raises the correct exception
+    import _thread
+
+    class ThreadNewAbortError(Exception):
+        pass
+
+    def hook(event, args):
+        if event == "cpython.PyThreadState_New":
+            raise ThreadNewAbortError()
+
+    sys.addaudithook(hook)
+
+    try:
+        _thread.start_new_thread(lambda: None, ())
+    except ThreadNewAbortError:
+        # Other exceptions are raised and the test will fail
+        pass
+
+
 def test_wmi_exec_query():
     import _wmi
 
diff --git a/Lib/test/test_audit.py b/Lib/test/test_audit.py
index e8d560ad5ce..5a2997ac9e6 100644
--- a/Lib/test/test_audit.py
+++ b/Lib/test/test_audit.py
@@ -186,6 +186,31 @@ class AuditTest(unittest.TestCase):
 
         self.assertEqual(actual, expected)
 
+
+    def test_threading(self):
+        returncode, events, stderr = self.run_python("test_threading")
+        if returncode:
+            self.fail(stderr)
+
+        if support.verbose:
+            print(*events, sep='\n')
+        actual = [(ev[0], ev[2]) for ev in events]
+        expected = [
+            ("_thread.start_new_thread", "(<test_func>, (), None)"),
+            ("cpython.PyThreadState_New", "(2,)"),
+            ("test.test_func", "()"),
+            ("cpython.PyThreadState_Clear", "(2,)"),
+        ]
+
+        self.assertEqual(actual, expected)
+
+    def test_threading_abort(self):
+        # Ensures that aborting PyThreadState_New raises the correct exception
+        returncode, events, stderr = self.run_python("test_threading_abort")
+        if returncode:
+            self.fail(stderr)
+
+
     def test_wmi_exec_query(self):
         import_helper.import_module("_wmi")
         returncode, events, stderr = self.run_python("test_wmi_exec_query")