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 earlier 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()