watchers.c | Explore in Territory

// clinic/watchers.c.h uses internal pycore_modsupport.h API
#define PYTESTCAPI_NEED_INTERNAL_API

#include "parts.h"

#include "clinic/watchers.c.h"

#define Py_BUILD_CORE
#include "pycore_function.h"  // FUNC_MAX_WATCHERS
#include "pycore_code.h"  // CODE_MAX_WATCHERS
#include "pycore_context.h" // CONTEXT_MAX_WATCHERS

/*[clinic input]
module _testcapi
[clinic start generated code]*/
/*[clinic end generated code: output=da39a3ee5e6b4b0d input=6361033e795369fc]*/

// Test dict watching
static PyObject *g_dict_watch_events = …;
static int g_dict_watchers_installed = …;

static int
dict_watch_callback(PyDict_WatchEvent event,
                    PyObject *dict,
                    PyObject *key,
                    PyObject *new_value)
{ … }

static int
dict_watch_callback_second(PyDict_WatchEvent event,
                           PyObject *dict,
                           PyObject *key,
                           PyObject *new_value)
{ … }

static int
dict_watch_callback_error(PyDict_WatchEvent event,
                          PyObject *dict,
                          PyObject *key,
                          PyObject *new_value)
{ … }

static PyObject *
add_dict_watcher(PyObject *self, PyObject *kind)
{ … }

static PyObject *
clear_dict_watcher(PyObject *self, PyObject *watcher_id)
{ … }

/*[clinic input]
_testcapi.watch_dict
    watcher_id: int
    dict: object
    /
[clinic start generated code]*/

static PyObject *
_testcapi_watch_dict_impl(PyObject *module, int watcher_id, PyObject *dict)
/*[clinic end generated code: output=1426e0273cebe2d8 input=269b006d60c358bd]*/
{ … }

/*[clinic input]
_testcapi.unwatch_dict = _testcapi.watch_dict
[clinic start generated code]*/

static PyObject *
_testcapi_unwatch_dict_impl(PyObject *module, int watcher_id, PyObject *dict)
/*[clinic end generated code: output=512b1a71ae33c351 input=cae7dc1b6f7713b8]*/
{ … }

static PyObject *
get_dict_watcher_events(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

// Test type watchers
static PyObject *g_type_modified_events;
static int g_type_watchers_installed;

static int
type_modified_callback(PyTypeObject *type)
{ … }

static int
type_modified_callback_wrap(PyTypeObject *type)
{ … }

static int
type_modified_callback_error(PyTypeObject *type)
{ … }

static PyObject *
add_type_watcher(PyObject *self, PyObject *kind)
{ … }

static PyObject *
clear_type_watcher(PyObject *self, PyObject *watcher_id)
{ … }

static PyObject *
get_type_modified_events(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

/*[clinic input]
_testcapi.watch_type
    watcher_id: int
    type: object
    /
[clinic start generated code]*/

static PyObject *
_testcapi_watch_type_impl(PyObject *module, int watcher_id, PyObject *type)
/*[clinic end generated code: output=fdf4777126724fc4 input=5a808bf12be7e3ed]*/
{ … }

/*[clinic input]
_testcapi.unwatch_type = _testcapi.watch_type
[clinic start generated code]*/

static PyObject *
_testcapi_unwatch_type_impl(PyObject *module, int watcher_id, PyObject *type)
/*[clinic end generated code: output=0389672d4ad5f68b input=6701911fb45edc9e]*/
{ … }


// Test code object watching

#define NUM_CODE_WATCHERS …
static int code_watcher_ids[NUM_CODE_WATCHERS] = …;
static int num_code_object_created_events[NUM_CODE_WATCHERS] = …;
static int num_code_object_destroyed_events[NUM_CODE_WATCHERS] = …;

static int
handle_code_object_event(int which_watcher, PyCodeEvent event, PyCodeObject *co) { … }

static int
first_code_object_callback(PyCodeEvent event, PyCodeObject *co)
{ … }

static int
second_code_object_callback(PyCodeEvent event, PyCodeObject *co)
{ … }

static int
noop_code_event_handler(PyCodeEvent event, PyCodeObject *co)
{ … }

static int
error_code_event_handler(PyCodeEvent event, PyCodeObject *co)
{ … }

static PyObject *
add_code_watcher(PyObject *self, PyObject *which_watcher)
{ … }

static PyObject *
clear_code_watcher(PyObject *self, PyObject *watcher_id)
{ … }

static PyObject *
get_code_watcher_num_created_events(PyObject *self, PyObject *watcher_id)
{ … }

static PyObject *
get_code_watcher_num_destroyed_events(PyObject *self, PyObject *watcher_id)
{ … }

static PyObject *
allocate_too_many_code_watchers(PyObject *self, PyObject *args)
{ … }

// Test function watchers

#define NUM_TEST_FUNC_WATCHERS …
static PyObject *pyfunc_watchers[NUM_TEST_FUNC_WATCHERS];
static int func_watcher_ids[NUM_TEST_FUNC_WATCHERS] = …;

static PyObject *
get_id(PyObject *obj)
{ … }

static int
call_pyfunc_watcher(PyObject *watcher, PyFunction_WatchEvent event,
                    PyFunctionObject *func, PyObject *new_value)
{ … }

static int
first_func_watcher_callback(PyFunction_WatchEvent event, PyFunctionObject *func,
                            PyObject *new_value)
{ … }

static int
second_func_watcher_callback(PyFunction_WatchEvent event,
                             PyFunctionObject *func, PyObject *new_value)
{ … }

static PyFunction_WatchCallback func_watcher_callbacks[NUM_TEST_FUNC_WATCHERS] = …;

static int
add_func_event(PyObject *module, const char *name, PyFunction_WatchEvent event)
{ … }

static PyObject *
add_func_watcher(PyObject *self, PyObject *func)
{ … }

static PyObject *
clear_func_watcher(PyObject *self, PyObject *watcher_id_obj)
{ … }

static int
noop_func_event_handler(PyFunction_WatchEvent event, PyFunctionObject *func,
             PyObject *new_value)
{ … }

static PyObject *
allocate_too_many_func_watchers(PyObject *self, PyObject *args)
{ … }

// Test contexct object watchers
#define NUM_CONTEXT_WATCHERS …
static int context_watcher_ids[NUM_CONTEXT_WATCHERS] = …;
static PyObject *context_switches[NUM_CONTEXT_WATCHERS];

static int
handle_context_watcher_event(int which_watcher, PyContextEvent event, PyObject *ctx) { … }

static int
first_context_watcher_callback(PyContextEvent event, PyObject *ctx) { … }

static int
second_context_watcher_callback(PyContextEvent event, PyObject *ctx) { … }

static int
noop_context_event_handler(PyContextEvent event, PyObject *ctx) { … }

static int
error_context_event_handler(PyContextEvent event, PyObject *ctx) { … }

static PyObject *
add_context_watcher(PyObject *self, PyObject *which_watcher)
{ … }

static PyObject *
clear_context_watcher(PyObject *self, PyObject *watcher_id)
{ … }

static PyObject *
clear_context_stack(PyObject *Py_UNUSED(self), PyObject *Py_UNUSED(args))
{ … }

static PyObject *
get_context_switches(PyObject *Py_UNUSED(self), PyObject *watcher_id)
{ … }

static PyObject *
allocate_too_many_context_watchers(PyObject *self, PyObject *args)
{ … }

/*[clinic input]
_testcapi.set_func_defaults_via_capi
    func: object
    defaults: object
    /
[clinic start generated code]*/

static PyObject *
_testcapi_set_func_defaults_via_capi_impl(PyObject *module, PyObject *func,
                                          PyObject *defaults)
/*[clinic end generated code: output=caf0cb39db31ac24 input=e04a8508ca9d42fc]*/
{ … }

/*[clinic input]
_testcapi.set_func_kwdefaults_via_capi = _testcapi.set_func_defaults_via_capi
[clinic start generated code]*/

static PyObject *
_testcapi_set_func_kwdefaults_via_capi_impl(PyObject *module, PyObject *func,
                                            PyObject *defaults)
/*[clinic end generated code: output=9ed3b08177025070 input=f3cd1ca3c18de8ce]*/
{ … }

static PyMethodDef test_methods[] = …;

int
_PyTestCapi_Init_Watchers(PyObject *mod)
{ … }
cpython/Modules/_testcapi/watchers.c
