Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 42 additions & 43 deletions Lib/test/test_capi.py
Original file line number Diff line number Diff line change
Expand Up @@ -385,7 +385,7 @@ def run_embedded_interpreter(self, *args):
(p.returncode, err))
return out, err

def test_subinterps(self):
def run_repeated_init_and_subinterpreters(self):
out, err = self.run_embedded_interpreter("repeated_init_and_subinterpreters")
self.assertEqual(err, "")

Expand All @@ -404,73 +404,72 @@ def test_subinterps(self):
r"id\(modules\) = ([\d]+)$")
Interp = namedtuple("Interp", "id interp tstate modules")

main = None
lastmain = None
numinner = None
numloops = 0
current_run = []
for line in out.splitlines():
if line == "--- Pass {} ---".format(numloops):
if numinner is not None:
self.assertEqual(numinner, 5)
self.assertEqual(len(current_run), 0)
if support.verbose:
print(line)
lastmain = main
main = None
mainid = 0
numloops += 1
numinner = 0
continue
numinner += 1

self.assertLessEqual(numinner, 5)
self.assertLess(len(current_run), 5)
match = re.match(interp_pat, line)
if match is None:
self.assertRegex(line, interp_pat)

# The last line in the loop should be the same as the first.
if numinner == 5:
self.assertEqual(match.groups(), main)
continue

# Parse the line from the loop. The first line is the main
# interpreter and the 3 afterward are subinterpreters.
interp = Interp(*match.groups())
if support.verbose:
print(interp)
if numinner == 1:
main = interp
id = str(mainid)
else:
subid = mainid + numinner - 1
id = str(subid)

# Validate the loop line for each interpreter.
self.assertEqual(interp.id, id)
self.assertTrue(interp.interp)
self.assertTrue(interp.tstate)
self.assertTrue(interp.modules)
if platform.system() == 'Windows':
# XXX Fix on Windows: something is going on with the
# pointers in Programs/_testembed.c. interp.interp
# is 0x0 and # interp.modules is the same between
# interpreters.
continue
if interp is main:
if lastmain is not None:
# A new main interpreter may have the same interp
# and/or tstate pointer as an earlier finalized/
# destroyed one. So we do not check interp or
# tstate here.
self.assertNotEqual(interp.modules, lastmain.modules)
else:
current_run.append(interp)

# The last line in the loop should be the same as the first.
if len(current_run) == 5:
main = current_run[0]
self.assertEqual(interp, main)
yield current_run
current_run = []

def test_subinterps_main(self):
for run in self.run_repeated_init_and_subinterpreters():
main = run[0]

self.assertEqual(main.id, '0')

def test_subinterps_different_ids(self):
for run in self.run_repeated_init_and_subinterpreters():
main, *subs, _ = run

mainid = int(main.id)
for i, sub in enumerate(subs):
self.assertEqual(sub.id, str(mainid + i + 1))

def test_subinterps_distinct_state(self):
for run in self.run_repeated_init_and_subinterpreters():
main, *subs, _ = run

if '0x0' in main:
# XXX Fix on Windows (and other platforms): something
# is going on with the pointers in Programs/_testembed.c.
# interp.interp is 0x0 and interp.modules is the same
# between interpreters.
raise unittest.SkipTest('platform prints pointers as 0x0')

for sub in subs:
# A new subinterpreter may have the same
# PyInterpreterState pointer as a previous one if
# the earlier one has already been destroyed. So
# we compare with the main interpreter. The same
# applies to tstate.
self.assertNotEqual(interp.interp, main.interp)
self.assertNotEqual(interp.tstate, main.tstate)
self.assertNotEqual(interp.modules, main.modules)
self.assertNotEqual(sub.interp, main.interp)
self.assertNotEqual(sub.tstate, main.tstate)
self.assertNotEqual(sub.modules, main.modules)

@staticmethod
def _get_default_pipe_encoding():
Expand Down
2 changes: 1 addition & 1 deletion Programs/_testembed.c
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ static void print_subinterp(void)
PyThreadState *ts = PyThreadState_Get();
PyInterpreterState *interp = ts->interp;
int64_t id = PyInterpreterState_GetID(interp);
printf("interp %lu <0x%" PRIXPTR ">, thread state <0x%" PRIXPTR ">: ",
printf("interp %" PRId64 " <0x%" PRIXPTR ">, thread state <0x%" PRIXPTR ">: ",
id, (uintptr_t)interp, (uintptr_t)ts);
fflush(stdout);
PyRun_SimpleString(
Expand Down