#ifndef Py_BUILD_CORE_BUILTIN

# define Py_BUILD_CORE_MODULE 1

#undef NDEBUG

#include "Python.h"

#include <string.h>

#include "pycore_backoff.h" // JUMP_BACKWARD_INITIAL_VALUE

#include "pycore_bitutils.h" // _Py_bswap32()

#include "pycore_bytesobject.h" // _PyBytes_Find()

#include "pycore_ceval.h" // _PyEval_AddPendingCall()

#include "pycore_code.h" // _PyCode_GetTLBCFast()

#include "pycore_compile.h" // _PyCompile_CodeGen()

#include "pycore_context.h" // _PyContext_NewHamtForTests()

#include "pycore_dict.h" // PyDictValues

#include "pycore_fileutils.h" // _Py_normpath()

#include "pycore_flowgraph.h" // _PyCompile_OptimizeCfg()

#include "pycore_frame.h" // _PyInterpreterFrame

#include "pycore_function.h" // _PyFunction_GET_BUILTINS

#include "pycore_gc.h" // PyGC_Head

#include "pycore_hashtable.h" // _Py_hashtable_new()

#include "pycore_import.h" // _PyImport_ClearExtension()

#include "pycore_initconfig.h" // _Py_GetConfigsAsDict()

#include "pycore_instruction_sequence.h" // _PyInstructionSequence_New()

#include "pycore_interpframe.h" // _PyFrame_GetFunction()

#include "pycore_jit.h" // _PyJIT_AddressInJitCode()

#include "pycore_object.h" // _PyObject_IsFreed()

#include "pycore_optimizer.h" // _Py_Executor_DependsOn

#include "pycore_pathconfig.h" // _PyPathConfig_ClearGlobal()

#include "pycore_pyerrors.h" // _PyErr_ChainExceptions1()

#include "pycore_pylifecycle.h" // _PyInterpreterConfig_InitFromDict()

#include "pycore_pystate.h" // _PyThreadState_GET()

#include "pycore_runtime_structs.h" // _PY_NSMALLPOSINTS

#include "pycore_unicodeobject.h" // _PyUnicode_TransformDecimalAndSpaceToASCII()

#include "clinic/_testinternalcapi.c.h"

#include "_testinternalcapi/parts.h"

#if defined(HAVE_DLADDR) && !defined(__wasi__)

# include <dlfcn.h>

#if defined(HAVE_EXECINFO_H)

# include <execinfo.h>

#ifdef MS_WINDOWS

# include <windows.h>

# include <intrin.h>

# include <winnt.h>

# include <wchar.h>

#define MODULE_NAME "_testinternalcapi"

static const uintptr_t min_frame_pointer_addr = 0x1000;

#define MAX_UNWIND_FRAMES 200

#ifdef __s390x__

# define S390X_FRAME_RETURN_ADDRESS_OFFSET 112

#if defined(__arm__) && !defined(__thumb__) && !defined(__clang__)

# define FRAME_POINTER_NEXT_OFFSET (-1)

# define FRAME_POINTER_RETURN_OFFSET 0

#elif defined(__s390x__)

# define FRAME_POINTER_NEXT_OFFSET 0

# define FRAME_POINTER_RETURN_OFFSET \

(S390X_FRAME_RETURN_ADDRESS_OFFSET / (Py_ssize_t)sizeof(uintptr_t))

#elif defined(__powerpc64__) || defined(__ppc64__)

# define FRAME_POINTER_NEXT_OFFSET 0

# define FRAME_POINTER_RETURN_OFFSET 2

#elif defined(__riscv)

# define FRAME_POINTER_NEXT_OFFSET -2

# define FRAME_POINTER_RETURN_OFFSET -1

#elif defined(__loongarch__)

# define FRAME_POINTER_NEXT_OFFSET -2

# define FRAME_POINTER_RETURN_OFFSET -1

# define FRAME_POINTER_NEXT_OFFSET 0

# define FRAME_POINTER_RETURN_OFFSET 1

static PyObject *

_get_current_module(void)

{

// We ensured it was imported in _run_script().

PyObject *name = PyUnicode_FromString(MODULE_NAME);

if (name == NULL) {

return NULL;

}

PyObject *mod = PyImport_GetModule(name);

Py_DECREF(name);

if (mod == NULL) {

return NULL;

}

assert(mod != Py_None);

return mod;

}

typedef struct {

PyObject *record_list;

} module_state;

static inline module_state *

get_module_state(PyObject *mod)

{

assert(mod != NULL);

module_state *state = PyModule_GetState(mod);

assert(state != NULL);

return state;

}

static int

traverse_module_state(module_state *state, visitproc visit, void *arg)

{

Py_VISIT(state->record_list);

return 0;

}

static int

clear_module_state(module_state *state)

{

Py_CLEAR(state->record_list);

return 0;

}

static PyObject *

get_configs(PyObject *self, PyObject *Py_UNUSED(args))

{

return _Py_GetConfigsAsDict();

}

static PyObject*

get_recursion_depth(PyObject *self, PyObject *Py_UNUSED(args))

{

PyThreadState *tstate = _PyThreadState_GET();

return PyLong_FromLong(tstate->py_recursion_limit - tstate->py_recursion_remaining);

}

static PyObject*

get_c_recursion_remaining(PyObject *self, PyObject *Py_UNUSED(args))

{

PyThreadState *tstate = _PyThreadState_GET();

uintptr_t here_addr = _Py_get_machine_stack_pointer();

_PyThreadStateImpl *_tstate = (_PyThreadStateImpl *)tstate;

int remaining = (int)((here_addr - _tstate->c_stack_soft_limit) / _PyOS_STACK_MARGIN_BYTES * 50);

return PyLong_FromLong(remaining);

}

static PyObject*

get_stack_pointer(PyObject *self, PyObject *Py_UNUSED(args))

{

uintptr_t here_addr = _Py_get_machine_stack_pointer();

return PyLong_FromSize_t(here_addr);

}

static PyObject*

get_stack_margin(PyObject *self, PyObject *Py_UNUSED(args))

{

return PyLong_FromSize_t(_PyOS_STACK_MARGIN_BYTES);

}

#ifdef MS_WINDOWS

static const char *

classify_address(uintptr_t addr, int jit_enabled, PyInterpreterState *interp)

{

HMODULE module = NULL;

if (GetModuleHandleExW(

GET_MODULE_HANDLE_EX_FLAG_FROM_ADDRESS

| GET_MODULE_HANDLE_EX_FLAG_UNCHANGED_REFCOUNT,

(LPCWSTR)addr,

&module)) {

wchar_t path[MAX_PATH];

DWORD len = GetModuleFileNameW(module, path, Py_ARRAY_LENGTH(path));

if (len > 0 && len < Py_ARRAY_LENGTH(path)) {

const wchar_t *base = wcsrchr(path, L'\\');

base = base ? base + 1 : path;

if (_wcsnicmp(base, L"python", 6) == 0) {

return "python";

}

return "other";

}

/* Module resolved but path unavailable: treat as non-JIT. */

return "other";

}

#ifdef _Py_JIT

if (jit_enabled && _PyJIT_AddressInJitCode(interp, addr)) {

return "jit";

}

return "other";

}

#elif defined(HAVE_DLADDR) && !defined(__wasi__)

static const char *

classify_address(uintptr_t addr, int jit_enabled, PyInterpreterState *interp)

{

Dl_info info;

if (dladdr((void *)addr, &info) != 0

&& info.dli_fname != NULL

&& info.dli_fname[0] != '\0') {

const char *base = strrchr(info.dli_fname, '/');

base = base ? base + 1 : info.dli_fname;

if (strncmp(base, "python", 6) == 0) {

return "python";

}

#ifdef __CYGWIN__

// Match Cygwin "cygpython3.16.dll"

if (strncmp(base, "cygpython", 9) == 0) {

return "python";

}

// Match "libpython3.15.so.1.0"

if (strncmp(base, "libpython", 9) == 0) {

return "python";

}

return "other";

}

#ifdef _Py_JIT

if (jit_enabled && _PyJIT_AddressInJitCode(interp, addr)) {

return "jit";

}

return "other";

}

static const char *

classify_address(uintptr_t addr, int jit_enabled, PyInterpreterState *interp)

{

#ifdef _Py_JIT

if (jit_enabled && _PyJIT_AddressInJitCode(interp, addr)) {

return "jit";

}

return "other";

}

static PyObject *

classify_stack_addresses(PyObject *self, PyObject *args)

{

PyObject *seq = NULL;

int jit_enabled = 0;

if (!PyArg_ParseTuple(args, "O|p:classify_stack_addresses",

&seq, &jit_enabled)) {

return NULL;

}

PyObject *fast = PySequence_Fast(seq, "addresses must be iterable");

if (fast == NULL) {

return NULL;

}

Py_ssize_t n = PySequence_Fast_GET_SIZE(fast);

PyObject *labels = PyList_New(n);

if (labels == NULL) {

Py_DECREF(fast);

return NULL;

}

PyThreadState *tstate = _PyThreadState_GET();

PyInterpreterState *interp = tstate ? tstate->interp : NULL;

PyObject **items = PySequence_Fast_ITEMS(fast);

for (Py_ssize_t i = 0; i < n; i++) {

unsigned long long value = PyLong_AsUnsignedLongLong(items[i]);

if (PyErr_Occurred()) {

Py_DECREF(labels);

Py_DECREF(fast);

return NULL;

}

const char *label = classify_address((uintptr_t)value, jit_enabled, interp);

PyObject *label_obj = PyUnicode_FromString(label);

if (label_obj == NULL) {

Py_DECREF(labels);

Py_DECREF(fast);

return NULL;

}

PyList_SET_ITEM(labels, i, label_obj);

}

Py_DECREF(fast);

return labels;

}

static PyObject *

get_jit_code_ranges(PyObject *self, PyObject *Py_UNUSED(args))

{

PyObject *ranges = PyList_New(0);

if (ranges == NULL) {

return NULL;

}

#ifdef _Py_JIT

PyThreadState *tstate = _PyThreadState_GET();

PyInterpreterState *interp = tstate ? tstate->interp : NULL;

if (interp == NULL) {

return ranges;

}

for (size_t i = 0; i < interp->executor_count; i++) {

_PyExecutorObject *exec = interp->executor_ptrs[i];

if (exec->jit_code == NULL || exec->jit_size == 0) {

continue;

}

uintptr_t start = (uintptr_t)exec->jit_code;

uintptr_t end = start + exec->jit_size;

PyObject *start_obj = PyLong_FromUnsignedLongLong(start);

PyObject *end_obj = PyLong_FromUnsignedLongLong(end);

if (start_obj == NULL || end_obj == NULL) {

Py_XDECREF(start_obj);

Py_XDECREF(end_obj);

Py_DECREF(ranges);

return NULL;

}

PyObject *pair = PyTuple_New(2);

if (pair == NULL) {

Py_DECREF(start_obj);

Py_DECREF(end_obj);

Py_DECREF(ranges);

return NULL;

}

PyTuple_SET_ITEM(pair, 0, start_obj);

PyTuple_SET_ITEM(pair, 1, end_obj);

if (PyList_Append(ranges, pair) < 0) {

Py_DECREF(pair);

Py_DECREF(ranges);

return NULL;

}

Py_DECREF(pair);

}

return ranges;

}

static PyObject *

get_jit_backend(PyObject *self, PyObject *Py_UNUSED(args))

{

#ifdef _Py_JIT

return PyUnicode_FromString("jit");

#elif defined(_Py_TIER2)

return PyUnicode_FromString("interpreter");

Py_RETURN_NONE;

}

static int

stack_address_is_valid(uintptr_t addr, uintptr_t stack_min, uintptr_t stack_max)

{

if (addr < min_frame_pointer_addr) {

return 0;

}

if (stack_min != 0 && (addr < stack_min || addr >= stack_max)) {

return 0;

}

return 1;

}

static int

frame_pointer_slot_is_valid(uintptr_t *frame_pointer, Py_ssize_t offset,

uintptr_t stack_min, uintptr_t stack_max)

{

uintptr_t fp_addr = (uintptr_t)frame_pointer;

uintptr_t slot_addr;

uintptr_t delta = (uintptr_t)Py_ABS(offset) * sizeof(uintptr_t);

if (offset < 0) {

if (fp_addr < delta) {

return 0;

}

slot_addr = fp_addr - delta;

}

else {

if (fp_addr > UINTPTR_MAX - delta) {

return 0;

}

slot_addr = fp_addr + delta;

}

if (!stack_address_is_valid(slot_addr, stack_min, stack_max)) {

return 0;

}

if (stack_max != 0) {

if (slot_addr > UINTPTR_MAX - sizeof(uintptr_t)) {

return 0;

}

if (slot_addr + sizeof(uintptr_t) > stack_max) {

return 0;

}

}

return 1;

}

static int

next_frame_pointer_is_valid(uintptr_t *frame_pointer, uintptr_t *next_fp,

uintptr_t stack_min, uintptr_t stack_max)

{

uintptr_t fp_addr = (uintptr_t)frame_pointer;

uintptr_t next_addr = (uintptr_t)next_fp;

if (!stack_address_is_valid(next_addr, stack_min, stack_max)) {

return 0;

}

if ((next_addr % sizeof(uintptr_t)) != 0) {

return 0;

}

#if _Py_STACK_GROWS_DOWN

return next_addr > fp_addr;

return next_addr < fp_addr;

}

static PyObject *

manual_unwind_from_fp(uintptr_t *frame_pointer)

{

uintptr_t stack_min = 0;

uintptr_t stack_max = 0;

#ifdef __s390x__

Py_BUILD_ASSERT(S390X_FRAME_RETURN_ADDRESS_OFFSET % sizeof(uintptr_t) == 0);

if (frame_pointer == NULL) {

return PyList_New(0);

}

PyThreadState *tstate = _PyThreadState_GET();

if (tstate != NULL) {

_PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;

#if _Py_STACK_GROWS_DOWN

stack_min = tstate_impl->c_stack_hard_limit;

stack_max = tstate_impl->c_stack_top;

stack_min = tstate_impl->c_stack_top;

stack_max = tstate_impl->c_stack_hard_limit;

}

PyObject *result = PyList_New(0);

if (result == NULL) {

return NULL;

}

Py_ssize_t depth = 0;

while (frame_pointer != NULL) {

uintptr_t fp_addr = (uintptr_t)frame_pointer;

if ((fp_addr % sizeof(uintptr_t)) != 0) {

break;

}

if (depth >= MAX_UNWIND_FRAMES) {

Py_DECREF(result);

PyErr_Format(

PyExc_RuntimeError,

"manual frame pointer unwind returned more than %d frames",

MAX_UNWIND_FRAMES);

return NULL;

}

if (!stack_address_is_valid(fp_addr, stack_min, stack_max)) {

break;

}

if (!frame_pointer_slot_is_valid(frame_pointer,

FRAME_POINTER_NEXT_OFFSET,

stack_min, stack_max)) {

break;

}

if (!frame_pointer_slot_is_valid(frame_pointer,

FRAME_POINTER_RETURN_OFFSET,

stack_min, stack_max)) {

break;

}

uintptr_t *next_fp = (uintptr_t *)frame_pointer[FRAME_POINTER_NEXT_OFFSET];

uintptr_t return_addr = frame_pointer[FRAME_POINTER_RETURN_OFFSET];

PyObject *addr_obj = PyLong_FromUnsignedLongLong(return_addr);

if (addr_obj == NULL) {

Py_DECREF(result);

return NULL;

}

if (PyList_Append(result, addr_obj) < 0) {

Py_DECREF(addr_obj);

Py_DECREF(result);

return NULL;

}

Py_DECREF(addr_obj);

depth++;

if (!next_frame_pointer_is_valid(frame_pointer, next_fp,

stack_min, stack_max)) {

break;

}

frame_pointer = next_fp;

}

return result;

}

#if defined(HAVE_EXECINFO_H) && defined(HAVE_BACKTRACE)

static PyObject *

gnu_backtrace_unwind(PyObject *self, PyObject *Py_UNUSED(args))

{

void *addresses[MAX_UNWIND_FRAMES + 1];

int frame_count = backtrace(addresses, (int)Py_ARRAY_LENGTH(addresses));

if (frame_count < 0) {

PyErr_SetString(PyExc_RuntimeError, "backtrace() failed");

return NULL;

}

if (frame_count > MAX_UNWIND_FRAMES) {

PyErr_Format(

PyExc_RuntimeError,

"backtrace() returned more than %d frames",

MAX_UNWIND_FRAMES);

return NULL;

}

PyObject *result = PyList_New(frame_count);

if (result == NULL) {

return NULL;

}

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

PyObject *addr_obj = PyLong_FromUnsignedLongLong((uintptr_t)addresses[i]);

if (addr_obj == NULL) {

Py_DECREF(result);

return NULL;

}

PyList_SET_ITEM(result, i, addr_obj);

}

return result;

}

static PyObject *

gnu_backtrace_unwind(PyObject *self, PyObject *Py_UNUSED(args))

{

PyErr_SetString(PyExc_RuntimeError,

"gnu_backtrace_unwind is not supported on this platform");

return NULL;

}

#if defined(__GNUC__) || defined(__clang__)

static PyObject *

manual_frame_pointer_unwind(PyObject *self, PyObject *args)

{

uintptr_t *frame_pointer = (uintptr_t *)__builtin_frame_address(0);

return manual_unwind_from_fp(frame_pointer);

}

#elif defined(MS_WINDOWS) && defined(_M_ARM64)

static PyObject *

manual_frame_pointer_unwind(PyObject *self, PyObject *args)

{

CONTEXT ctx;

uintptr_t *frame_pointer = NULL;

RtlCaptureContext(&ctx);

frame_pointer = (uintptr_t *)ctx.Fp;

return manual_unwind_from_fp(frame_pointer);

}

static PyObject *

manual_frame_pointer_unwind(PyObject *self, PyObject *Py_UNUSED(args))

{

PyErr_SetString(PyExc_RuntimeError,

"manual_frame_pointer_unwind is not supported on this platform");

return NULL;

}

static PyObject*

test_bswap(PyObject *self, PyObject *Py_UNUSED(args))

{

uint16_t u16 = _Py_bswap16(UINT16_C(0x3412));

if (u16 != UINT16_C(0x1234)) {

PyErr_Format(PyExc_AssertionError,

"_Py_bswap16(0x3412) returns %d", u16);

return NULL;

}

uint32_t u32 = _Py_bswap32(UINT32_C(0x78563412));

if (u32 != UINT32_C(0x12345678)) {

PyErr_Format(PyExc_AssertionError,

"_Py_bswap32(0x78563412) returns %u", u32);

return NULL;

}

uint64_t u64 = _Py_bswap64(UINT64_C(0xEFCDAB9078563412));

if (u64 != UINT64_C(0x1234567890ABCDEF)) {

PyErr_Format(PyExc_AssertionError,

"_Py_bswap64(0xEFCDAB9078563412) returns %llu", u64);

return NULL;

}

Py_RETURN_NONE;

}

static int

check_popcount(uint32_t x, int expected)

{

// Use volatile to prevent the compiler to optimize out the whole test

volatile uint32_t u = x;

int bits = _Py_popcount32(u);

if (bits != expected) {

PyErr_Format(PyExc_AssertionError,

"_Py_popcount32(%lu) returns %i, expected %i",

(unsigned long)x, bits, expected);

return -1;

}

return 0;

}

static PyObject*

test_popcount(PyObject *self, PyObject *Py_UNUSED(args))

{

#define CHECK(X, RESULT) \

do { \

if (check_popcount(X, RESULT) < 0) { \

return NULL; \

} \

} while (0)

CHECK(0, 0);

CHECK(1, 1);

CHECK(0x08080808, 4);

CHECK(0x10000001, 2);

CHECK(0x10101010, 4);

CHECK(0x10204080, 4);

CHECK(0xDEADCAFE, 22);

CHECK(0xFFFFFFFF, 32);

Py_RETURN_NONE;

#undef CHECK

}

static int

check_bit_length(unsigned long x, int expected)

{

// Use volatile to prevent the compiler to optimize out the whole test

volatile unsigned long u = x;

int len = _Py_bit_length(u);

if (len != expected) {

PyErr_Format(PyExc_AssertionError,

"_Py_bit_length(%lu) returns %i, expected %i",

x, len, expected);

return -1;

}

return 0;

}

static PyObject*

test_bit_length(PyObject *self, PyObject *Py_UNUSED(args))

{

#define CHECK(X, RESULT) \

do { \

if (check_bit_length(X, RESULT) < 0) { \

return NULL; \

} \

} while (0)

CHECK(0, 0);

CHECK(1, 1);

CHECK(0x1000, 13);

CHECK(0x1234, 13);

CHECK(0x54321, 19);

CHECK(0x7FFFFFFF, 31);

CHECK(0xFFFFFFFF, 32);

Py_RETURN_NONE;

#undef CHECK

}

#define TO_PTR(ch) ((void*)(uintptr_t)ch)

#define FROM_PTR(ptr) ((uintptr_t)ptr)

#define VALUE(key) (1 + ((int)(key) - 'a'))

static Py_uhash_t

hash_char(const void *key)

{

char ch = (char)FROM_PTR(key);

return ch;

}

static int

hashtable_cb(_Py_hashtable_t *table,

const void *key_ptr, const void *value_ptr,

void *user_data)

{

int *count = (int *)user_data;

char key = (char)FROM_PTR(key_ptr);

int value = (int)FROM_PTR(value_ptr);

assert(value == VALUE(key));

*count += 1;

return 0;

}

static PyObject*

test_hashtable(PyObject *self, PyObject *Py_UNUSED(args))

{

_Py_hashtable_t *table = _Py_hashtable_new(hash_char,

_Py_hashtable_compare_direct);

if (table == NULL) {

return PyErr_NoMemory();

}

// Using an newly allocated table must not crash

assert(table->nentries == 0);

assert(table->nbuckets > 0);

assert(_Py_hashtable_get(table, TO_PTR('x')) == NULL);

// Test _Py_hashtable_set()

char key;

for (key='a'; key <= 'z'; key++) {

int value = VALUE(key);

if (_Py_hashtable_set(table, TO_PTR(key), TO_PTR(value)) < 0) {

_Py_hashtable_destroy(table);

return PyErr_NoMemory();

}

}

assert(table->nentries == 26);

assert(table->nbuckets > table->nentries);

// Test _Py_hashtable_get_entry()

for (key='a'; key <= 'z'; key++) {

_Py_hashtable_entry_t *entry = _Py_hashtable_get_entry(table, TO_PTR(key));

assert(entry != NULL);

assert(entry->key == TO_PTR(key));

assert(entry->value == TO_PTR(VALUE(key)));

}

// Test _Py_hashtable_get()

for (key='a'; key <= 'z'; key++) {

void *value_ptr = _Py_hashtable_get(table, TO_PTR(key));

assert((int)FROM_PTR(value_ptr) == VALUE(key));

}

// Test _Py_hashtable_steal()

key = 'p';

void *value_ptr = _Py_hashtable_steal(table, TO_PTR(key));

assert((int)FROM_PTR(value_ptr) == VALUE(key));

assert(table->nentries == 25);

assert(_Py_hashtable_get_entry(table, TO_PTR(key)) == NULL);

// Test _Py_hashtable_foreach()

int count = 0;

int res = _Py_hashtable_foreach(table, hashtable_cb, &count);

assert(res == 0);

assert(count == 25);

// Test _Py_hashtable_clear()

_Py_hashtable_clear(table);

assert(table->nentries == 0);

assert(table->nbuckets > 0);

assert(_Py_hashtable_get(table, TO_PTR('x')) == NULL);

_Py_hashtable_destroy(table);

Py_RETURN_NONE;

}

static PyObject *

test_reset_path_config(PyObject *Py_UNUSED(self), PyObject *Py_UNUSED(arg))

{

_PyPathConfig_ClearGlobal();

Py_RETURN_NONE;

}

static int

check_edit_cost(const char *a, const char *b, Py_ssize_t expected)

{

int ret = -1;

PyObject *a_obj = NULL;

PyObject *b_obj = NULL;

a_obj = PyUnicode_FromString(a);

if (a_obj == NULL) {

goto exit;

}

b_obj = PyUnicode_FromString(b);

if (b_obj == NULL) {

goto exit;

}

Py_ssize_t result = _Py_UTF8_Edit_Cost(a_obj, b_obj, -1);

if (result != expected) {

PyErr_Format(PyExc_AssertionError,

"Edit cost from '%s' to '%s' returns %zd, expected %zd",

a, b, result, expected);

goto exit;

}

// Check that smaller max_edits thresholds are exceeded.

Py_ssize_t max_edits = result;

while (max_edits > 0) {

max_edits /= 2;

Py_ssize_t result2 = _Py_UTF8_Edit_Cost(a_obj, b_obj, max_edits);

if (result2 <= max_edits) {

PyErr_Format(PyExc_AssertionError,

"Edit cost from '%s' to '%s' (threshold %zd) "

"returns %zd, expected greater than %zd",

a, b, max_edits, result2, max_edits);

goto exit;

}

}

// Check that bigger max_edits thresholds don't change anything

Py_ssize_t result3 = _Py_UTF8_Edit_Cost(a_obj, b_obj, result * 2 + 1);

if (result3 != result) {

PyErr_Format(PyExc_AssertionError,

"Edit cost from '%s' to '%s' (threshold %zd) "

"returns %zd, expected %zd",

a, b, result * 2, result3, result);

goto exit;

}

ret = 0;

exit:

Py_XDECREF(a_obj);

Py_XDECREF(b_obj);

return ret;

}

static PyObject *

test_edit_cost(PyObject *self, PyObject *Py_UNUSED(args))

{

#define CHECK(a, b, n) do { \

if (check_edit_cost(a, b, n) < 0) { \

return NULL; \

} \

} while (0) \

CHECK("", "", 0);

CHECK("", "a", 2);

CHECK("a", "A", 1);

CHECK("Apple", "Aple", 2);

CHECK("Banana", "B@n@n@", 6);

CHECK("Cherry", "Cherry!", 2);

CHECK("---0---", "------", 2);

CHECK("abc", "y", 6);

CHECK("aa", "bb", 4);

CHECK("aaaaa", "AAAAA", 5);

CHECK("wxyz", "wXyZ", 2);

CHECK("wxyz", "wXyZ123", 8);

CHECK("Python", "Java", 12);

CHECK("Java", "C#", 8);

CHECK("AbstractFoobarManager", "abstract_foobar_manager", 3+2*2);

CHECK("CPython", "PyPy", 10);

CHECK("CPython", "pypy", 11);

CHECK("AttributeError", "AttributeErrop", 2);

CHECK("AttributeError", "AttributeErrorTests", 10);

#undef CHECK

Py_RETURN_NONE;

}

static int

check_bytes_find(const char *haystack0, const char *needle0,

Py_ssize_t offset, Py_ssize_t expected)

{

Py_ssize_t len_haystack = strlen(haystack0);

Py_ssize_t len_needle = strlen(needle0);

Py_ssize_t result_1 = _PyBytes_Find(haystack0, len_haystack,

needle0, len_needle, offset);

if (result_1 != expected) {

PyErr_Format(PyExc_AssertionError,

"Incorrect result_1: '%s' in '%s' (offset=%zd)",

needle0, haystack0, offset);

return -1;

}

// Allocate new buffer with no NULL terminator.

char *haystack = PyMem_Malloc(len_haystack);

if (haystack == NULL) {

PyErr_NoMemory();

return -1;

}

char *needle = PyMem_Malloc(len_needle);

if (needle == NULL) {

PyMem_Free(haystack);

PyErr_NoMemory();

return -1;

}

memcpy(haystack, haystack0, len_haystack);

memcpy(needle, needle0, len_needle);

Py_ssize_t result_2 = _PyBytes_Find(haystack, len_haystack,

needle, len_needle, offset);

PyMem_Free(haystack);

PyMem_Free(needle);

if (result_2 != expected) {

PyErr_Format(PyExc_AssertionError,

"Incorrect result_2: '%s' in '%s' (offset=%zd)",

needle0, haystack0, offset);

return -1;

}

return 0;

}

static int

check_bytes_find_large(Py_ssize_t len_haystack, Py_ssize_t len_needle,

const char *needle)

{

char *zeros = PyMem_RawCalloc(len_haystack, 1);

if (zeros == NULL) {

PyErr_NoMemory();

return -1;

}

Py_ssize_t res = _PyBytes_Find(zeros, len_haystack, needle, len_needle, 0);

PyMem_RawFree(zeros);

if (res != -1) {

PyErr_Format(PyExc_AssertionError,

"check_bytes_find_large(%zd, %zd) found %zd",

len_haystack, len_needle, res);

return -1;

}

return 0;

}

static PyObject *

test_bytes_find(PyObject *self, PyObject *Py_UNUSED(args))

{

#define CHECK(H, N, O, E) do { \

if (check_bytes_find(H, N, O, E) < 0) { \

return NULL; \

} \

} while (0)

CHECK("", "", 0, 0);

CHECK("Python", "", 0, 0);

CHECK("Python", "", 3, 3);

CHECK("Python", "", 6, 6);

CHECK("Python", "yth", 0, 1);

CHECK("ython", "yth", 1, 1);

CHECK("thon", "yth", 2, -1);

CHECK("Python", "thon", 0, 2);

CHECK("ython", "thon", 1, 2);

CHECK("thon", "thon", 2, 2);

CHECK("hon", "thon", 3, -1);

CHECK("Pytho", "zz", 0, -1);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "ab", 0, -1);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "ba", 0, -1);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "bb", 0, -1);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaab", "ab", 0, 30);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaba", "ba", 0, 30);

CHECK("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaabb", "bb", 0, 30);

#undef CHECK

// Hunt for segfaults

// n, m chosen here so that (n - m) % (m + 1) == 0

// This would make default_find in fastsearch.h access haystack[n].

if (check_bytes_find_large(2048, 2, "ab") < 0) {

return NULL;

}

if (check_bytes_find_large(4096, 16, "0123456789abcdef") < 0) {

return NULL;

}

if (check_bytes_find_large(8192, 2, "ab") < 0) {

return NULL;

}

if (check_bytes_find_large(16384, 4, "abcd") < 0) {