#ifndef Py_BUILD_CORE_MODULE

# define Py_BUILD_CORE_MODULE

#undef NDEBUG

#include <Python.h>

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

#include "pycore_runtime.h" // _PyRuntime

#include "pycore_lock.h" // PyEvent

#include "pycore_pythread.h" // PyThread_start_joinable_thread()

#include "pycore_pystate.h" // _PyInterpreterState_GuardCountdown

#include "pycore_import.h" // _PyImport_FrozenBootstrap

#include <inttypes.h>

#include <stdio.h>

#include <stdlib.h> // putenv()

#include <wchar.h>

extern void PySys_AddWarnOption(const wchar_t *s);

extern void PySys_AddXOption(const wchar_t *s);

extern void Py_SetPath(const wchar_t *path);

extern void Py_SetPythonHome(const wchar_t *);

extern void PySys_ResetWarnOptions(void);

PyAPI_DATA(int) Py_UTF8Mode;

int main_argc;

char **main_argv;

#define PROGRAM "test_embed"

#ifdef __CYGWIN__

# define PROGRAM_NAME L"./_testembed.exe"

# define PROGRAM_NAME_UTF8 "./_testembed.exe"

# define PROGRAM_NAME L"./_testembed"

# define PROGRAM_NAME_UTF8 "./_testembed"

#define INIT_LOOPS 4

static void error(const char *msg)

{

fprintf(stderr, "ERROR: %s\n", msg);

fflush(stderr);

}

static void error_fmt(const char *format, ...)

{

va_list vargs;

va_start(vargs, format);

fprintf(stderr, "ERROR: ");

vfprintf(stderr, format, vargs);

fprintf(stderr, "\n");

va_end(vargs);

fflush(stderr);

}

static wchar_t* py_getenv(const char *name)

{

const char *env = getenv(name);

if (env == NULL) {

error_fmt("need %s env var", name);

return NULL;

}

wchar_t *result = Py_DecodeLocale(env, NULL);

if (result == NULL) {

error("Py_DecodeLocale() failed");

return NULL;

}

return result;

}

static wchar_t* get_cmdline_arg(const char *arg_name)

{

if (main_argc < 3) {

const char *test = main_argv[1];

fprintf(stderr, "usage: %s %s %s\n", PROGRAM, test, arg_name);

return NULL;

}

const char *arg = main_argv[2];

wchar_t *result = Py_DecodeLocale(arg, NULL);

if (result == NULL) {

error_fmt("failed to decode %s command line argument", arg_name);

return NULL;

}

return result;

}

static void config_set_string(PyConfig *config, wchar_t **config_str, const wchar_t *str)

{

PyStatus status = PyConfig_SetString(config, config_str, str);

if (PyStatus_Exception(status)) {

PyConfig_Clear(config);

Py_ExitStatusException(status);

}

}

static void config_set_program_name(PyConfig *config)

{

const wchar_t *program_name = PROGRAM_NAME;

config_set_string(config, &config->program_name, program_name);

}

static void init_from_config_clear(PyConfig *config)

{

PyStatus status = Py_InitializeFromConfig(config);

PyConfig_Clear(config);

if (PyStatus_Exception(status)) {

Py_ExitStatusException(status);

}

}

static void _testembed_initialize(void)

{

PyConfig config;

_PyConfig_InitCompatConfig(&config);

config_set_program_name(&config);

init_from_config_clear(&config);

}

static int test_import_in_subinterpreters(void)

{

_testembed_initialize();

PyThreadState_Swap(Py_NewInterpreter());

return PyRun_SimpleString("import readline"); // gh-124160

}

static void print_subinterp(void)

{

/* Output information about the interpreter in the format

PyThreadState *ts = PyThreadState_Get();

PyInterpreterState *interp = ts->interp;

int64_t id = PyInterpreterState_GetID(interp);

printf("interp %" PRId64 " <0x%" PRIXPTR ">, thread state <0x%" PRIXPTR ">: ",

id, (uintptr_t)interp, (uintptr_t)ts);

fflush(stdout);

PyRun_SimpleString(

"import sys;"

"print('id(modules) =', id(sys.modules));"

"sys.stdout.flush()"

);

}

static int test_repeated_init_and_subinterpreters(void)

{

PyThreadState *mainstate, *substate;

PyGILState_STATE gilstate;

for (int i=1; i <= INIT_LOOPS; i++) {

printf("--- Pass %d ---\n", i);

_testembed_initialize();

mainstate = PyThreadState_Get();

PyEval_ReleaseThread(mainstate);

gilstate = PyGILState_Ensure();

print_subinterp();

PyThreadState_Swap(NULL);

for (int j=0; j<3; j++) {

substate = Py_NewInterpreter();

print_subinterp();

Py_EndInterpreter(substate);

}

PyThreadState_Swap(mainstate);

print_subinterp();

PyGILState_Release(gilstate);

PyEval_RestoreThread(mainstate);

Py_Finalize();

}

return 0;

}

#define EMBEDDED_EXT_NAME "embedded_ext"

static PyModuleDef embedded_ext = {

PyModuleDef_HEAD_INIT,

.m_name = EMBEDDED_EXT_NAME,

.m_size = 0,

};

static PyObject*

PyInit_embedded_ext(void)

{

// keep this as a single-phase initialization module;

// see test_create_module_from_initfunc

return PyModule_Create(&embedded_ext);

}

static int test_repeated_init_exec(void)

{

if (main_argc < 3) {

fprintf(stderr,

"usage: %s test_repeated_init_exec CODE ...\n", PROGRAM);

exit(1);

}

const char *code = main_argv[2];

int loops = main_argc > 3

? main_argc - 2

: INIT_LOOPS;

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

fprintf(stderr, "--- Loop #%d ---\n", i+1);

fflush(stderr);

if (main_argc > 3) {

code = main_argv[i+2];

}

_testembed_initialize();

int err = PyRun_SimpleString(code);

Py_Finalize();

if (err) {

return 1;

}

}

return 0;

}

static int test_interpreter_clear_and_delete(void)

{

assert(main_argc == 3);

_testembed_initialize();

if (PyRun_SimpleString(main_argv[2]) < 0) {

return 1;

}

PyInterpreterState *interp = PyInterpreterState_Get();

PyInterpreterState_Clear(interp);

PyThreadState_Swap(NULL);

PyInterpreterState_Delete(interp);

// The interpreter has already been deleted, so don't call Py_FinalizeEx().

return 0;

}

static int test_repeated_simple_init(void)

{

for (int i=1; i <= INIT_LOOPS; i++) {

fprintf(stderr, "--- Loop #%d ---\n", i);

fflush(stderr);

_testembed_initialize();

Py_Finalize();

printf("Finalized\n"); // Give test_embed some output to check

}

return 0;

}

static void check_stdio_details(const wchar_t *encoding, const wchar_t *errors)

{

/* Output info for the test case to check */

if (encoding) {

printf("Expected encoding: %ls\n", encoding);

} else {

printf("Expected encoding: default\n");

}

if (errors) {

printf("Expected errors: %ls\n", errors);

} else {

printf("Expected errors: default\n");

}

fflush(stdout);

PyConfig config;

_PyConfig_InitCompatConfig(&config);

/* Force the given IO encoding */

if (encoding) {

config_set_string(&config, &config.stdio_encoding, encoding);

}

if (errors) {

config_set_string(&config, &config.stdio_errors, errors);

}

#ifdef MS_WINDOWS

// gh-106659: On Windows, don't use _io._WindowsConsoleIO which always

// announce UTF-8 for sys.stdin.encoding.

config.legacy_windows_stdio = 1;

config_set_program_name(&config);

init_from_config_clear(&config);

PyRun_SimpleString(

"import sys;"

"print('stdin: {0.encoding}:{0.errors}'.format(sys.stdin));"

"print('stdout: {0.encoding}:{0.errors}'.format(sys.stdout));"

"print('stderr: {0.encoding}:{0.errors}'.format(sys.stderr));"

"sys.stdout.flush()"

);

Py_Finalize();

}

static int test_forced_io_encoding(void)

{

/* Check various combinations */

printf("--- Use defaults ---\n");

check_stdio_details(NULL, NULL);

printf("--- Set errors only ---\n");

check_stdio_details(NULL, L"ignore");

printf("--- Set encoding only ---\n");

check_stdio_details(L"iso8859-1", NULL);

printf("--- Set encoding and errors ---\n");

check_stdio_details(L"iso8859-1", L"replace");

return 0;

}

#define _Py_EMBED_PREINIT_CHECK(msg) \

do {printf(msg); fflush(stdout);} while (0);

static int test_pre_initialization_api(void)

{

/* the test doesn't support custom memory allocators */

putenv("PYTHONMALLOC=");

_Py_EMBED_PREINIT_CHECK("Initializing interpreter\n");

_testembed_initialize();

_Py_EMBED_PREINIT_CHECK("Checking Py_IsInitialized post-initialization\n");

if (!Py_IsInitialized()) {

fprintf(stderr, "Fatal error: not initialized after initialization!\n");

return 1;

}

_Py_EMBED_PREINIT_CHECK("Check sys module contents\n");

PyRun_SimpleString(

"import sys; "

"print('sys.executable:', sys.executable); "

"sys.stdout.flush(); "

);

_Py_EMBED_PREINIT_CHECK("Finalizing interpreter\n");

Py_Finalize();

_Py_EMBED_PREINIT_CHECK("Checking !Py_IsInitialized post-finalization\n");

if (Py_IsInitialized()) {

fprintf(stderr, "Fatal error: still initialized after finalization!\n");

return 1;

}

return 0;

}

static int test_pre_initialization_sys_options(void)

{

/* We allocate a couple of the options dynamically, and then delete

const wchar_t *static_warnoption = L"once";

const wchar_t *static_xoption = L"also_not_an_option=2";

size_t warnoption_len = wcslen(static_warnoption);

size_t xoption_len = wcslen(static_xoption);

wchar_t *dynamic_once_warnoption = \

(wchar_t *) calloc(warnoption_len+1, sizeof(wchar_t));

if (dynamic_once_warnoption == NULL) {

error("out of memory allocating warnoption");

return 1;

}

wchar_t *dynamic_xoption = \

(wchar_t *) calloc(xoption_len+1, sizeof(wchar_t));

if (dynamic_xoption == NULL) {

free(dynamic_once_warnoption);

error("out of memory allocating xoption");

return 1;

}

wcsncpy(dynamic_once_warnoption, static_warnoption, warnoption_len+1);

wcsncpy(dynamic_xoption, static_xoption, xoption_len+1);

_Py_EMBED_PREINIT_CHECK("Checking PySys_AddWarnOption\n");

PySys_AddWarnOption(L"default");

_Py_EMBED_PREINIT_CHECK("Checking PySys_ResetWarnOptions\n");

PySys_ResetWarnOptions();

_Py_EMBED_PREINIT_CHECK("Checking PySys_AddWarnOption linked list\n");

PySys_AddWarnOption(dynamic_once_warnoption);

PySys_AddWarnOption(L"module");

PySys_AddWarnOption(L"default");

_Py_EMBED_PREINIT_CHECK("Checking PySys_AddXOption\n");

PySys_AddXOption(L"not_an_option=1");

PySys_AddXOption(dynamic_xoption);

/* Delete the dynamic options early */

free(dynamic_once_warnoption);

dynamic_once_warnoption = NULL;

free(dynamic_xoption);

dynamic_xoption = NULL;

_Py_EMBED_PREINIT_CHECK("Initializing interpreter\n");

_testembed_initialize();

_Py_EMBED_PREINIT_CHECK("Check sys module contents\n");

PyRun_SimpleString(

"import sys; "

"print('sys.warnoptions:', sys.warnoptions); "

"print('sys._xoptions:', sys._xoptions); "

"warnings = sys.modules['warnings']; "

"latest_filters = [f[0] for f in warnings.filters[:3]]; "

"print('warnings.filters[:3]:', latest_filters); "

"sys.stdout.flush(); "

);

_Py_EMBED_PREINIT_CHECK("Finalizing interpreter\n");

Py_Finalize();

return 0;

}

static void bpo20891_thread(void *eventp)

{

PyEvent *event = (PyEvent *)eventp;

PyGILState_STATE state = PyGILState_Ensure();

if (!PyGILState_Check()) {

error("PyGILState_Check failed!");

abort();

}

PyGILState_Release(state);

_PyEvent_Notify(event);

}

static int test_bpo20891(void)

{

/* the test doesn't support custom memory allocators */

putenv("PYTHONMALLOC=");

/* bpo-20891: Calling PyGILState_Ensure in a non-Python thread must not

PyEvent event = {0};

_testembed_initialize();

unsigned long thrd = PyThread_start_new_thread(bpo20891_thread, &event);

if (thrd == PYTHREAD_INVALID_THREAD_ID) {

error("PyThread_start_new_thread failed!");

return 1;

}

PyEvent_Wait(&event);

Py_Finalize();

return 0;

}

static int test_initialize_twice(void)

{

_testembed_initialize();

/* bpo-33932: Calling Py_Initialize() twice should do nothing

Py_Initialize();

Py_Finalize();

return 0;

}

static int test_initialize_pymain(void)

{

wchar_t *argv[] = {L"PYTHON", L"-c",

(L"import sys; "

L"print(f'Py_Main() after Py_Initialize: "

L"sys.argv={sys.argv}')"),

L"arg2"};

_testembed_initialize();

/* bpo-34008: Calling Py_Main() after Py_Initialize() must not crash */

Py_Main(Py_ARRAY_LENGTH(argv), argv);

Py_Finalize();

return 0;

}

static void

dump_config(void)

{

(void) PyRun_SimpleStringFlags(

"import _testinternalcapi, json; "

"print(json.dumps(_testinternalcapi.get_configs()))",

0);

}

static int test_init_initialize_config(void)

{

_testembed_initialize();

dump_config();

Py_Finalize();

return 0;

}

static void config_set_argv(PyConfig *config, Py_ssize_t argc, wchar_t * const *argv)

{

PyStatus status = PyConfig_SetArgv(config, argc, argv);

if (PyStatus_Exception(status)) {

PyConfig_Clear(config);

Py_ExitStatusException(status);

}

}

static void

config_set_wide_string_list(PyConfig *config, PyWideStringList *list,

Py_ssize_t length, wchar_t **items)

{

PyStatus status = PyConfig_SetWideStringList(config, list, length, items);

if (PyStatus_Exception(status)) {

PyConfig_Clear(config);

Py_ExitStatusException(status);

}

}

static int check_init_compat_config(int preinit)

{

PyStatus status;

if (preinit) {

PyPreConfig preconfig;

_PyPreConfig_InitCompatConfig(&preconfig);

status = Py_PreInitialize(&preconfig);

if (PyStatus_Exception(status)) {

Py_ExitStatusException(status);

}

}

PyConfig config;

_PyConfig_InitCompatConfig(&config);

config_set_program_name(&config);

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int test_preinit_compat_config(void)

{

return check_init_compat_config(1);

}

static int test_init_compat_config(void)

{

return check_init_compat_config(0);

}

static int test_init_global_config(void)

{

putenv("PYTHONUTF8=0");

Py_UTF8Mode = 1;

_testembed_initialize();

dump_config();

Py_Finalize();

return 0;

}

static int test_init_from_config(void)

{

PyPreConfig preconfig;

_PyPreConfig_InitCompatConfig(&preconfig);

putenv("PYTHONMALLOC=malloc_debug");

#ifndef Py_GIL_DISABLED

preconfig.allocator = PYMEM_ALLOCATOR_MALLOC;

preconfig.allocator = PYMEM_ALLOCATOR_MIMALLOC;

putenv("PYTHONUTF8=0");

Py_UTF8Mode = 0;

preconfig.utf8_mode = 1;

PyStatus status = Py_PreInitialize(&preconfig);

if (PyStatus_Exception(status)) {

Py_ExitStatusException(status);

}

PyConfig config;

_PyConfig_InitCompatConfig(&config);

config.install_signal_handlers = 0;

/* FIXME: test use_environment */

putenv("PYTHONHASHSEED=42");

config.use_hash_seed = 1;

config.hash_seed = 123;

/* dev_mode=1 is tested in test_init_dev_mode() */

putenv("PYTHONFAULTHANDLER=");

config.faulthandler = 1;

putenv("PYTHONTRACEMALLOC=0");

config.tracemalloc = 2;

putenv("PYTHONPROFILEIMPORTTIME=1");

config.import_time = 2;

putenv("PYTHONNODEBUGRANGES=0");

config.code_debug_ranges = 0;

config.show_ref_count = 1;

/* FIXME: test dump_refs: bpo-34223 */

putenv("PYTHONMALLOCSTATS=0");

config.malloc_stats = 1;

config.pymalloc_hugepages = 1;

putenv("PYTHONPYCACHEPREFIX=env_pycache_prefix");

config_set_string(&config, &config.pycache_prefix, L"conf_pycache_prefix");

config_set_string(&config, &config.program_name, L"./conf_program_name");

wchar_t* argv[] = {

L"python3",

L"-W",

L"cmdline_warnoption",

L"-X",

L"cmdline_xoption",

L"-c",

L"pass",

L"arg2",

};

config_set_argv(&config, Py_ARRAY_LENGTH(argv), argv);

config.parse_argv = 1;

wchar_t* xoptions[3] = {

L"config_xoption1=3",

L"config_xoption2=",

L"config_xoption3",

};

config_set_wide_string_list(&config, &config.xoptions,

Py_ARRAY_LENGTH(xoptions), xoptions);

wchar_t* warnoptions[1] = {

L"config_warnoption",

};

config_set_wide_string_list(&config, &config.warnoptions,

Py_ARRAY_LENGTH(warnoptions), warnoptions);

/* FIXME: test pythonpath_env */

/* FIXME: test home */

/* FIXME: test path config: module_search_path .. dll_path */

putenv("PYTHONPLATLIBDIR=env_platlibdir");

config_set_string(&config, &config.platlibdir, L"my_platlibdir");

putenv("PYTHONVERBOSE=0");

config.verbose = 1;

config.site_import = 0;

config.bytes_warning = 1;

putenv("PYTHONINSPECT=");

config.inspect = 1;

config.interactive = 1;

putenv("PYTHONOPTIMIZE=0");

config.optimization_level = 2;

/* FIXME: test parser_debug */

putenv("PYTHONDONTWRITEBYTECODE=");

config.write_bytecode = 0;

config.quiet = 1;

config.configure_c_stdio = 1;

putenv("PYTHONUNBUFFERED=");

config.buffered_stdio = 0;

putenv("PYTHONIOENCODING=cp424");

config_set_string(&config, &config.stdio_encoding, L"iso8859-1");

config_set_string(&config, &config.stdio_errors, L"replace");

putenv("PYTHONNOUSERSITE=");

config.user_site_directory = 0;

config_set_string(&config, &config.check_hash_pycs_mode, L"always");

config.pathconfig_warnings = 0;

config.safe_path = 1;

#ifdef Py_STATS

putenv("PYTHONSTATS=");

config._pystats = 1;

putenv("PYTHONINTMAXSTRDIGITS=6666");

config.int_max_str_digits = 31337;

config.cpu_count = 4321;

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int check_init_parse_argv(int parse_argv)

{

PyConfig config;

PyConfig_InitPythonConfig(&config);

config.parse_argv = parse_argv;

wchar_t* argv[] = {

L"./argv0",

L"-E",

L"-c",

L"pass",

L"arg1",

L"-v",

L"arg3",

};

config_set_argv(&config, Py_ARRAY_LENGTH(argv), argv);

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int test_init_parse_argv(void)

{

return check_init_parse_argv(1);

}

static int test_init_dont_parse_argv(void)

{

return check_init_parse_argv(0);

}

static void set_most_env_vars(void)

{

putenv("PYTHONHASHSEED=42");

#ifndef Py_GIL_DISABLED

putenv("PYTHONMALLOC=malloc");

putenv("PYTHONMALLOC=mimalloc");

putenv("PYTHONTRACEMALLOC=2");

putenv("PYTHONPROFILEIMPORTTIME=1");

putenv("PYTHONNODEBUGRANGES=1");

putenv("PYTHONMALLOCSTATS=1");

putenv("PYTHON_PYMALLOC_HUGEPAGES=1");

putenv("PYTHONUTF8=1");

putenv("PYTHONVERBOSE=1");

putenv("PYTHONINSPECT=1");

putenv("PYTHONOPTIMIZE=2");

putenv("PYTHONDONTWRITEBYTECODE=1");

putenv("PYTHONUNBUFFERED=1");

putenv("PYTHONPYCACHEPREFIX=env_pycache_prefix");

putenv("PYTHONNOUSERSITE=1");

putenv("PYTHONFAULTHANDLER=1");

putenv("PYTHONIOENCODING=iso8859-1:replace");

putenv("PYTHONPLATLIBDIR=env_platlibdir");

putenv("PYTHONSAFEPATH=1");

putenv("PYTHONINTMAXSTRDIGITS=4567");

#ifdef Py_STATS

putenv("PYTHONSTATS=1");

putenv("PYTHONPERFSUPPORT=1");

}

static void set_all_env_vars(void)

{

set_most_env_vars();

putenv("PYTHONWARNINGS=EnvVar");

putenv("PYTHONPATH=/my/path");

}

static int test_init_compat_env(void)

{

/* Test initialization from environment variables */

set_all_env_vars();

_testembed_initialize();

dump_config();

Py_Finalize();

return 0;

}

static int test_init_python_env(void)

{

set_all_env_vars();

PyConfig config;

PyConfig_InitPythonConfig(&config);

config_set_program_name(&config);

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static void set_all_env_vars_dev_mode(void)

{

putenv("PYTHONMALLOC=");

putenv("PYTHONFAULTHANDLER=");

putenv("PYTHONDEVMODE=1");

}

static int test_init_env_dev_mode(void)

{

/* Test initialization from environment variables */

set_all_env_vars_dev_mode();

_testembed_initialize();

dump_config();

Py_Finalize();

return 0;

}

static int test_init_env_dev_mode_alloc(void)

{

/* Test initialization from environment variables */

set_all_env_vars_dev_mode();

#ifndef Py_GIL_DISABLED

putenv("PYTHONMALLOC=malloc");

putenv("PYTHONMALLOC=mimalloc");

_testembed_initialize();

dump_config();

Py_Finalize();

return 0;

}

static int test_init_isolated_flag(void)

{

/* Test PyConfig.isolated=1 */

PyConfig config;

PyConfig_InitPythonConfig(&config);

config.isolated = 1;

// These options are set to 1 by isolated=1

config.safe_path = 0;

config.use_environment = 1;

config.user_site_directory = 1;

config_set_program_name(&config);

set_all_env_vars();

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int test_preinit_isolated1(void)

{

PyPreConfig preconfig;

_PyPreConfig_InitCompatConfig(&preconfig);

preconfig.isolated = 1;

PyStatus status = Py_PreInitialize(&preconfig);

if (PyStatus_Exception(status)) {

Py_ExitStatusException(status);

}

PyConfig config;

_PyConfig_InitCompatConfig(&config);

config_set_program_name(&config);

set_all_env_vars();

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int test_preinit_isolated2(void)

{

PyPreConfig preconfig;

_PyPreConfig_InitCompatConfig(&preconfig);

preconfig.isolated = 0;

PyStatus status = Py_PreInitialize(&preconfig);

if (PyStatus_Exception(status)) {

Py_ExitStatusException(status);

}

/* Test PyConfig.isolated=1 */

PyConfig config;

_PyConfig_InitCompatConfig(&config);

config.isolated = 1;

config_set_program_name(&config);

set_all_env_vars();

init_from_config_clear(&config);

dump_config();

Py_Finalize();

return 0;

}

static int test_preinit_dont_parse_argv(void)