gh-91053: Add an optional callback that is invoked whenever a function is modified (#98175)

1 files changed, 93 insertions, 0 deletions

diff --git a/Lib/test/test_capi/test_watchers.py b/Lib/test/test_capi/test_watchers.py
index f635c74..5e4f42a 100644
--- a/Lib/test/test_capi/test_watchers.py
+++ b/Lib/test/test_capi/test_watchers.py
@@ -336,5 +336,98 @@ class TestTypeWatchers(unittest.TestCase):
                 self.add_watcher()
 
 
+class TestFuncWatchers(unittest.TestCase):
+    @contextmanager
+    def add_watcher(self, func):
+        wid = _testcapi.add_func_watcher(func)
+        try:
+            yield
+        finally:
+            _testcapi.clear_func_watcher(wid)
+
+    def test_func_events_dispatched(self):
+        events = []
+        def watcher(*args):
+            events.append(args)
+
+        with self.add_watcher(watcher):
+            def myfunc():
+                pass
+            self.assertIn((_testcapi.PYFUNC_EVENT_CREATE, myfunc, None), events)
+            myfunc_id = id(myfunc)
+
+            new_code = self.test_func_events_dispatched.__code__
+            myfunc.__code__ = new_code
+            self.assertIn((_testcapi.PYFUNC_EVENT_MODIFY_CODE, myfunc, new_code), events)
+
+            new_defaults = (123,)
+            myfunc.__defaults__ = new_defaults
+            self.assertIn((_testcapi.PYFUNC_EVENT_MODIFY_DEFAULTS, myfunc, new_defaults), events)
+
+            new_defaults = (456,)
+            _testcapi.set_func_defaults_via_capi(myfunc, new_defaults)
+            self.assertIn((_testcapi.PYFUNC_EVENT_MODIFY_DEFAULTS, myfunc, new_defaults), events)
+
+            new_kwdefaults = {"self": 123}
+            myfunc.__kwdefaults__ = new_kwdefaults
+            self.assertIn((_testcapi.PYFUNC_EVENT_MODIFY_KWDEFAULTS, myfunc, new_kwdefaults), events)
+
+            new_kwdefaults = {"self": 456}
+            _testcapi.set_func_kwdefaults_via_capi(myfunc, new_kwdefaults)
+            self.assertIn((_testcapi.PYFUNC_EVENT_MODIFY_KWDEFAULTS, myfunc, new_kwdefaults), events)
+
+            # Clear events reference to func
+            events = []
+            del myfunc
+            self.assertIn((_testcapi.PYFUNC_EVENT_DESTROY, myfunc_id, None), events)
+
+    def test_multiple_watchers(self):
+        events0 = []
+        def first_watcher(*args):
+            events0.append(args)
+
+        events1 = []
+        def second_watcher(*args):
+            events1.append(args)
+
+        with self.add_watcher(first_watcher):
+            with self.add_watcher(second_watcher):
+                def myfunc():
+                    pass
+
+                event = (_testcapi.PYFUNC_EVENT_CREATE, myfunc, None)
+                self.assertIn(event, events0)
+                self.assertIn(event, events1)
+
+    def test_watcher_raises_error(self):
+        class MyError(Exception):
+            pass
+
+        def watcher(*args):
+            raise MyError("testing 123")
+
+        with self.add_watcher(watcher):
+            with catch_unraisable_exception() as cm:
+                def myfunc():
+                    pass
+
+                self.assertIs(cm.unraisable.object, myfunc)
+                self.assertIsInstance(cm.unraisable.exc_value, MyError)
+
+    def test_clear_out_of_range_watcher_id(self):
+        with self.assertRaisesRegex(ValueError, r"invalid func watcher ID -1"):
+            _testcapi.clear_func_watcher(-1)
+        with self.assertRaisesRegex(ValueError, r"invalid func watcher ID 8"):
+            _testcapi.clear_func_watcher(8)  # FUNC_MAX_WATCHERS = 8
+
+    def test_clear_unassigned_watcher_id(self):
+        with self.assertRaisesRegex(ValueError, r"no func watcher set for ID 1"):
+            _testcapi.clear_func_watcher(1)
+
+    def test_allocate_too_many_watchers(self):
+        with self.assertRaisesRegex(RuntimeError, r"no more func watcher IDs"):
+            _testcapi.allocate_too_many_func_watchers()
+
+
 if __name__ == "__main__":
     unittest.main()