Modules/_testcapi/object.c

#include "parts.h"
#include "util.h"

static PyObject *
call_pyobject_print(PyObject *self, PyObject * args)
{
    PyObject *object;
    PyObject *filename;
    PyObject *print_raw;
    FILE *fp;
    int flags = 0;

    if (!PyArg_UnpackTuple(args, "call_pyobject_print", 3, 3,
                           &object, &filename, &print_raw)) {
        return NULL;
    }

    fp = Py_fopen(filename, "w+");
    if (fp == NULL) {
        return NULL;
    }

    if (Py_IsTrue(print_raw)) {
        flags = Py_PRINT_RAW;
    }

    if (PyObject_Print(object, fp, flags) < 0) {
        fclose(fp);
        return NULL;
    }

    fclose(fp);

    Py_RETURN_NONE;
}

static PyObject *
pyobject_print_null(PyObject *self, PyObject *filename)
{
    FILE *fp;

    fp = Py_fopen(filename, "w+");
    if (fp == NULL) {
        return NULL;
    }

    if (PyObject_Print(NULL, fp, 0) < 0) {
        fclose(fp);
        return NULL;
    }

    fclose(fp);

    Py_RETURN_NONE;
}

static PyObject *
pyobject_print_noref_object(PyObject *self, PyObject *filename)
{
    PyObject *test_string;
    FILE *fp;
    char correct_string[100];

    test_string = PyUnicode_FromString("Spam spam spam");
    if (test_string == NULL) {
        return NULL;
    }

    Py_SET_REFCNT(test_string, 0);

    PyOS_snprintf(correct_string, 100, "<refcnt %zd at %p>",
                  Py_REFCNT(test_string), (void *)test_string);

    fp = Py_fopen(filename, "w+");
    if (fp == NULL) {
        Py_SET_REFCNT(test_string, 1);
        Py_DECREF(test_string);
        return NULL;
    }

    if (PyObject_Print(test_string, fp, 0) < 0){
        fclose(fp);
        Py_SET_REFCNT(test_string, 1);
        Py_DECREF(test_string);
        return NULL;
    }

    fclose(fp);

    Py_SET_REFCNT(test_string, 1);
    Py_DECREF(test_string);

    return PyUnicode_FromString(correct_string);
}

static PyObject *
pyobject_print_os_error(PyObject *self, PyObject *filename)
{
    PyObject *test_string;
    FILE *fp;

    test_string = PyUnicode_FromString("Spam spam spam");
    if (test_string == NULL) {
        return NULL;
    }

    // open file in read mode to induce OSError
    fp = Py_fopen(filename, "r");
    if (fp == NULL) {
        Py_DECREF(test_string);
        return NULL;
    }

    if (PyObject_Print(test_string, fp, 0) < 0) {
        fclose(fp);
        Py_DECREF(test_string);
        return NULL;
    }

    fclose(fp);
    Py_DECREF(test_string);

    Py_RETURN_NONE;
}

static PyObject *
pyobject_clear_weakrefs_no_callbacks(PyObject *self, PyObject *obj)
{
    PyUnstable_Object_ClearWeakRefsNoCallbacks(obj);
    Py_RETURN_NONE;
}

static PyObject *
pyobject_enable_deferred_refcount(PyObject *self, PyObject *obj)
{
    int result = PyUnstable_Object_EnableDeferredRefcount(obj);
    return PyLong_FromLong(result);
}

static PyObject *
pyobject_is_unique_temporary(PyObject *self, PyObject *obj)
{
    int result = PyUnstable_Object_IsUniqueReferencedTemporary(obj);
    return PyLong_FromLong(result);
}

static PyObject *
pyobject_is_unique_temporary_new_object(PyObject *self, PyObject *unused)
{
    PyObject *obj = PyList_New(0);
    int result = PyUnstable_Object_IsUniqueReferencedTemporary(obj);
    Py_DECREF(obj);
    return PyLong_FromLong(result);
}

static int MyObject_dealloc_called = 0;

static void
MyObject_dealloc(PyObject *op)
{
    // PyUnstable_TryIncRef should return 0 if object is being deallocated
    assert(Py_REFCNT(op) == 0);
    assert(!PyUnstable_TryIncRef(op));
    assert(Py_REFCNT(op) == 0);

    MyObject_dealloc_called++;
    Py_TYPE(op)->tp_free(op);
}

static PyTypeObject MyType = {
    PyVarObject_HEAD_INIT(NULL, 0)
    .tp_name = "MyType",
    .tp_basicsize = sizeof(PyObject),
    .tp_dealloc = MyObject_dealloc,
};

static PyObject *
test_py_try_inc_ref(PyObject *self, PyObject *unused)
{
    if (PyType_Ready(&MyType) < 0) {
        return NULL;
    }

    MyObject_dealloc_called = 0;

    PyObject *op = PyObject_New(PyObject, &MyType);
    if (op == NULL) {
        return NULL;
    }

    PyUnstable_EnableTryIncRef(op);
#ifdef Py_GIL_DISABLED
    // PyUnstable_EnableTryIncRef sets the shared flags to
    // `_Py_REF_MAYBE_WEAKREF` if the flags are currently zero to ensure that
    // the shared reference count is merged on deallocation.
    assert((op->ob_ref_shared & _Py_REF_SHARED_FLAG_MASK) >= _Py_REF_MAYBE_WEAKREF);
#endif

    if (!PyUnstable_TryIncRef(op)) {
        PyErr_SetString(PyExc_AssertionError, "PyUnstable_TryIncRef failed");
        Py_DECREF(op);
        return NULL;
    }
    Py_DECREF(op);  // undo try-incref
    Py_DECREF(op);  // dealloc
    assert(MyObject_dealloc_called == 1);
    Py_RETURN_NONE;
}

static PyObject *
test_py_set_immortal(PyObject *self, PyObject *unused)
{
    // the object is allocated on C stack as otherwise,
    // it would trip the refleak checker when the object
    // is made immortal and leak memory, for the same
    // reason we cannot call PyObject_Init() on it.
    PyObject object = {0};
#ifdef Py_GIL_DISABLED
    object.ob_tid = _Py_ThreadId();
    object.ob_gc_bits = 0;
    object.ob_ref_local = 1;
    object.ob_ref_shared = 0;
#else
    object.ob_refcnt = 1;
#endif
    object.ob_type = &PyBaseObject_Type;

    assert(!PyUnstable_IsImmortal(&object));
    int rc = PyUnstable_SetImmortal(&object);
    assert(rc == 1);
    assert(PyUnstable_IsImmortal(&object));
    Py_DECREF(&object);  // should not dealloc
    assert(PyUnstable_IsImmortal(&object));

    // Check already immortal object
    rc = PyUnstable_SetImmortal(&object);
    assert(rc == 0);

    // Check unicode objects
    PyObject *unicode = PyUnicode_FromString("test");
    assert(!PyUnstable_IsImmortal(unicode));
    rc = PyUnstable_SetImmortal(unicode);
    assert(rc == 0);
    assert(!PyUnstable_IsImmortal(unicode));
    Py_DECREF(unicode);
    Py_RETURN_NONE;
}

#ifdef Py_REF_DEBUG
static PyObject *
negative_refcount(PyObject *self, PyObject *Py_UNUSED(args))
{
    PyObject *obj = PyUnicode_FromString("negative_refcount");
    if (obj == NULL) {
        return NULL;
    }
    assert(Py_REFCNT(obj) == 1);

    Py_SET_REFCNT(obj,  0);
    /* Py_DECREF() must call _Py_NegativeRefcount() and abort Python */
    Py_DECREF(obj);

    Py_RETURN_NONE;
}


static PyObject *
decref_freed_object(PyObject *self, PyObject *Py_UNUSED(args))
{
    PyObject *obj = PyUnicode_FromString("decref_freed_object");
    if (obj == NULL) {
        return NULL;
    }
    assert(Py_REFCNT(obj) == 1);

    // Deallocate the memory
    Py_DECREF(obj);
    // obj is a now a dangling pointer

    // gh-109496: If Python is built in debug mode, Py_DECREF() must call
    // _Py_NegativeRefcount() and abort Python.
    Py_DECREF(obj);

    Py_RETURN_NONE;
}
#endif


static PyObject *
clear_managed_dict(PyObject *self, PyObject *obj)
{
    PyObject_ClearManagedDict(obj);
    Py_RETURN_NONE;
}


static PyObject *
is_uniquely_referenced(PyObject *self, PyObject *op)
{
    return PyBool_FromLong(PyUnstable_Object_IsUniquelyReferenced(op));
}


static PyObject *
pyobject_dump(PyObject *self, PyObject *args)
{
    PyObject *op;
    int release_gil = 0;

    if (!PyArg_ParseTuple(args, "O|i", &op, &release_gil)) {
        return NULL;
    }
    NULLABLE(op);

    if (release_gil) {
        Py_BEGIN_ALLOW_THREADS
        PyObject_Dump(op);
        Py_END_ALLOW_THREADS

    }
    else {
        PyObject_Dump(op);
    }
    Py_RETURN_NONE;
}

static PyObject *
pysentinel_new(PyObject *self, PyObject *args)
{
    const char *name;
    const char *module_name = NULL;
    const char *repr = NULL;
    if (!PyArg_ParseTuple(args, "s|ss", &name, &module_name, &repr)) {
        return NULL;
    }
    return PySentinel_New(name, module_name, repr);
}

static PyObject *
pysentinel_check(PyObject *self, PyObject *obj)
{
    return PyBool_FromLong(PySentinel_Check(obj));
}

static PyObject *
pysentinel_checkexact(PyObject *self, PyObject *obj)
{
    return PyBool_FromLong(PySentinel_CheckExact(obj));
}


static PyMethodDef test_methods[] = {
    {"call_pyobject_print", call_pyobject_print, METH_VARARGS},
    {"pyobject_print_null", pyobject_print_null, METH_O},
    {"pyobject_print_noref_object", pyobject_print_noref_object, METH_O},
    {"pyobject_print_os_error", pyobject_print_os_error, METH_O},
    {"pyobject_clear_weakrefs_no_callbacks", pyobject_clear_weakrefs_no_callbacks, METH_O},
    {"pyobject_enable_deferred_refcount", pyobject_enable_deferred_refcount, METH_O},
    {"pyobject_is_unique_temporary", pyobject_is_unique_temporary, METH_O},
    {"pyobject_is_unique_temporary_new_object", pyobject_is_unique_temporary_new_object, METH_NOARGS},
    {"test_py_try_inc_ref", test_py_try_inc_ref, METH_NOARGS},
    {"test_py_set_immortal", test_py_set_immortal, METH_NOARGS},
#ifdef Py_REF_DEBUG
    {"negative_refcount", negative_refcount, METH_NOARGS},
    {"decref_freed_object", decref_freed_object, METH_NOARGS},
#endif
    {"clear_managed_dict", clear_managed_dict, METH_O, NULL},
    {"is_uniquely_referenced", is_uniquely_referenced, METH_O},
    {"pyobject_dump", pyobject_dump, METH_VARARGS},
    {"pysentinel_new", pysentinel_new, METH_VARARGS},
    {"pysentinel_check", pysentinel_check, METH_O},
    {"pysentinel_checkexact", pysentinel_checkexact, METH_O},
    {NULL},
};

int
_PyTestCapi_Init_Object(PyObject *m)
{
    return PyModule_AddFunctions(m, test_methods);
}