/*

* C Extension module to test Python interpreter C APIs.

*

* The 'test_*' functions exported by this module are run as part of the

* standard Python regression test, via Lib/test/test_capi.py.

*/

// Include parts.h first since it takes care of NDEBUG and Py_BUILD_CORE macros

// and including Python.h.

//

// Several parts of this module are broken out into files in _testcapi/.

// Include definitions from there.

#include "_testcapi/parts.h"

#include "frameobject.h" // PyFrame_New()

#include "interpreteridobject.h" // PyInterpreterID_Type

#include "marshal.h" // PyMarshal_WriteLongToFile()

#include <float.h> // FLT_MAX

#include <signal.h>

#include <stddef.h> // offsetof()

#ifdef HAVE_SYS_WAIT_H

# include <sys/wait.h> // W_STOPCODE

#endif

#ifdef bool

# error "The public headers should not include <stdbool.h>, see gh-48924"

#endif

#include "_testcapi/util.h"

// Forward declarations

static struct PyModuleDef _testcapimodule;

// Module state

typedef struct {

PyObject *error; // _testcapi.error object

} testcapistate_t;

static testcapistate_t*

get_testcapi_state(PyObject *module)

{

void *state = PyModule_GetState(module);

assert(state != NULL);

return (testcapistate_t *)state;

}

static PyObject *

get_testerror(PyObject *self) {

testcapistate_t *state = get_testcapi_state(self);

return state->error;

}

/* Raise _testcapi.error with test_name + ": " + msg, and return NULL. */

static PyObject *

raiseTestError(PyObject *self, const char* test_name, const char* msg)

{

PyErr_Format(get_testerror(self), "%s: %s", test_name, msg);

return NULL;

}

/* Test #defines from pyconfig.h (particularly the SIZEOF_* defines).

The ones derived from autoconf on the UNIX-like OSes can be relied

upon (in the absence of sloppy cross-compiling), but the Windows

platforms have these hardcoded. Better safe than sorry.

*/

static PyObject*

sizeof_error(PyObject *self, const char* fatname, const char* typname,

int expected, int got)

{

PyErr_Format(get_testerror(self),

"%s #define == %d but sizeof(%s) == %d",

fatname, expected, typname, got);

return (PyObject*)NULL;

}

static PyObject*

test_config(PyObject *self, PyObject *Py_UNUSED(ignored))

{

#define CHECK_SIZEOF(FATNAME, TYPE) \

if (FATNAME != sizeof(TYPE)) \

return sizeof_error(self, #FATNAME, #TYPE, FATNAME, sizeof(TYPE))

CHECK_SIZEOF(SIZEOF_SHORT, short);

CHECK_SIZEOF(SIZEOF_INT, int);

CHECK_SIZEOF(SIZEOF_LONG, long);

CHECK_SIZEOF(SIZEOF_VOID_P, void*);

CHECK_SIZEOF(SIZEOF_TIME_T, time_t);

CHECK_SIZEOF(SIZEOF_LONG_LONG, long long);

#undef CHECK_SIZEOF

Py_RETURN_NONE;

}

static PyObject*

test_sizeof_c_types(PyObject *self, PyObject *Py_UNUSED(ignored))

{

#if defined(__GNUC__) && ((__GNUC__ > 4) || ((__GNUC__ == 4) && (__GNUC_MINOR__ > 5)))

#pragma GCC diagnostic push

#pragma GCC diagnostic ignored "-Wtype-limits"

#endif

#define CHECK_SIZEOF(TYPE, EXPECTED) \

if (EXPECTED != sizeof(TYPE)) { \

PyErr_Format(get_testerror(self), \

"sizeof(%s) = %u instead of %u", \

#TYPE, sizeof(TYPE), EXPECTED); \

return (PyObject*)NULL; \

}

#define IS_SIGNED(TYPE) (((TYPE)-1) < (TYPE)0)

#define CHECK_SIGNNESS(TYPE, SIGNED) \

if (IS_SIGNED(TYPE) != SIGNED) { \

PyErr_Format(get_testerror(self), \

"%s signness is %i, instead of %i", \

#TYPE, IS_SIGNED(TYPE), SIGNED); \

return (PyObject*)NULL; \

}

/* integer types */

CHECK_SIZEOF(Py_UCS1, 1);

CHECK_SIZEOF(Py_UCS2, 2);

CHECK_SIZEOF(Py_UCS4, 4);

CHECK_SIGNNESS(Py_UCS1, 0);

CHECK_SIGNNESS(Py_UCS2, 0);

CHECK_SIGNNESS(Py_UCS4, 0);

CHECK_SIZEOF(int32_t, 4);

CHECK_SIGNNESS(int32_t, 1);

CHECK_SIZEOF(uint32_t, 4);

CHECK_SIGNNESS(uint32_t, 0);

CHECK_SIZEOF(int64_t, 8);

CHECK_SIGNNESS(int64_t, 1);

CHECK_SIZEOF(uint64_t, 8);

CHECK_SIGNNESS(uint64_t, 0);

/* pointer/size types */

CHECK_SIZEOF(size_t, sizeof(void *));

CHECK_SIGNNESS(size_t, 0);

CHECK_SIZEOF(Py_ssize_t, sizeof(void *));

CHECK_SIGNNESS(Py_ssize_t, 1);

CHECK_SIZEOF(uintptr_t, sizeof(void *));

CHECK_SIGNNESS(uintptr_t, 0);

CHECK_SIZEOF(intptr_t, sizeof(void *));

CHECK_SIGNNESS(intptr_t, 1);

Py_RETURN_NONE;

#undef IS_SIGNED

#undef CHECK_SIGNESS

#undef CHECK_SIZEOF

#if defined(__GNUC__) && ((__GNUC__ > 4) || ((__GNUC__ == 4) && (__GNUC_MINOR__ > 5)))

#pragma GCC diagnostic pop

#endif

}

static PyObject*

test_list_api(PyObject *self, PyObject *Py_UNUSED(ignored))

{

PyObject* list;

int i;

/* SF bug 132008: PyList_Reverse segfaults */

#define NLIST 30

list = PyList_New(NLIST);

if (list == (PyObject*)NULL)

return (PyObject*)NULL;

/* list = range(NLIST) */

for (i = 0; i < NLIST; ++i) {

PyObject* anint = PyLong_FromLong(i);

if (anint == (PyObject*)NULL) {

Py_DECREF(list);

return (PyObject*)NULL;

}

PyList_SET_ITEM(list, i, anint);

}

/* list.reverse(), via PyList_Reverse() */

i = PyList_Reverse(list); /* should not blow up! */

if (i != 0) {

Py_DECREF(list);

return (PyObject*)NULL;

}

/* Check that list == range(29, -1, -1) now */

for (i = 0; i < NLIST; ++i) {

PyObject* anint = PyList_GET_ITEM(list, i);

if (PyLong_AS_LONG(anint) != NLIST-1-i) {

PyErr_SetString(get_testerror(self),

"test_list_api: reverse screwed up");

Py_DECREF(list);

return (PyObject*)NULL;

}

}

Py_DECREF(list);

#undef NLIST

Py_RETURN_NONE;

}

static int

test_dict_inner(PyObject *self, int count)

{

Py_ssize_t pos = 0, iterations = 0;

int i;

PyObject *dict = PyDict_New();

PyObject *v, *k;

if (dict == NULL)

return -1;

for (i = 0; i < count; i++) {

v = PyLong_FromLong(i);

if (v == NULL) {

goto error;

}

if (PyDict_SetItem(dict, v, v) < 0) {

Py_DECREF(v);

goto error;

}

Py_DECREF(v);

}

k = v = UNINITIALIZED_PTR;

while (PyDict_Next(dict, &pos, &k, &v)) {

PyObject *o;

iterations++;

assert(k != UNINITIALIZED_PTR);

assert(v != UNINITIALIZED_PTR);

i = PyLong_AS_LONG(v) + 1;

o = PyLong_FromLong(i);

if (o == NULL) {

goto error;

}

if (PyDict_SetItem(dict, k, o) < 0) {

Py_DECREF(o);

goto error;

}

Py_DECREF(o);

k = v = UNINITIALIZED_PTR;

}

assert(k == UNINITIALIZED_PTR);

assert(v == UNINITIALIZED_PTR);

Py_DECREF(dict);

if (iterations != count) {

PyErr_SetString(

get_testerror(self),

"test_dict_iteration: dict iteration went wrong ");

return -1;

} else {

return 0;

}

error:

Py_DECREF(dict);

return -1;

}

static PyObject*

test_dict_iteration(PyObject* self, PyObject *Py_UNUSED(ignored))

{

int i;

for (i = 0; i < 200; i++) {

if (test_dict_inner(self, i) < 0) {

return NULL;

}

}

Py_RETURN_NONE;

}

/* Issue #4701: Check that PyObject_Hash implicitly calls

* PyType_Ready if it hasn't already been called

*/

static PyTypeObject _HashInheritanceTester_Type = {

PyVarObject_HEAD_INIT(NULL, 0)

"hashinheritancetester", /* Name of this type */

sizeof(PyObject), /* Basic object size */

0, /* Item size for varobject */

(destructor)PyObject_Del, /* tp_dealloc */

0, /* tp_vectorcall_offset */

0, /* tp_getattr */

0, /* tp_setattr */

0, /* tp_as_async */

0, /* tp_repr */

0, /* tp_as_number */

0, /* tp_as_sequence */

0, /* tp_as_mapping */

0, /* tp_hash */

0, /* tp_call */

0, /* tp_str */

PyObject_GenericGetAttr, /* tp_getattro */

0, /* tp_setattro */

0, /* tp_as_buffer */

Py_TPFLAGS_DEFAULT, /* tp_flags */

0, /* tp_doc */

0, /* tp_traverse */

0, /* tp_clear */

0, /* tp_richcompare */

0, /* tp_weaklistoffset */

0, /* tp_iter */

0, /* tp_iternext */

0, /* tp_methods */

0, /* tp_members */

0, /* tp_getset */

0, /* tp_base */

0, /* tp_dict */

0, /* tp_descr_get */

0, /* tp_descr_set */

0, /* tp_dictoffset */

0, /* tp_init */

0, /* tp_alloc */

PyType_GenericNew, /* tp_new */

};

static PyObject*

pycompilestring(PyObject* self, PyObject *obj) {

if (PyBytes_CheckExact(obj) == 0) {

PyErr_SetString(PyExc_ValueError, "Argument must be a bytes object");

return NULL;

}

const char *the_string = PyBytes_AsString(obj);

if (the_string == NULL) {

return NULL;

}

return Py_CompileString(the_string, "<string>", Py_file_input);

}

static PyObject*

test_lazy_hash_inheritance(PyObject* self, PyObject *Py_UNUSED(ignored))

{

PyTypeObject *type;

PyObject *obj;

Py_hash_t hash;

type = &_HashInheritanceTester_Type;

if (type->tp_dict != NULL)

/* The type has already been initialized. This probably means

-R is being used. */

Py_RETURN_NONE;

obj = PyObject_New(PyObject, type);

if (obj == NULL) {

PyErr_Clear();

PyErr_SetString(

get_testerror(self),

"test_lazy_hash_inheritance: failed to create object");

return NULL;

}

if (type->tp_dict != NULL) {

PyErr_SetString(

get_testerror(self),

"test_lazy_hash_inheritance: type initialised too soon");

Py_DECREF(obj);

return NULL;

}

hash = PyObject_Hash(obj);

if ((hash == -1) && PyErr_Occurred()) {

PyErr_Clear();

PyErr_SetString(

get_testerror(self),

"test_lazy_hash_inheritance: could not hash object");

Py_DECREF(obj);

return NULL;

}

if (type->tp_dict == NULL) {

PyErr_SetString(

get_testerror(self),

"test_lazy_hash_inheritance: type not initialised by hash()");

Py_DECREF(obj);

return NULL;

}

if (type->tp_hash != PyType_Type.tp_hash) {

PyErr_SetString(

get_testerror(self),

"test_lazy_hash_inheritance: unexpected hash function");

Py_DECREF(obj);

return NULL;

}

Py_DECREF(obj);

Py_RETURN_NONE;

}

static PyObject *

return_none(void *unused)

{

Py_RETURN_NONE;

}

static PyObject *

raise_error(void *unused)

{

PyErr_SetNone(PyExc_ValueError);

return NULL;

}

static PyObject *

py_buildvalue(PyObject *self, PyObject *args)

{

const char *fmt;

PyObject *objs[10] = {NULL};

if (!PyArg_ParseTuple(args, "s|OOOOOOOOOO", &fmt,

&objs[0], &objs[1], &objs[2], &objs[3], &objs[4],

&objs[5], &objs[6], &objs[7], &objs[8], &objs[9]))

{

return NULL;

}

for(int i = 0; i < 10; i++) {

NULLABLE(objs[i]);

}

return Py_BuildValue(fmt,

objs[0], objs[1], objs[2], objs[3], objs[4],

objs[5], objs[6], objs[7], objs[8], objs[9]);

}

static PyObject *

py_buildvalue_ints(PyObject *self, PyObject *args)

{

const char *fmt;

unsigned int values[10] = {0};

if (!PyArg_ParseTuple(args, "s|IIIIIIIIII", &fmt,

&values[0], &values[1], &values[2], &values[3], &values[4],

&values[5], &values[6], &values[7], &values[8], &values[9]))

{

return NULL;

}

return Py_BuildValue(fmt,

values[0], values[1], values[2], values[3], values[4],

values[5], values[6], values[7], values[8], values[9]);

}

static int

test_buildvalue_N_error(PyObject *self, const char *fmt)

{

PyObject *arg, *res;

arg = PyList_New(0);

if (arg == NULL) {

return -1;

}

Py_INCREF(arg);

res = Py_BuildValue(fmt, return_none, NULL, arg);

if (res == NULL) {

return -1;

}

Py_DECREF(res);

if (Py_REFCNT(arg) != 1) {

PyErr_Format(get_testerror(self), "test_buildvalue_N: "

"arg was not decrefed in successful "

"Py_BuildValue(\"%s\")", fmt);

return -1;

}

Py_INCREF(arg);

res = Py_BuildValue(fmt, raise_error, NULL, arg);

if (res != NULL || !PyErr_Occurred()) {

PyErr_Format(get_testerror(self), "test_buildvalue_N: "

"Py_BuildValue(\"%s\") didn't complain", fmt);

return -1;

}

PyErr_Clear();

if (Py_REFCNT(arg) != 1) {

PyErr_Format(get_testerror(self), "test_buildvalue_N: "

"arg was not decrefed in failed "

"Py_BuildValue(\"%s\")", fmt);

return -1;

}

Py_DECREF(arg);

return 0;

}

static PyObject *

test_buildvalue_N(PyObject *self, PyObject *Py_UNUSED(ignored))

{

PyObject *arg, *res;

arg = PyList_New(0);

if (arg == NULL) {

return NULL;

}

Py_INCREF(arg);

res = Py_BuildValue("N", arg);

if (res == NULL) {

return NULL;

}

if (res != arg) {

return raiseTestError(self, "test_buildvalue_N",

"Py_BuildValue(\"N\") returned wrong result");

}

if (Py_REFCNT(arg) != 2) {

return raiseTestError(self, "test_buildvalue_N",

"arg was not decrefed in Py_BuildValue(\"N\")");

}

Py_DECREF(res);

Py_DECREF(arg);

if (test_buildvalue_N_error(self, "O&N") < 0)

return NULL;

if (test_buildvalue_N_error(self, "(O&N)") < 0)

return NULL;

if (test_buildvalue_N_error(self, "[O&N]") < 0)

return NULL;

if (test_buildvalue_N_error(self, "{O&N}") < 0)

return NULL;

if (test_buildvalue_N_error(self, "{()O&(())N}") < 0)

return NULL;

Py_RETURN_NONE;

}

static PyObject *

test_get_statictype_slots(PyObject *self, PyObject *Py_UNUSED(ignored))

{

newfunc tp_new = PyType_GetSlot(&PyLong_Type, Py_tp_new);

if (PyLong_Type.tp_new != tp_new) {

PyErr_SetString(PyExc_AssertionError, "mismatch: tp_new of long");

return NULL;

}

reprfunc tp_repr = PyType_GetSlot(&PyLong_Type, Py_tp_repr);

if (PyLong_Type.tp_repr != tp_repr) {

PyErr_SetString(PyExc_AssertionError, "mismatch: tp_repr of long");

return NULL;

}

ternaryfunc tp_call = PyType_GetSlot(&PyLong_Type, Py_tp_call);

if (tp_call != NULL) {

PyErr_SetString(PyExc_AssertionError, "mismatch: tp_call of long");

return NULL;

}

binaryfunc nb_add = PyType_GetSlot(&PyLong_Type, Py_nb_add);

if (PyLong_Type.tp_as_number->nb_add != nb_add) {

PyErr_SetString(PyExc_AssertionError, "mismatch: nb_add of long");

return NULL;

}

lenfunc mp_length = PyType_GetSlot(&PyLong_Type, Py_mp_length);

if (mp_length != NULL) {

PyErr_SetString(PyExc_AssertionError, "mismatch: mp_length of long");

return NULL;

}

void *over_value = PyType_GetSlot(&PyLong_Type, Py_bf_releasebuffer + 1);

if (over_value != NULL) {

PyErr_SetString(PyExc_AssertionError, "mismatch: max+1 of long");

return NULL;

}

tp_new = PyType_GetSlot(&PyLong_Type, 0);

if (tp_new != NULL) {

PyErr_SetString(PyExc_AssertionError, "mismatch: slot 0 of long");

return NULL;

}

if (PyErr_ExceptionMatches(PyExc_SystemError)) {

// This is the right exception

PyErr_Clear();

}

else {

return NULL;

}

Py_RETURN_NONE;

}

static PyType_Slot HeapTypeNameType_slots[] = {

{0},

};

static PyType_Spec HeapTypeNameType_Spec = {

.name = "_testcapi.HeapTypeNameType",

.basicsize = sizeof(PyObject),

.flags = Py_TPFLAGS_DEFAULT,

.slots = HeapTypeNameType_slots,

};

static PyObject *

get_heaptype_for_name(PyObject *self, PyObject *Py_UNUSED(ignored))

{

return PyType_FromSpec(&HeapTypeNameType_Spec);

}

static PyObject *

test_get_type_name(PyObject *self, PyObject *Py_UNUSED(ignored))

{

PyObject *tp_name = PyType_GetName(&PyLong_Type);

assert(strcmp(PyUnicode_AsUTF8(tp_name), "int") == 0);

Py_DECREF(tp_name);

tp_name = PyType_GetName(&PyModule_Type);

assert(strcmp(PyUnicode_AsUTF8(tp_name), "module") == 0);

Py_DECREF(tp_name);

PyObject *HeapTypeNameType = PyType_FromSpec(&HeapTypeNameType_Spec);

if (HeapTypeNameType == NULL) {

Py_RETURN_NONE;

}

tp_name = PyType_GetName((PyTypeObject *)HeapTypeNameType);

assert(strcmp(PyUnicode_AsUTF8(tp_name), "HeapTypeNameType") == 0);

Py_DECREF(tp_name);

PyObject *name = PyUnicode_FromString("test_name");

if (name == NULL) {

goto done;

}

if (PyObject_SetAttrString(HeapTypeNameType, "__name__", name) < 0) {

Py_DECREF(name);

goto done;

}

tp_name = PyType_GetName((PyTypeObject *)HeapTypeNameType);

assert(strcmp(PyUnicode_AsUTF8(tp_name), "test_name") == 0);

Py_DECREF(name);

Py_DECREF(tp_name);

done:

Py_DECREF(HeapTypeNameType);

Py_RETURN_NONE;

}

static PyObject *

test_get_type_qualname(PyObject *self, PyObject *Py_UNUSED(ignored))

{

PyObject *tp_qualname = PyType_GetQualName(&PyLong_Type);

assert(strcmp(PyUnicode_AsUTF8(tp_qualname), "int") == 0);

Py_DECREF(tp_qualname);

tp_qualname = PyType_GetQualName(&PyODict_Type);

assert(strcmp(PyUnicode_AsUTF8(tp_qualname), "OrderedDict") == 0);

Py_DECREF(tp_qualname);

PyObject *HeapTypeNameType = PyType_FromSpec(&HeapTypeNameType_Spec);

if (HeapTypeNameType == NULL) {

Py_RETURN_NONE;

}

tp_qualname = PyType_GetQualName((PyTypeObject *)HeapTypeNameType);

assert(strcmp(PyUnicode_AsUTF8(tp_qualname), "HeapTypeNameType") == 0);

Py_DECREF(tp_qualname);

PyObject *spec_name = PyUnicode_FromString(HeapTypeNameType_Spec.name);

if (spec_name == NULL) {

goto done;

}

if (PyObject_SetAttrString(HeapTypeNameType,

"__qualname__", spec_name) < 0) {

Py_DECREF(spec_name);

goto done;

}

tp_qualname = PyType_GetQualName((PyTypeObject *)HeapTypeNameType);

assert(strcmp(PyUnicode_AsUTF8(tp_qualname),

"_testcapi.HeapTypeNameType") == 0);

Py_DECREF(spec_name);

Py_DECREF(tp_qualname);

done:

Py_DECREF(HeapTypeNameType);

Py_RETURN_NONE;

}

static PyObject *

test_get_type_dict(PyObject *self, PyObject *Py_UNUSED(ignored))

{

/* Test for PyType_GetDict */

// Assert ints have a `to_bytes` method

PyObject *long_dict = PyType_GetDict(&PyLong_Type);

assert(long_dict);

assert(PyDict_GetItemString(long_dict, "to_bytes")); // borrowed ref

Py_DECREF(long_dict);

// Make a new type, add an attribute to it and assert it's there

PyObject *HeapTypeNameType = PyType_FromSpec(&HeapTypeNameType_Spec);

assert(HeapTypeNameType);

assert(PyObject_SetAttrString(

HeapTypeNameType, "new_attr", Py_NewRef(Py_None)) >= 0);

PyObject *type_dict = PyType_GetDict((PyTypeObject*)HeapTypeNameType);

assert(type_dict);

assert(PyDict_GetItemString(type_dict, "new_attr")); // borrowed ref

Py_DECREF(HeapTypeNameType);

Py_DECREF(type_dict);

Py_RETURN_NONE;

}

static PyObject *

pyobject_repr_from_null(PyObject *self, PyObject *Py_UNUSED(ignored))

{

return PyObject_Repr(NULL);

}

static PyObject *

pyobject_str_from_null(PyObject *self, PyObject *Py_UNUSED(ignored))

{

return PyObject_Str(NULL);

}

static PyObject *

pyobject_bytes_from_null(PyObject *self, PyObject *Py_UNUSED(ignored))

{

return PyObject_Bytes(NULL);

}

static PyObject *

set_errno(PyObject *self, PyObject *args)

{

int new_errno;

if (!PyArg_ParseTuple(args, "i:set_errno", &new_errno))

return NULL;

errno = new_errno;

Py_RETURN_NONE;

}

/* test_thread_state spawns a thread of its own, and that thread releases

* `thread_done` when it's finished. The driver code has to know when the

* thread finishes, because the thread uses a PyObject (the callable) that

* may go away when the driver finishes. The former lack of this explicit

* synchronization caused rare segfaults, so rare that they were seen only

* on a Mac buildbot (although they were possible on any box).

*/

static PyThread_type_lock thread_done = NULL;

static int

_make_call(void *callable)

{

PyObject *rc;

int success;

PyGILState_STATE s = PyGILState_Ensure();

rc = PyObject_CallNoArgs((PyObject *)callable);

success = (rc != NULL);

Py_XDECREF(rc);

PyGILState_Release(s);

return success;

}

/* Same thing, but releases `thread_done` when it returns. This variant

* should be called only from threads spawned by test_thread_state().

*/

static void

_make_call_from_thread(void *callable)

{

_make_call(callable);

PyThread_release_lock(thread_done);

}

static PyObject *

test_thread_state(PyObject *self, PyObject *args)

{

PyObject *fn;

int success = 1;

if (!PyArg_ParseTuple(args, "O:test_thread_state", &fn))

return NULL;

if (!PyCallable_Check(fn)) {

PyErr_Format(PyExc_TypeError, "'%s' object is not callable",

Py_TYPE(fn)->tp_name);

return NULL;

}

thread_done = PyThread_allocate_lock();

if (thread_done == NULL)

return PyErr_NoMemory();

PyThread_acquire_lock(thread_done, 1);

/* Start a new thread with our callback. */

PyThread_start_new_thread(_make_call_from_thread, fn);

/* Make the callback with the thread lock held by this thread */

success &= _make_call(fn);

/* Do it all again, but this time with the thread-lock released */

Py_BEGIN_ALLOW_THREADS

success &= _make_call(fn);

PyThread_acquire_lock(thread_done, 1); /* wait for thread to finish */

Py_END_ALLOW_THREADS

/* And once more with and without a thread

XXX - should use a lock and work out exactly what we are trying

to test <wink>

*/

Py_BEGIN_ALLOW_THREADS

PyThread_start_new_thread(_make_call_from_thread, fn);

success &= _make_call(fn);

PyThread_acquire_lock(thread_done, 1); /* wait for thread to finish */

Py_END_ALLOW_THREADS

/* Release lock we acquired above. This is required on HP-UX. */

PyThread_release_lock(thread_done);

PyThread_free_lock(thread_done);

if (!success)

return NULL;

Py_RETURN_NONE;

}

#ifndef MS_WINDOWS

static PyThread_type_lock wait_done = NULL;

static void wait_for_lock(void *unused) {

PyThread_acquire_lock(wait_done, 1);

PyThread_release_lock(wait_done);

PyThread_free_lock(wait_done);

wait_done = NULL;

}

// These can be used to test things that care about the existence of another

// thread that the threading module doesn't know about.

static PyObject *

spawn_pthread_waiter(PyObject *self, PyObject *Py_UNUSED(ignored))

{

if (wait_done) {

PyErr_SetString(PyExc_RuntimeError, "thread already running");

return NULL;

}

wait_done = PyThread_allocate_lock();

if (wait_done == NULL)

return PyErr_NoMemory();

PyThread_acquire_lock(wait_done, 1);

PyThread_start_new_thread(wait_for_lock, NULL);

Py_RETURN_NONE;

}

static PyObject *

end_spawned_pthread(PyObject *self, PyObject *Py_UNUSED(ignored))

{

if (!wait_done) {

PyErr_SetString(PyExc_RuntimeError, "call _spawn_pthread_waiter 1st");

return NULL;

}

PyThread_release_lock(wait_done);

Py_RETURN_NONE;

}

#endif // not MS_WINDOWS

/* test Py_AddPendingCalls using threads */

static int _pending_callback(void *arg)

{

/* we assume the argument is callable object to which we own a reference */

PyObject *callable = (PyObject *)arg;

PyObject *r = PyObject_CallNoArgs(callable);

Py_DECREF(callable);

Py_XDECREF(r);

return r != NULL ? 0 : -1;

}

/* The following requests n callbacks to _pending_callback. It can be

* run from any python thread.

*/

static PyObject *

pending_threadfunc(PyObject *self, PyObject *arg)

{

PyObject *callable;

int r;

if (PyArg_ParseTuple(arg, "O", &callable) == 0)

return NULL;

/* create the reference for the callbackwhile we hold the lock */

Py_INCREF(callable);

Py_BEGIN_ALLOW_THREADS

r = Py_AddPendingCall(&_pending_callback, callable);

Py_END_ALLOW_THREADS

if (r<0) {

Py_DECREF(callable); /* unsuccessful add, destroy the extra reference */

Py_RETURN_FALSE;

}

Py_RETURN_TRUE;

}

/* Test PyOS_string_to_double. */

static PyObject *

test_string_to_double(PyObject *self, PyObject *Py_UNUSED(ignored)) {

double result;

const char *msg;

#define CHECK_STRING(STR, expected) \

result = PyOS_string_to_double(STR, NULL, NULL); \

if (result == -1.0 && PyErr_Occurred()) \

return NULL; \

if (result != (double)expected) { \

msg = "conversion of " STR " to float failed"; \

goto fail; \

}

#define CHECK_INVALID(STR) \

result = PyOS_string_to_double(STR, NULL, NULL); \

if (result == -1.0 && PyErr_Occurred()) { \

if (PyErr_ExceptionMatches(PyExc_ValueError)) \

PyErr_Clear(); \

else \

return NULL; \

} \

else { \

msg = "conversion of " STR " didn't raise ValueError"; \

goto fail; \

}

CHECK_STRING("0.1", 0.1);

CHECK_STRING("1.234", 1.234);

CHECK_STRING("-1.35", -1.35);

CHECK_STRING(".1e01", 1.0);

CHECK_STRING("2.e-2", 0.02);

CHECK_INVALID(" 0.1");

CHECK_INVALID("\t\n-3");

CHECK_INVALID(".123 ");

CHECK_INVALID("3\n");

CHECK_INVALID("123abc");

Py_RETURN_NONE;

fail:

return raiseTestError(self, "test_string_to_double", msg);

#undef CHECK_STRING

#undef CHECK_INVALID

}

/* Coverage testing of capsule objects. */

static const char *capsule_name = "capsule name";

static char *capsule_pointer = "capsule pointer";

static char *capsule_context = "capsule context";

static const char *capsule_error = NULL;

static int

capsule_destructor_call_count = 0;

static void

capsule_destructor(PyObject *o) {

capsule_destructor_call_count++;

if (PyCapsule_GetContext(o) != capsule_context) {

capsule_error = "context did not match in destructor!";

} else if (PyCapsule_GetDestructor(o) != capsule_destructor) {

capsule_error = "destructor did not match in destructor! (woah!)";

} else if (PyCapsule_GetName(o) != capsule_name) {

capsule_error = "name did not match in destructor!";

} else if (PyCapsule_GetPointer(o, capsule_name) != capsule_pointer) {

capsule_error = "pointer did not match in destructor!";

}

}

typedef struct {

char *name;

char *module;

char *attribute;

} known_capsule;

static PyObject *

test_capsule(PyObject *self, PyObject *Py_UNUSED(ignored))

{

PyObject *object;

const char *error = NULL;

void *pointer;

void *pointer2;

known_capsule known_capsules[] = {

#define KNOWN_CAPSULE(module, name) { module "." name, module, name }

KNOWN_CAPSULE("_socket", "CAPI"),

KNOWN_CAPSULE("_curses", "_C_API"),

KNOWN_CAPSULE("datetime", "datetime_CAPI"),

{ NULL, NULL },

};

known_capsule *known = &known_capsules[0];

#define FAIL(x) { error = (x); goto exit; }

#define CHECK_DESTRUCTOR \

if (capsule_error) { \

FAIL(capsule_error); \

} \

else if (!capsule_destructor_call_count) { \

FAIL("destructor not called!"); \

} \

capsule_destructor_call_count = 0; \

object = PyCapsule_New(capsule_pointer, capsule_name, capsule_destructor);

PyCapsule_SetContext(object, capsule_context);

capsule_destructor(object);

CHECK_DESTRUCTOR;

Py_DECREF(object);

CHECK_DESTRUCTOR;

object = PyCapsule_New(known, "ignored", NULL);

PyCapsule_SetPointer(object, capsule_pointer);