mirror of https://github.com/python/cpython
1152 lines
34 KiB
Python
1152 lines
34 KiB
Python
import contextlib
|
|
import itertools
|
|
import os
|
|
import pickle
|
|
import sys
|
|
from textwrap import dedent
|
|
import threading
|
|
import unittest
|
|
|
|
from test import support
|
|
from test.support import import_helper
|
|
from test.support import os_helper
|
|
from test.support import script_helper
|
|
|
|
|
|
_interpreters = import_helper.import_module('_interpreters')
|
|
_testinternalcapi = import_helper.import_module('_testinternalcapi')
|
|
from _interpreters import InterpreterNotFoundError
|
|
|
|
|
|
##################################
|
|
# helpers
|
|
|
|
def _captured_script(script):
|
|
r, w = os.pipe()
|
|
indented = script.replace('\n', '\n ')
|
|
wrapped = dedent(f"""
|
|
import contextlib
|
|
with open({w}, 'w', encoding="utf-8") as spipe:
|
|
with contextlib.redirect_stdout(spipe):
|
|
{indented}
|
|
""")
|
|
return wrapped, open(r, encoding="utf-8")
|
|
|
|
|
|
def _run_output(interp, request):
|
|
script, rpipe = _captured_script(request)
|
|
with rpipe:
|
|
_interpreters.run_string(interp, script)
|
|
return rpipe.read()
|
|
|
|
|
|
def _wait_for_interp_to_run(interp, timeout=None):
|
|
# bpo-37224: Running this test file in multiprocesses will fail randomly.
|
|
# The failure reason is that the thread can't acquire the cpu to
|
|
# run subinterpreter eariler than the main thread in multiprocess.
|
|
if timeout is None:
|
|
timeout = support.SHORT_TIMEOUT
|
|
for _ in support.sleeping_retry(timeout, error=False):
|
|
if _interpreters.is_running(interp):
|
|
break
|
|
else:
|
|
raise RuntimeError('interp is not running')
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _running(interp):
|
|
r, w = os.pipe()
|
|
def run():
|
|
_interpreters.run_string(interp, dedent(f"""
|
|
# wait for "signal"
|
|
with open({r}, encoding="utf-8") as rpipe:
|
|
rpipe.read()
|
|
"""))
|
|
|
|
t = threading.Thread(target=run)
|
|
t.start()
|
|
_wait_for_interp_to_run(interp)
|
|
|
|
yield
|
|
|
|
with open(w, 'w', encoding="utf-8") as spipe:
|
|
spipe.write('done')
|
|
t.join()
|
|
|
|
|
|
def clean_up_interpreters():
|
|
for id, *_ in _interpreters.list_all():
|
|
if id == 0: # main
|
|
continue
|
|
try:
|
|
_interpreters.destroy(id)
|
|
except _interpreters.InterpreterError:
|
|
pass # already destroyed
|
|
|
|
|
|
class TestBase(unittest.TestCase):
|
|
|
|
def tearDown(self):
|
|
clean_up_interpreters()
|
|
|
|
|
|
##################################
|
|
# misc. tests
|
|
|
|
class IsShareableTests(unittest.TestCase):
|
|
|
|
def test_default_shareables(self):
|
|
shareables = [
|
|
# singletons
|
|
None,
|
|
# builtin objects
|
|
b'spam',
|
|
'spam',
|
|
10,
|
|
-10,
|
|
True,
|
|
False,
|
|
100.0,
|
|
(1, ('spam', 'eggs')),
|
|
]
|
|
for obj in shareables:
|
|
with self.subTest(obj):
|
|
self.assertTrue(
|
|
_interpreters.is_shareable(obj))
|
|
|
|
def test_not_shareable(self):
|
|
class Cheese:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class SubBytes(bytes):
|
|
"""A subclass of a shareable type."""
|
|
|
|
not_shareables = [
|
|
# singletons
|
|
NotImplemented,
|
|
...,
|
|
# builtin types and objects
|
|
type,
|
|
object,
|
|
object(),
|
|
Exception(),
|
|
# user-defined types and objects
|
|
Cheese,
|
|
Cheese('Wensleydale'),
|
|
SubBytes(b'spam'),
|
|
]
|
|
for obj in not_shareables:
|
|
with self.subTest(repr(obj)):
|
|
self.assertFalse(
|
|
_interpreters.is_shareable(obj))
|
|
|
|
|
|
class ShareableTypeTests(unittest.TestCase):
|
|
|
|
def _assert_values(self, values):
|
|
for obj in values:
|
|
with self.subTest(obj):
|
|
xid = _testinternalcapi.get_crossinterp_data(obj)
|
|
got = _testinternalcapi.restore_crossinterp_data(xid)
|
|
|
|
self.assertEqual(got, obj)
|
|
self.assertIs(type(got), type(obj))
|
|
|
|
def test_singletons(self):
|
|
for obj in [None]:
|
|
with self.subTest(obj):
|
|
xid = _testinternalcapi.get_crossinterp_data(obj)
|
|
got = _testinternalcapi.restore_crossinterp_data(xid)
|
|
|
|
# XXX What about between interpreters?
|
|
self.assertIs(got, obj)
|
|
|
|
def test_types(self):
|
|
self._assert_values([
|
|
b'spam',
|
|
9999,
|
|
])
|
|
|
|
def test_bytes(self):
|
|
self._assert_values(i.to_bytes(2, 'little', signed=True)
|
|
for i in range(-1, 258))
|
|
|
|
def test_strs(self):
|
|
self._assert_values(['hello world', '你好世界', ''])
|
|
|
|
def test_int(self):
|
|
self._assert_values(itertools.chain(range(-1, 258),
|
|
[sys.maxsize, -sys.maxsize - 1]))
|
|
|
|
def test_non_shareable_int(self):
|
|
ints = [
|
|
sys.maxsize + 1,
|
|
-sys.maxsize - 2,
|
|
2**1000,
|
|
]
|
|
for i in ints:
|
|
with self.subTest(i):
|
|
with self.assertRaises(OverflowError):
|
|
_testinternalcapi.get_crossinterp_data(i)
|
|
|
|
def test_bool(self):
|
|
self._assert_values([True, False])
|
|
|
|
def test_float(self):
|
|
self._assert_values([0.0, 1.1, -1.0, 0.12345678, -0.12345678])
|
|
|
|
def test_tuple(self):
|
|
self._assert_values([(), (1,), ("hello", "world", ), (1, True, "hello")])
|
|
# Test nesting
|
|
self._assert_values([
|
|
((1,),),
|
|
((1, 2), (3, 4)),
|
|
((1, 2), (3, 4), (5, 6)),
|
|
])
|
|
|
|
def test_tuples_containing_non_shareable_types(self):
|
|
non_shareables = [
|
|
Exception(),
|
|
object(),
|
|
]
|
|
for s in non_shareables:
|
|
value = tuple([0, 1.0, s])
|
|
with self.subTest(repr(value)):
|
|
# XXX Assert the NotShareableError when it is exported
|
|
with self.assertRaises(ValueError):
|
|
_testinternalcapi.get_crossinterp_data(value)
|
|
# Check nested as well
|
|
value = tuple([0, 1., (s,)])
|
|
with self.subTest("nested " + repr(value)):
|
|
# XXX Assert the NotShareableError when it is exported
|
|
with self.assertRaises(ValueError):
|
|
_testinternalcapi.get_crossinterp_data(value)
|
|
|
|
|
|
class ModuleTests(TestBase):
|
|
|
|
def test_import_in_interpreter(self):
|
|
_run_output(
|
|
_interpreters.create(),
|
|
'import _interpreters',
|
|
)
|
|
|
|
|
|
##################################
|
|
# interpreter tests
|
|
|
|
class ListAllTests(TestBase):
|
|
|
|
def test_initial(self):
|
|
main, *_ = _interpreters.get_main()
|
|
ids = [id for id, *_ in _interpreters.list_all()]
|
|
self.assertEqual(ids, [main])
|
|
|
|
def test_after_creating(self):
|
|
main, *_ = _interpreters.get_main()
|
|
first = _interpreters.create()
|
|
second = _interpreters.create()
|
|
ids = [id for id, *_ in _interpreters.list_all()]
|
|
self.assertEqual(ids, [main, first, second])
|
|
|
|
def test_after_destroying(self):
|
|
main, *_ = _interpreters.get_main()
|
|
first = _interpreters.create()
|
|
second = _interpreters.create()
|
|
_interpreters.destroy(first)
|
|
ids = [id for id, *_ in _interpreters.list_all()]
|
|
self.assertEqual(ids, [main, second])
|
|
|
|
|
|
class GetCurrentTests(TestBase):
|
|
|
|
def test_main(self):
|
|
main, *_ = _interpreters.get_main()
|
|
cur, *_ = _interpreters.get_current()
|
|
self.assertEqual(cur, main)
|
|
self.assertIsInstance(cur, int)
|
|
|
|
def test_subinterpreter(self):
|
|
main, *_ = _interpreters.get_main()
|
|
interp = _interpreters.create()
|
|
out = _run_output(interp, dedent("""
|
|
import _interpreters
|
|
cur, *_ = _interpreters.get_current()
|
|
print(cur)
|
|
assert isinstance(cur, int)
|
|
"""))
|
|
cur = int(out.strip())
|
|
_, expected = [id for id, *_ in _interpreters.list_all()]
|
|
self.assertEqual(cur, expected)
|
|
self.assertNotEqual(cur, main)
|
|
|
|
|
|
class GetMainTests(TestBase):
|
|
|
|
def test_from_main(self):
|
|
[expected] = [id for id, *_ in _interpreters.list_all()]
|
|
main, *_ = _interpreters.get_main()
|
|
self.assertEqual(main, expected)
|
|
self.assertIsInstance(main, int)
|
|
|
|
def test_from_subinterpreter(self):
|
|
[expected] = [id for id, *_ in _interpreters.list_all()]
|
|
interp = _interpreters.create()
|
|
out = _run_output(interp, dedent("""
|
|
import _interpreters
|
|
main, *_ = _interpreters.get_main()
|
|
print(main)
|
|
assert isinstance(main, int)
|
|
"""))
|
|
main = int(out.strip())
|
|
self.assertEqual(main, expected)
|
|
|
|
|
|
class IsRunningTests(TestBase):
|
|
|
|
def test_main(self):
|
|
main, *_ = _interpreters.get_main()
|
|
self.assertTrue(_interpreters.is_running(main))
|
|
|
|
@unittest.skip('Fails on FreeBSD')
|
|
def test_subinterpreter(self):
|
|
interp = _interpreters.create()
|
|
self.assertFalse(_interpreters.is_running(interp))
|
|
|
|
with _running(interp):
|
|
self.assertTrue(_interpreters.is_running(interp))
|
|
self.assertFalse(_interpreters.is_running(interp))
|
|
|
|
def test_from_subinterpreter(self):
|
|
interp = _interpreters.create()
|
|
out = _run_output(interp, dedent(f"""
|
|
import _interpreters
|
|
if _interpreters.is_running({interp}):
|
|
print(True)
|
|
else:
|
|
print(False)
|
|
"""))
|
|
self.assertEqual(out.strip(), 'True')
|
|
|
|
def test_already_destroyed(self):
|
|
interp = _interpreters.create()
|
|
_interpreters.destroy(interp)
|
|
with self.assertRaises(InterpreterNotFoundError):
|
|
_interpreters.is_running(interp)
|
|
|
|
def test_does_not_exist(self):
|
|
with self.assertRaises(InterpreterNotFoundError):
|
|
_interpreters.is_running(1_000_000)
|
|
|
|
def test_bad_id(self):
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.is_running(-1)
|
|
|
|
|
|
class CreateTests(TestBase):
|
|
|
|
def test_in_main(self):
|
|
id = _interpreters.create()
|
|
self.assertIsInstance(id, int)
|
|
|
|
after = [id for id, *_ in _interpreters.list_all()]
|
|
self.assertIn(id, after)
|
|
|
|
@unittest.skip('enable this test when working on pystate.c')
|
|
def test_unique_id(self):
|
|
seen = set()
|
|
for _ in range(100):
|
|
id = _interpreters.create()
|
|
_interpreters.destroy(id)
|
|
seen.add(id)
|
|
|
|
self.assertEqual(len(seen), 100)
|
|
|
|
def test_in_thread(self):
|
|
lock = threading.Lock()
|
|
id = None
|
|
def f():
|
|
nonlocal id
|
|
id = _interpreters.create()
|
|
lock.acquire()
|
|
lock.release()
|
|
|
|
t = threading.Thread(target=f)
|
|
with lock:
|
|
t.start()
|
|
t.join()
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertIn(id, after)
|
|
|
|
def test_in_subinterpreter(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
id1 = _interpreters.create()
|
|
out = _run_output(id1, dedent("""
|
|
import _interpreters
|
|
id = _interpreters.create()
|
|
print(id)
|
|
assert isinstance(id, int)
|
|
"""))
|
|
id2 = int(out.strip())
|
|
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, {main, id1, id2})
|
|
|
|
def test_in_threaded_subinterpreter(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
id1 = _interpreters.create()
|
|
id2 = None
|
|
def f():
|
|
nonlocal id2
|
|
out = _run_output(id1, dedent("""
|
|
import _interpreters
|
|
id = _interpreters.create()
|
|
print(id)
|
|
"""))
|
|
id2 = int(out.strip())
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, {main, id1, id2})
|
|
|
|
def test_after_destroy_all(self):
|
|
before = set(id for id, *_ in _interpreters.list_all())
|
|
# Create 3 subinterpreters.
|
|
ids = []
|
|
for _ in range(3):
|
|
id = _interpreters.create()
|
|
ids.append(id)
|
|
# Now destroy them.
|
|
for id in ids:
|
|
_interpreters.destroy(id)
|
|
# Finally, create another.
|
|
id = _interpreters.create()
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, before | {id})
|
|
|
|
def test_after_destroy_some(self):
|
|
before = set(id for id, *_ in _interpreters.list_all())
|
|
# Create 3 subinterpreters.
|
|
id1 = _interpreters.create()
|
|
id2 = _interpreters.create()
|
|
id3 = _interpreters.create()
|
|
# Now destroy 2 of them.
|
|
_interpreters.destroy(id1)
|
|
_interpreters.destroy(id3)
|
|
# Finally, create another.
|
|
id = _interpreters.create()
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, before | {id, id2})
|
|
|
|
|
|
class DestroyTests(TestBase):
|
|
|
|
def test_one(self):
|
|
id1 = _interpreters.create()
|
|
id2 = _interpreters.create()
|
|
id3 = _interpreters.create()
|
|
before = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertIn(id2, before)
|
|
|
|
_interpreters.destroy(id2)
|
|
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertNotIn(id2, after)
|
|
self.assertIn(id1, after)
|
|
self.assertIn(id3, after)
|
|
|
|
def test_all(self):
|
|
initial = set(id for id, *_ in _interpreters.list_all())
|
|
ids = set()
|
|
for _ in range(3):
|
|
id = _interpreters.create()
|
|
ids.add(id)
|
|
before = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(before, initial | ids)
|
|
for id in ids:
|
|
_interpreters.destroy(id)
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, initial)
|
|
|
|
def test_main(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
with self.assertRaises(_interpreters.InterpreterError):
|
|
_interpreters.destroy(main)
|
|
|
|
def f():
|
|
with self.assertRaises(_interpreters.InterpreterError):
|
|
_interpreters.destroy(main)
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
|
|
def test_already_destroyed(self):
|
|
id = _interpreters.create()
|
|
_interpreters.destroy(id)
|
|
with self.assertRaises(InterpreterNotFoundError):
|
|
_interpreters.destroy(id)
|
|
|
|
def test_does_not_exist(self):
|
|
with self.assertRaises(InterpreterNotFoundError):
|
|
_interpreters.destroy(1_000_000)
|
|
|
|
def test_bad_id(self):
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.destroy(-1)
|
|
|
|
def test_from_current(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
id = _interpreters.create()
|
|
script = dedent(f"""
|
|
import _interpreters
|
|
try:
|
|
_interpreters.destroy({id})
|
|
except _interpreters.InterpreterError:
|
|
pass
|
|
""")
|
|
|
|
_interpreters.run_string(id, script)
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, {main, id})
|
|
|
|
def test_from_sibling(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
id1 = _interpreters.create()
|
|
id2 = _interpreters.create()
|
|
script = dedent(f"""
|
|
import _interpreters
|
|
_interpreters.destroy({id2})
|
|
""")
|
|
_interpreters.run_string(id1, script)
|
|
|
|
after = set(id for id, *_ in _interpreters.list_all())
|
|
self.assertEqual(after, {main, id1})
|
|
|
|
def test_from_other_thread(self):
|
|
id = _interpreters.create()
|
|
def f():
|
|
_interpreters.destroy(id)
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
|
|
def test_still_running(self):
|
|
main, = [id for id, *_ in _interpreters.list_all()]
|
|
interp = _interpreters.create()
|
|
with _running(interp):
|
|
self.assertTrue(_interpreters.is_running(interp),
|
|
msg=f"Interp {interp} should be running before destruction.")
|
|
|
|
with self.assertRaises(_interpreters.InterpreterError,
|
|
msg=f"Should not be able to destroy interp {interp} while it's still running."):
|
|
_interpreters.destroy(interp)
|
|
self.assertTrue(_interpreters.is_running(interp))
|
|
|
|
|
|
class RunStringTests(TestBase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.id = _interpreters.create()
|
|
|
|
def test_success(self):
|
|
script, file = _captured_script('print("it worked!", end="")')
|
|
with file:
|
|
_interpreters.run_string(self.id, script)
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_in_thread(self):
|
|
script, file = _captured_script('print("it worked!", end="")')
|
|
with file:
|
|
def f():
|
|
_interpreters.run_string(self.id, script)
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_create_thread(self):
|
|
subinterp = _interpreters.create()
|
|
script, file = _captured_script("""
|
|
import threading
|
|
def f():
|
|
print('it worked!', end='')
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
""")
|
|
with file:
|
|
_interpreters.run_string(subinterp, script)
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_create_daemon_thread(self):
|
|
with self.subTest('isolated'):
|
|
expected = 'spam spam spam spam spam'
|
|
subinterp = _interpreters.create('isolated')
|
|
script, file = _captured_script(f"""
|
|
import threading
|
|
def f():
|
|
print('it worked!', end='')
|
|
|
|
try:
|
|
t = threading.Thread(target=f, daemon=True)
|
|
t.start()
|
|
t.join()
|
|
except RuntimeError:
|
|
print('{expected}', end='')
|
|
""")
|
|
with file:
|
|
_interpreters.run_string(subinterp, script)
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, expected)
|
|
|
|
with self.subTest('not isolated'):
|
|
subinterp = _interpreters.create('legacy')
|
|
script, file = _captured_script("""
|
|
import threading
|
|
def f():
|
|
print('it worked!', end='')
|
|
|
|
t = threading.Thread(target=f, daemon=True)
|
|
t.start()
|
|
t.join()
|
|
""")
|
|
with file:
|
|
_interpreters.run_string(subinterp, script)
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_shareable_types(self):
|
|
interp = _interpreters.create()
|
|
objects = [
|
|
None,
|
|
'spam',
|
|
b'spam',
|
|
42,
|
|
]
|
|
for obj in objects:
|
|
with self.subTest(obj):
|
|
_interpreters.set___main___attrs(interp, dict(obj=obj))
|
|
_interpreters.run_string(
|
|
interp,
|
|
f'assert(obj == {obj!r})',
|
|
)
|
|
|
|
def test_os_exec(self):
|
|
expected = 'spam spam spam spam spam'
|
|
subinterp = _interpreters.create()
|
|
script, file = _captured_script(f"""
|
|
import os, sys
|
|
try:
|
|
os.execl(sys.executable)
|
|
except RuntimeError:
|
|
print('{expected}', end='')
|
|
""")
|
|
with file:
|
|
_interpreters.run_string(subinterp, script)
|
|
out = file.read()
|
|
|
|
self.assertEqual(out, expected)
|
|
|
|
@support.requires_fork()
|
|
def test_fork(self):
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile('w+', encoding="utf-8") as file:
|
|
file.write('')
|
|
file.flush()
|
|
|
|
expected = 'spam spam spam spam spam'
|
|
script = dedent(f"""
|
|
import os
|
|
try:
|
|
os.fork()
|
|
except RuntimeError:
|
|
with open('{file.name}', 'w', encoding='utf-8') as out:
|
|
out.write('{expected}')
|
|
""")
|
|
_interpreters.run_string(self.id, script)
|
|
|
|
file.seek(0)
|
|
content = file.read()
|
|
self.assertEqual(content, expected)
|
|
|
|
def test_already_running(self):
|
|
with _running(self.id):
|
|
with self.assertRaises(_interpreters.InterpreterError):
|
|
_interpreters.run_string(self.id, 'print("spam")')
|
|
|
|
def test_does_not_exist(self):
|
|
id = 0
|
|
while id in set(id for id, *_ in _interpreters.list_all()):
|
|
id += 1
|
|
with self.assertRaises(InterpreterNotFoundError):
|
|
_interpreters.run_string(id, 'print("spam")')
|
|
|
|
def test_error_id(self):
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_string(-1, 'print("spam")')
|
|
|
|
def test_bad_id(self):
|
|
with self.assertRaises(TypeError):
|
|
_interpreters.run_string('spam', 'print("spam")')
|
|
|
|
def test_bad_script(self):
|
|
with self.assertRaises(TypeError):
|
|
_interpreters.run_string(self.id, 10)
|
|
|
|
def test_bytes_for_script(self):
|
|
with self.assertRaises(TypeError):
|
|
_interpreters.run_string(self.id, b'print("spam")')
|
|
|
|
def test_with_shared(self):
|
|
r, w = os.pipe()
|
|
|
|
shared = {
|
|
'spam': b'ham',
|
|
'eggs': b'-1',
|
|
'cheddar': None,
|
|
}
|
|
script = dedent(f"""
|
|
eggs = int(eggs)
|
|
spam = 42
|
|
result = spam + eggs
|
|
|
|
ns = dict(vars())
|
|
del ns['__builtins__']
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
""")
|
|
_interpreters.set___main___attrs(self.id, shared)
|
|
_interpreters.run_string(self.id, script)
|
|
with open(r, 'rb') as chan:
|
|
ns = pickle.load(chan)
|
|
|
|
self.assertEqual(ns['spam'], 42)
|
|
self.assertEqual(ns['eggs'], -1)
|
|
self.assertEqual(ns['result'], 41)
|
|
self.assertIsNone(ns['cheddar'])
|
|
|
|
def test_shared_overwrites(self):
|
|
_interpreters.run_string(self.id, dedent("""
|
|
spam = 'eggs'
|
|
ns1 = dict(vars())
|
|
del ns1['__builtins__']
|
|
"""))
|
|
|
|
shared = {'spam': b'ham'}
|
|
script = dedent("""
|
|
ns2 = dict(vars())
|
|
del ns2['__builtins__']
|
|
""")
|
|
_interpreters.set___main___attrs(self.id, shared)
|
|
_interpreters.run_string(self.id, script)
|
|
|
|
r, w = os.pipe()
|
|
script = dedent(f"""
|
|
ns = dict(vars())
|
|
del ns['__builtins__']
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
""")
|
|
_interpreters.run_string(self.id, script)
|
|
with open(r, 'rb') as chan:
|
|
ns = pickle.load(chan)
|
|
|
|
self.assertEqual(ns['ns1']['spam'], 'eggs')
|
|
self.assertEqual(ns['ns2']['spam'], b'ham')
|
|
self.assertEqual(ns['spam'], b'ham')
|
|
|
|
def test_shared_overwrites_default_vars(self):
|
|
r, w = os.pipe()
|
|
|
|
shared = {'__name__': b'not __main__'}
|
|
script = dedent(f"""
|
|
spam = 42
|
|
|
|
ns = dict(vars())
|
|
del ns['__builtins__']
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
""")
|
|
_interpreters.set___main___attrs(self.id, shared)
|
|
_interpreters.run_string(self.id, script)
|
|
with open(r, 'rb') as chan:
|
|
ns = pickle.load(chan)
|
|
|
|
self.assertEqual(ns['__name__'], b'not __main__')
|
|
|
|
def test_main_reused(self):
|
|
r, w = os.pipe()
|
|
_interpreters.run_string(self.id, dedent(f"""
|
|
spam = True
|
|
|
|
ns = dict(vars())
|
|
del ns['__builtins__']
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
del ns, pickle, chan
|
|
"""))
|
|
with open(r, 'rb') as chan:
|
|
ns1 = pickle.load(chan)
|
|
|
|
r, w = os.pipe()
|
|
_interpreters.run_string(self.id, dedent(f"""
|
|
eggs = False
|
|
|
|
ns = dict(vars())
|
|
del ns['__builtins__']
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
"""))
|
|
with open(r, 'rb') as chan:
|
|
ns2 = pickle.load(chan)
|
|
|
|
self.assertIn('spam', ns1)
|
|
self.assertNotIn('eggs', ns1)
|
|
self.assertIn('eggs', ns2)
|
|
self.assertIn('spam', ns2)
|
|
|
|
def test_execution_namespace_is_main(self):
|
|
r, w = os.pipe()
|
|
|
|
script = dedent(f"""
|
|
spam = 42
|
|
|
|
ns = dict(vars())
|
|
ns['__builtins__'] = str(ns['__builtins__'])
|
|
import pickle
|
|
with open({w}, 'wb') as chan:
|
|
pickle.dump(ns, chan)
|
|
""")
|
|
_interpreters.run_string(self.id, script)
|
|
with open(r, 'rb') as chan:
|
|
ns = pickle.load(chan)
|
|
|
|
ns.pop('__builtins__')
|
|
ns.pop('__loader__')
|
|
self.assertEqual(ns, {
|
|
'__name__': '__main__',
|
|
'__annotations__': {},
|
|
'__doc__': None,
|
|
'__package__': None,
|
|
'__spec__': None,
|
|
'spam': 42,
|
|
})
|
|
|
|
# XXX Fix this test!
|
|
@unittest.skip('blocking forever')
|
|
def test_still_running_at_exit(self):
|
|
script = dedent("""
|
|
from textwrap import dedent
|
|
import threading
|
|
import _interpreters
|
|
id = _interpreters.create()
|
|
def f():
|
|
_interpreters.run_string(id, dedent('''
|
|
import time
|
|
# Give plenty of time for the main interpreter to finish.
|
|
time.sleep(1_000_000)
|
|
'''))
|
|
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
""")
|
|
with support.temp_dir() as dirname:
|
|
filename = script_helper.make_script(dirname, 'interp', script)
|
|
with script_helper.spawn_python(filename) as proc:
|
|
retcode = proc.wait()
|
|
|
|
self.assertEqual(retcode, 0)
|
|
|
|
|
|
class RunFailedTests(TestBase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.id = _interpreters.create()
|
|
|
|
def add_module(self, modname, text):
|
|
import tempfile
|
|
tempdir = tempfile.mkdtemp()
|
|
self.addCleanup(lambda: os_helper.rmtree(tempdir))
|
|
_interpreters.run_string(self.id, dedent(f"""
|
|
import sys
|
|
sys.path.insert(0, {tempdir!r})
|
|
"""))
|
|
return script_helper.make_script(tempdir, modname, text)
|
|
|
|
def run_script(self, text, *, fails=False):
|
|
r, w = os.pipe()
|
|
try:
|
|
script = dedent(f"""
|
|
import os, sys
|
|
os.write({w}, b'0')
|
|
|
|
# This raises an exception:
|
|
{{}}
|
|
|
|
# Nothing from here down should ever run.
|
|
os.write({w}, b'1')
|
|
class NeverError(Exception): pass
|
|
raise NeverError # never raised
|
|
""").format(dedent(text))
|
|
if fails:
|
|
err = _interpreters.run_string(self.id, script)
|
|
self.assertIsNot(err, None)
|
|
return err
|
|
else:
|
|
err = _interpreters.run_string(self.id, script)
|
|
self.assertIs(err, None)
|
|
return None
|
|
except:
|
|
raise # re-raise
|
|
else:
|
|
msg = os.read(r, 100)
|
|
self.assertEqual(msg, b'0')
|
|
finally:
|
|
os.close(r)
|
|
os.close(w)
|
|
|
|
def _assert_run_failed(self, exctype, msg, script):
|
|
if isinstance(exctype, str):
|
|
exctype_name = exctype
|
|
exctype = None
|
|
else:
|
|
exctype_name = exctype.__name__
|
|
|
|
# Run the script.
|
|
excinfo = self.run_script(script, fails=True)
|
|
|
|
# Check the wrapper exception.
|
|
self.assertEqual(excinfo.type.__name__, exctype_name)
|
|
if msg is None:
|
|
self.assertEqual(excinfo.formatted.split(':')[0],
|
|
exctype_name)
|
|
else:
|
|
self.assertEqual(excinfo.formatted,
|
|
'{}: {}'.format(exctype_name, msg))
|
|
|
|
return excinfo
|
|
|
|
def assert_run_failed(self, exctype, script):
|
|
self._assert_run_failed(exctype, None, script)
|
|
|
|
def assert_run_failed_msg(self, exctype, msg, script):
|
|
self._assert_run_failed(exctype, msg, script)
|
|
|
|
def test_exit(self):
|
|
with self.subTest('sys.exit(0)'):
|
|
# XXX Should an unhandled SystemExit(0) be handled as not-an-error?
|
|
self.assert_run_failed(SystemExit, """
|
|
sys.exit(0)
|
|
""")
|
|
|
|
with self.subTest('sys.exit()'):
|
|
self.assert_run_failed(SystemExit, """
|
|
import sys
|
|
sys.exit()
|
|
""")
|
|
|
|
with self.subTest('sys.exit(42)'):
|
|
self.assert_run_failed_msg(SystemExit, '42', """
|
|
import sys
|
|
sys.exit(42)
|
|
""")
|
|
|
|
with self.subTest('SystemExit'):
|
|
self.assert_run_failed_msg(SystemExit, '42', """
|
|
raise SystemExit(42)
|
|
""")
|
|
|
|
# XXX Also check os._exit() (via a subprocess)?
|
|
|
|
def test_plain_exception(self):
|
|
self.assert_run_failed_msg(Exception, 'spam', """
|
|
raise Exception("spam")
|
|
""")
|
|
|
|
def test_invalid_syntax(self):
|
|
script = dedent("""
|
|
x = 1 + 2
|
|
y = 2 + 4
|
|
z = 4 + 8
|
|
|
|
# missing close paren
|
|
print("spam"
|
|
|
|
if x + y + z < 20:
|
|
...
|
|
""")
|
|
|
|
with self.subTest('script'):
|
|
self.assert_run_failed(SyntaxError, script)
|
|
|
|
with self.subTest('module'):
|
|
modname = 'spam_spam_spam'
|
|
filename = self.add_module(modname, script)
|
|
self.assert_run_failed(SyntaxError, f"""
|
|
import {modname}
|
|
""")
|
|
|
|
def test_NameError(self):
|
|
self.assert_run_failed(NameError, """
|
|
res = spam + eggs
|
|
""")
|
|
# XXX check preserved suggestions
|
|
|
|
def test_AttributeError(self):
|
|
self.assert_run_failed(AttributeError, """
|
|
object().spam
|
|
""")
|
|
# XXX check preserved suggestions
|
|
|
|
def test_ExceptionGroup(self):
|
|
self.assert_run_failed(ExceptionGroup, """
|
|
raise ExceptionGroup('exceptions', [
|
|
Exception('spam'),
|
|
ImportError('eggs'),
|
|
])
|
|
""")
|
|
|
|
def test_user_defined_exception(self):
|
|
self.assert_run_failed_msg('MyError', 'spam', """
|
|
class MyError(Exception):
|
|
pass
|
|
raise MyError('spam')
|
|
""")
|
|
|
|
|
|
class RunFuncTests(TestBase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.id = _interpreters.create()
|
|
|
|
def test_success(self):
|
|
r, w = os.pipe()
|
|
def script():
|
|
global w
|
|
import contextlib
|
|
with open(w, 'w', encoding="utf-8") as spipe:
|
|
with contextlib.redirect_stdout(spipe):
|
|
print('it worked!', end='')
|
|
_interpreters.set___main___attrs(self.id, dict(w=w))
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
with open(r, encoding="utf-8") as outfile:
|
|
out = outfile.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_in_thread(self):
|
|
r, w = os.pipe()
|
|
def script():
|
|
global w
|
|
import contextlib
|
|
with open(w, 'w', encoding="utf-8") as spipe:
|
|
with contextlib.redirect_stdout(spipe):
|
|
print('it worked!', end='')
|
|
def f():
|
|
_interpreters.set___main___attrs(self.id, dict(w=w))
|
|
_interpreters.run_func(self.id, script)
|
|
t = threading.Thread(target=f)
|
|
t.start()
|
|
t.join()
|
|
|
|
with open(r, encoding="utf-8") as outfile:
|
|
out = outfile.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_code_object(self):
|
|
r, w = os.pipe()
|
|
|
|
def script():
|
|
global w
|
|
import contextlib
|
|
with open(w, 'w', encoding="utf-8") as spipe:
|
|
with contextlib.redirect_stdout(spipe):
|
|
print('it worked!', end='')
|
|
code = script.__code__
|
|
_interpreters.set___main___attrs(self.id, dict(w=w))
|
|
_interpreters.run_func(self.id, code)
|
|
|
|
with open(r, encoding="utf-8") as outfile:
|
|
out = outfile.read()
|
|
|
|
self.assertEqual(out, 'it worked!')
|
|
|
|
def test_closure(self):
|
|
spam = True
|
|
def script():
|
|
assert spam
|
|
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
# XXX This hasn't been fixed yet.
|
|
@unittest.expectedFailure
|
|
def test_return_value(self):
|
|
def script():
|
|
return 'spam'
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
def test_args(self):
|
|
with self.subTest('args'):
|
|
def script(a, b=0):
|
|
assert a == b
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
with self.subTest('*args'):
|
|
def script(*args):
|
|
assert not args
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
with self.subTest('**kwargs'):
|
|
def script(**kwargs):
|
|
assert not kwargs
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
with self.subTest('kwonly'):
|
|
def script(*, spam=True):
|
|
assert spam
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
with self.subTest('posonly'):
|
|
def script(spam, /):
|
|
assert spam
|
|
with self.assertRaises(ValueError):
|
|
_interpreters.run_func(self.id, script)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|