test_watchers.py

importunittest
importcontextvars

fromcontextlibimportcontextmanager, ExitStack
fromtest.supportimport (
catch_unraisable_exception, import_helper,
gc_collect)


# Skip this test if the _testcapi module isn't available.
_testcapi=import_helper.import_module('_testcapi')


classTestDictWatchers(unittest.TestCase):
# types of watchers testcapimodule can add:
EVENTS=0# appends dict events as strings to global event list
ERROR=1# unconditionally sets and signals a RuntimeException
SECOND=2# always appends "second" to global event list

defadd_watcher(self, kind=EVENTS):
return_testcapi.add_dict_watcher(kind)

defclear_watcher(self, watcher_id):
_testcapi.clear_dict_watcher(watcher_id)

@contextmanager
defwatcher(self, kind=EVENTS):
wid=self.add_watcher(kind)
try:
yieldwid
finally:
self.clear_watcher(wid)

defassert_events(self, expected):
actual=_testcapi.get_dict_watcher_events()
self.assertEqual(actual, expected)

defwatch(self, wid, d):
_testcapi.watch_dict(wid, d)

defunwatch(self, wid, d):
_testcapi.unwatch_dict(wid, d)

deftest_set_new_item(self):
d= {}
withself.watcher() aswid:
self.watch(wid, d)
d["foo"] ="bar"
self.assert_events(["new:foo:bar"])

deftest_set_existing_item(self):
d= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
d["foo"] ="baz"
self.assert_events(["mod:foo:baz"])

deftest_clone(self):
d= {}
d2= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
d.update(d2)
self.assert_events(["clone"])

deftest_no_event_if_not_watched(self):
d= {}
withself.watcher() aswid:
d["foo"] ="bar"
self.assert_events([])

deftest_del(self):
d= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
deld["foo"]
self.assert_events(["del:foo"])

deftest_pop(self):
d= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
d.pop("foo")
self.assert_events(["del:foo"])

deftest_clear(self):
d= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
d.clear()
self.assert_events(["clear"])

deftest_dealloc(self):
d= {"foo": "bar"}
withself.watcher() aswid:
self.watch(wid, d)
deld
self.assert_events(["dealloc"])

deftest_object_dict(self):
classMyObj: pass
o=MyObj()

withself.watcher() aswid:
self.watch(wid, o.__dict__)
o.foo="bar"
o.foo="baz"
delo.foo
self.assert_events(["new:foo:bar", "mod:foo:baz", "del:foo"])

withself.watcher() aswid:
self.watch(wid, o.__dict__)
for_inrange(100):
o.foo="bar"
self.assert_events(["new:foo:bar"] + ["mod:foo:bar"] *99)

deftest_unwatch(self):
d= {}
withself.watcher() aswid:
self.watch(wid, d)
d["foo"] ="bar"
self.unwatch(wid, d)
d["hmm"] ="baz"
self.assert_events(["new:foo:bar"])

deftest_error(self):
d= {}
withself.watcher(kind=self.ERROR) aswid:
self.watch(wid, d)
withcatch_unraisable_exception() ascm:
d["foo"] ="bar"
self.assertIn(
"Exception ignored in "
"PyDict_EVENT_ADDED watcher callback for <dict at ",
cm.unraisable.err_msg
 )
self.assertIsNone(cm.unraisable.object)
self.assertEqual(str(cm.unraisable.exc_value), "boom!")
self.assert_events([])

deftest_dealloc_error(self):
d= {}
withself.watcher(kind=self.ERROR) aswid:
self.watch(wid, d)
withcatch_unraisable_exception() ascm:
deld
self.assertEqual(str(cm.unraisable.exc_value), "boom!")

deftest_two_watchers(self):
d1= {}
d2= {}
withself.watcher() aswid1:
withself.watcher(kind=self.SECOND) aswid2:
self.watch(wid1, d1)
self.watch(wid2, d2)
d1["foo"] ="bar"
d2["hmm"] ="baz"
self.assert_events(["new:foo:bar", "second"])

deftest_watch_non_dict(self):
withself.watcher() aswid:
withself.assertRaisesRegex(ValueError, r"Cannot watch non-dictionary"):
self.watch(wid, 1)

deftest_watch_out_of_range_watcher_id(self):
d= {}
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID -1"):
self.watch(-1, d)
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID 8"):
self.watch(8, d) # DICT_MAX_WATCHERS = 8

deftest_watch_unassigned_watcher_id(self):
d= {}
withself.assertRaisesRegex(ValueError, r"No dict watcher set for ID 3"):
self.watch(3, d)

deftest_unwatch_non_dict(self):
withself.watcher() aswid:
withself.assertRaisesRegex(ValueError, r"Cannot watch non-dictionary"):
self.unwatch(wid, 1)

deftest_unwatch_out_of_range_watcher_id(self):
d= {}
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID -1"):
self.unwatch(-1, d)
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID 8"):
self.unwatch(8, d) # DICT_MAX_WATCHERS = 8

deftest_unwatch_unassigned_watcher_id(self):
d= {}
withself.assertRaisesRegex(ValueError, r"No dict watcher set for ID 3"):
self.unwatch(3, d)

deftest_clear_out_of_range_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID -1"):
self.clear_watcher(-1)
withself.assertRaisesRegex(ValueError, r"Invalid dict watcher ID 8"):
self.clear_watcher(8) # DICT_MAX_WATCHERS = 8

deftest_clear_unassigned_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"No dict watcher set for ID 3"):
self.clear_watcher(3)


classTestTypeWatchers(unittest.TestCase):
# types of watchers testcapimodule can add:
TYPES=0# appends modified types to global event list
ERROR=1# unconditionally sets and signals a RuntimeException
WRAP=2# appends modified type wrapped in list to global event list

# duplicating the C constant
TYPE_MAX_WATCHERS=8

defadd_watcher(self, kind=TYPES):
return_testcapi.add_type_watcher(kind)

defclear_watcher(self, watcher_id):
_testcapi.clear_type_watcher(watcher_id)

@contextmanager
defwatcher(self, kind=TYPES):
wid=self.add_watcher(kind)
try:
yieldwid
finally:
self.clear_watcher(wid)

defassert_events(self, expected):
actual=_testcapi.get_type_modified_events()
self.assertEqual(actual, expected)

defwatch(self, wid, t):
_testcapi.watch_type(wid, t)

defunwatch(self, wid, t):
_testcapi.unwatch_type(wid, t)

deftest_watch_type(self):
classC: pass
withself.watcher() aswid:
self.watch(wid, C)
C.foo="bar"
self.assert_events([C])

deftest_event_aggregation(self):
classC: pass
withself.watcher() aswid:
self.watch(wid, C)
C.foo="bar"
C.bar="baz"
# only one event registered for both modifications
self.assert_events([C])

deftest_lookup_resets_aggregation(self):
classC: pass
withself.watcher() aswid:
self.watch(wid, C)
C.foo="bar"
# lookup resets type version tag
self.assertEqual(C.foo, "bar")
C.bar="baz"
# both events registered
self.assert_events([C, C])

deftest_unwatch_type(self):
classC: pass
withself.watcher() aswid:
self.watch(wid, C)
C.foo="bar"
self.assertEqual(C.foo, "bar")
self.assert_events([C])
self.unwatch(wid, C)
C.bar="baz"
self.assert_events([C])

deftest_clear_watcher(self):
classC: pass
# outer watcher is unused, it's just to keep events list alive
withself.watcher() as_:
withself.watcher() aswid:
self.watch(wid, C)
C.foo="bar"
self.assertEqual(C.foo, "bar")
self.assert_events([C])
C.bar="baz"
# Watcher on C has been cleared, no new event
self.assert_events([C])

deftest_watch_type_subclass(self):
classC: pass
classD(C): pass
withself.watcher() aswid:
self.watch(wid, D)
C.foo="bar"
self.assert_events([D])

deftest_error(self):
classC: pass
withself.watcher(kind=self.ERROR) aswid:
self.watch(wid, C)
withcatch_unraisable_exception() ascm:
C.foo="bar"
self.assertEqual(
cm.unraisable.err_msg,
f"Exception ignored in type watcher callback #1 for {C!r}",
 )
self.assertIs(cm.unraisable.object, None)
self.assertEqual(str(cm.unraisable.exc_value), "boom!")
self.assert_events([])

deftest_two_watchers(self):
classC1: pass
classC2: pass
withself.watcher() aswid1:
withself.watcher(kind=self.WRAP) aswid2:
self.assertNotEqual(wid1, wid2)
self.watch(wid1, C1)
self.watch(wid2, C2)
C1.foo="bar"
C2.hmm="baz"
self.assert_events([C1, [C2]])

deftest_all_watchers(self):
classC: pass
withExitStack() asstack:
last_wid=-1
# don't make assumptions about how many watchers are already
# registered, just go until we reach the max ID
whilelast_wid<self.TYPE_MAX_WATCHERS-1:
last_wid=stack.enter_context(self.watcher())
self.watch(last_wid, C)
C.foo="bar"
self.assert_events([C])

deftest_watch_non_type(self):
withself.watcher() aswid:
withself.assertRaisesRegex(ValueError, r"Cannot watch non-type"):
self.watch(wid, 1)

deftest_watch_out_of_range_watcher_id(self):
classC: pass
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
self.watch(-1, C)
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
self.watch(self.TYPE_MAX_WATCHERS, C)

deftest_watch_unassigned_watcher_id(self):
classC: pass
withself.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
self.watch(1, C)

deftest_unwatch_non_type(self):
withself.watcher() aswid:
withself.assertRaisesRegex(ValueError, r"Cannot watch non-type"):
self.unwatch(wid, 1)

deftest_unwatch_out_of_range_watcher_id(self):
classC: pass
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
self.unwatch(-1, C)
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
self.unwatch(self.TYPE_MAX_WATCHERS, C)

deftest_unwatch_unassigned_watcher_id(self):
classC: pass
withself.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
self.unwatch(1, C)

deftest_clear_out_of_range_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
self.clear_watcher(-1)
withself.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
self.clear_watcher(self.TYPE_MAX_WATCHERS)

deftest_clear_unassigned_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
self.clear_watcher(1)

deftest_no_more_ids_available(self):
withself.assertRaisesRegex(RuntimeError, r"no more type watcher IDs"):
withExitStack() asstack:
for_inrange(self.TYPE_MAX_WATCHERS+1):
stack.enter_context(self.watcher())


classTestCodeObjectWatchers(unittest.TestCase):
@contextmanager
defcode_watcher(self, which_watcher):
wid=_testcapi.add_code_watcher(which_watcher)
try:
yieldwid
finally:
_testcapi.clear_code_watcher(wid)

defassert_event_counts(self, exp_created_0, exp_destroyed_0,
exp_created_1, exp_destroyed_1):
gc_collect() # code objects are collected by GC in free-threaded build
self.assertEqual(
exp_created_0, _testcapi.get_code_watcher_num_created_events(0))
self.assertEqual(
exp_destroyed_0, _testcapi.get_code_watcher_num_destroyed_events(0))
self.assertEqual(
exp_created_1, _testcapi.get_code_watcher_num_created_events(1))
self.assertEqual(
exp_destroyed_1, _testcapi.get_code_watcher_num_destroyed_events(1))

deftest_code_object_events_dispatched(self):
# verify that all counts are zero before any watchers are registered
self.assert_event_counts(0, 0, 0, 0)

# verify that all counts remain zero when a code object is
# created and destroyed with no watchers registered
co1=_testcapi.code_newempty("test_watchers", "dummy1", 0)
self.assert_event_counts(0, 0, 0, 0)
delco1
self.assert_event_counts(0, 0, 0, 0)

# verify counts are as expected when first watcher is registered
withself.code_watcher(0):
self.assert_event_counts(0, 0, 0, 0)
co2=_testcapi.code_newempty("test_watchers", "dummy2", 0)
self.assert_event_counts(1, 0, 0, 0)
delco2
self.assert_event_counts(1, 1, 0, 0)

# again with second watcher registered
withself.code_watcher(1):
self.assert_event_counts(1, 1, 0, 0)
co3=_testcapi.code_newempty("test_watchers", "dummy3", 0)
self.assert_event_counts(2, 1, 1, 0)
delco3
self.assert_event_counts(2, 2, 1, 1)

# verify counts are reset and don't change after both watchers are cleared
co4=_testcapi.code_newempty("test_watchers", "dummy4", 0)
self.assert_event_counts(0, 0, 0, 0)
delco4
self.assert_event_counts(0, 0, 0, 0)

deftest_error(self):
withself.code_watcher(2):
withcatch_unraisable_exception() ascm:
co=_testcapi.code_newempty("test_watchers", "dummy0", 0)

self.assertEqual(
cm.unraisable.err_msg,
f"Exception ignored in "
f"PY_CODE_EVENT_CREATE watcher callback for {co!r}"
 )
self.assertIsNone(cm.unraisable.object)
self.assertEqual(str(cm.unraisable.exc_value), "boom!")

deftest_dealloc_error(self):
co=_testcapi.code_newempty("test_watchers", "dummy0", 0)
withself.code_watcher(2):
withcatch_unraisable_exception() ascm:
delco
gc_collect()

self.assertEqual(str(cm.unraisable.exc_value), "boom!")

deftest_clear_out_of_range_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"Invalid code watcher ID -1"):
_testcapi.clear_code_watcher(-1)
withself.assertRaisesRegex(ValueError, r"Invalid code watcher ID 8"):
_testcapi.clear_code_watcher(8) # CODE_MAX_WATCHERS = 8

deftest_clear_unassigned_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"No code watcher set for ID 1"):
_testcapi.clear_code_watcher(1)

deftest_allocate_too_many_watchers(self):
withself.assertRaisesRegex(RuntimeError, r"no more code watcher IDs available"):
_testcapi.allocate_too_many_code_watchers()


classTestFuncWatchers(unittest.TestCase):
@contextmanager
defadd_watcher(self, func):
wid=_testcapi.add_func_watcher(func)
try:
yield
finally:
_testcapi.clear_func_watcher(wid)

deftest_func_events_dispatched(self):
events= []
defwatcher(*args):
events.append(args)

withself.add_watcher(watcher):
defmyfunc():
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= []
delmyfunc
self.assertIn((_testcapi.PYFUNC_EVENT_DESTROY, myfunc_id, None), events)

deftest_multiple_watchers(self):
events0= []
deffirst_watcher(*args):
events0.append(args)

events1= []
defsecond_watcher(*args):
events1.append(args)

withself.add_watcher(first_watcher):
withself.add_watcher(second_watcher):
defmyfunc():
pass

event= (_testcapi.PYFUNC_EVENT_CREATE, myfunc, None)
self.assertIn(event, events0)
self.assertIn(event, events1)

deftest_watcher_raises_error(self):
classMyError(Exception):
pass

defwatcher(*args):
raiseMyError("testing 123")

withself.add_watcher(watcher):
withcatch_unraisable_exception() ascm:
defmyfunc():
pass

self.assertEqual(
cm.unraisable.err_msg,
f"Exception ignored in "
f"PyFunction_EVENT_CREATE watcher callback for {repr(myfunc)[1:-1]}"
 )
self.assertIsNone(cm.unraisable.object)

deftest_dealloc_watcher_raises_error(self):
classMyError(Exception):
pass

defwatcher(*args):
raiseMyError("testing 123")

defmyfunc():
pass

withself.add_watcher(watcher):
withcatch_unraisable_exception() ascm:
delmyfunc

self.assertIsInstance(cm.unraisable.exc_value, MyError)

deftest_clear_out_of_range_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"invalid func watcher ID -1"):
_testcapi.clear_func_watcher(-1)
withself.assertRaisesRegex(ValueError, r"invalid func watcher ID 8"):
_testcapi.clear_func_watcher(8) # FUNC_MAX_WATCHERS = 8

deftest_clear_unassigned_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"no func watcher set for ID 1"):
_testcapi.clear_func_watcher(1)

deftest_allocate_too_many_watchers(self):
withself.assertRaisesRegex(RuntimeError, r"no more func watcher IDs"):
_testcapi.allocate_too_many_func_watchers()


classTestContextObjectWatchers(unittest.TestCase):
@contextmanager
defcontext_watcher(self, which_watcher):
wid=_testcapi.add_context_watcher(which_watcher)
try:
switches=_testcapi.get_context_switches(which_watcher)
exceptValueError:
switches=None
try:
yieldswitches
finally:
_testcapi.clear_context_watcher(wid)

defassert_event_counts(self, want_0, want_1):
self.assertEqual(len(_testcapi.get_context_switches(0)), want_0)
self.assertEqual(len(_testcapi.get_context_switches(1)), want_1)

deftest_context_object_events_dispatched(self):
# verify that all counts are zero before any watchers are registered
self.assert_event_counts(0, 0)

# verify that all counts remain zero when a context object is
# entered and exited with no watchers registered
ctx=contextvars.copy_context()
ctx.run(self.assert_event_counts, 0, 0)
self.assert_event_counts(0, 0)

# verify counts are as expected when first watcher is registered
withself.context_watcher(0):
self.assert_event_counts(0, 0)
ctx.run(self.assert_event_counts, 1, 0)
self.assert_event_counts(2, 0)

# again with second watcher registered
withself.context_watcher(1):
self.assert_event_counts(2, 0)
ctx.run(self.assert_event_counts, 3, 1)
self.assert_event_counts(4, 2)

# verify counts are reset and don't change after both watchers are cleared
ctx.run(self.assert_event_counts, 0, 0)
self.assert_event_counts(0, 0)

deftest_callback_error(self):
ctx_outer=contextvars.copy_context()
ctx_inner=contextvars.copy_context()
unraisables= []

def_in_outer():
withself.context_watcher(2):
withcatch_unraisable_exception() ascm:
ctx_inner.run(lambda: unraisables.append(cm.unraisable))
unraisables.append(cm.unraisable)

try:
ctx_outer.run(_in_outer)
self.assertEqual([x.err_msgforxinunraisables],
 ["Exception ignored in Py_CONTEXT_SWITCHED "
f"watcher callback for {ctx!r}"
forctxin [ctx_inner, ctx_outer]])
self.assertEqual([str(x.exc_value) forxinunraisables],
 ["boom!", "boom!"])
finally:
# Break reference cycle
unraisables=None

deftest_clear_out_of_range_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"Invalid context watcher ID -1"):
_testcapi.clear_context_watcher(-1)
withself.assertRaisesRegex(ValueError, r"Invalid context watcher ID 8"):
_testcapi.clear_context_watcher(8) # CONTEXT_MAX_WATCHERS = 8

deftest_clear_unassigned_watcher_id(self):
withself.assertRaisesRegex(ValueError, r"No context watcher set for ID 1"):
_testcapi.clear_context_watcher(1)

deftest_allocate_too_many_watchers(self):
withself.assertRaisesRegex(RuntimeError, r"no more context watcher IDs available"):
_testcapi.allocate_too_many_context_watchers()

deftest_exit_base_context(self):
ctx=contextvars.Context()
_testcapi.clear_context_stack()
withself.context_watcher(0) asswitches:
ctx.run(lambda: None)
self.assertEqual(switches, [ctx, None])

if__name__=="__main__":
unittest.main()