bpo-38920: Add audit hooks for when sys.excepthook and sys.unraisable hooks are invoked (GH-17392)

Also fixes some potential segfaults in unraisable hook handling.

2 files changed, 79 insertions, 14 deletions

diff --git a/Lib/test/audit-tests.py b/Lib/test/audit-tests.py
index ddeff22..ed08612 100644
--- a/Lib/test/audit-tests.py
+++ b/Lib/test/audit-tests.py
@@ -263,13 +263,50 @@ def test_cantrace():
 
 def test_mmap():
     import mmap
+
     with TestHook() as hook:
         mmap.mmap(-1, 8)
         assertEqual(hook.seen[0][1][:2], (-1, 8))
 
 
+def test_excepthook():
+    def excepthook(exc_type, exc_value, exc_tb):
+        if exc_type is not RuntimeError:
+            sys.__excepthook__(exc_type, exc_value, exc_tb)
+
+    def hook(event, args):
+        if event == "sys.excepthook":
+            if not isinstance(args[2], args[1]):
+                raise TypeError(f"Expected isinstance({args[2]!r}, " f"{args[1]!r})")
+            if args[0] != excepthook:
+                raise ValueError(f"Expected {args[0]} == {excepthook}")
+            print(event, repr(args[2]))
+
+    sys.addaudithook(hook)
+    sys.excepthook = excepthook
+    raise RuntimeError("fatal-error")
+
+
+def test_unraisablehook():
+    from _testcapi import write_unraisable_exc
+
+    def unraisablehook(hookargs):
+        pass
+
+    def hook(event, args):
+        if event == "sys.unraisablehook":
+            if args[0] != unraisablehook:
+                raise ValueError(f"Expected {args[0]} == {unraisablehook}")
+            print(event, repr(args[1].exc_value), args[1].err_msg)
+
+    sys.addaudithook(hook)
+    sys.unraisablehook = unraisablehook
+    write_unraisable_exc(RuntimeError("nonfatal-error"), "for audit hook test", None)
+
+
 if __name__ == "__main__":
     from test.libregrtest.setup import suppress_msvcrt_asserts
+
     suppress_msvcrt_asserts(False)
 
     test = sys.argv[1]
diff --git a/Lib/test/test_audit.py b/Lib/test/test_audit.py
index 41f9fae..31a0855 100644
--- a/Lib/test/test_audit.py
+++ b/Lib/test/test_audit.py
@@ -24,7 +24,23 @@ class AuditTest(unittest.TestCase):
             sys.stdout.writelines(p.stdout)
             sys.stderr.writelines(p.stderr)
             if p.returncode:
-                self.fail(''.join(p.stderr))
+                self.fail("".join(p.stderr))
+
+    def run_python(self, *args):
+        events = []
+        with subprocess.Popen(
+            [sys.executable, "-X utf8", AUDIT_TESTS_PY, *args],
+            encoding="utf-8",
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        ) as p:
+            p.wait()
+            sys.stderr.writelines(p.stderr)
+            return (
+                p.returncode,
+                [line.strip().partition(" ") for line in p.stdout],
+                "".join(p.stderr),
+            )
 
     def test_basic(self):
         self.do_test("test_basic")
@@ -36,19 +52,11 @@ class AuditTest(unittest.TestCase):
         self.do_test("test_block_add_hook_baseexception")
 
     def test_finalize_hooks(self):
-        events = []
-        with subprocess.Popen(
-            [sys.executable, "-X utf8", AUDIT_TESTS_PY, "test_finalize_hooks"],
-            encoding="utf-8",
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        ) as p:
-            p.wait()
-            for line in p.stdout:
-                events.append(line.strip().partition(" "))
-            sys.stderr.writelines(p.stderr)
-            if p.returncode:
-                self.fail(''.join(p.stderr))
+        returncode, events, stderr = self.run_python("test_finalize_hooks")
+        if stderr:
+            print(stderr, file=sys.stderr)
+        if returncode:
+            self.fail(stderr)
 
         firstId = events[0][2]
         self.assertSequenceEqual(
@@ -76,6 +84,26 @@ class AuditTest(unittest.TestCase):
     def test_mmap(self):
         self.do_test("test_mmap")
 
+    def test_excepthook(self):
+        returncode, events, stderr = self.run_python("test_excepthook")
+        if not returncode:
+            self.fail(f"Expected fatal exception\n{stderr}")
+
+        self.assertSequenceEqual(
+            [("sys.excepthook", " ", "RuntimeError('fatal-error')")], events
+        )
+
+    def test_unraisablehook(self):
+        returncode, events, stderr = self.run_python("test_unraisablehook")
+        if returncode:
+            self.fail(stderr)
+
+        self.assertEqual(events[0][0], "sys.unraisablehook")
+        self.assertEqual(
+            events[0][2],
+            "RuntimeError('nonfatal-error') Exception ignored for audit hook test",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()