diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/support/interpreters/__init__.py | 79 | ||||
| -rw-r--r-- | Lib/test/test_interpreters/test_api.py | 205 | ||||
| -rw-r--r-- | Lib/test/test_interpreters/utils.py | 15 |
3 files changed, 203 insertions, 96 deletions
diff --git a/Lib/test/support/interpreters/__init__.py b/Lib/test/support/interpreters/__init__.py index 60323c9..0a5a925 100644 --- a/Lib/test/support/interpreters/__init__.py +++ b/Lib/test/support/interpreters/__init__.py @@ -74,51 +74,77 @@ class ExecutionFailed(InterpreterError): def create(): """Return a new (idle) Python interpreter.""" id = _interpreters.create(reqrefs=True) - return Interpreter(id) + return Interpreter(id, _ownsref=True) def list_all(): """Return all existing interpreters.""" - return [Interpreter(id) - for id, in _interpreters.list_all()] + return [Interpreter(id, _whence=whence) + for id, whence in _interpreters.list_all(require_ready=True)] def get_current(): """Return the currently running interpreter.""" - id, = _interpreters.get_current() - return Interpreter(id) + id, whence = _interpreters.get_current() + return Interpreter(id, _whence=whence) def get_main(): """Return the main interpreter.""" - id, = _interpreters.get_main() - return Interpreter(id) + id, whence = _interpreters.get_main() + assert whence == _interpreters.WHENCE_RUNTIME, repr(whence) + return Interpreter(id, _whence=whence) _known = weakref.WeakValueDictionary() class Interpreter: - """A single Python interpreter.""" + """A single Python interpreter. - def __new__(cls, id, /): + Attributes: + + "id" - the unique process-global ID number for the interpreter + "whence" - indicates where the interpreter was created + + If the interpreter wasn't created by this module + then any method that modifies the interpreter will fail, + i.e. .close(), .prepare_main(), .exec(), and .call() + """ + + _WHENCE_TO_STR = { + _interpreters.WHENCE_UNKNOWN: 'unknown', + _interpreters.WHENCE_RUNTIME: 'runtime init', + _interpreters.WHENCE_LEGACY_CAPI: 'legacy C-API', + _interpreters.WHENCE_CAPI: 'C-API', + _interpreters.WHENCE_XI: 'cross-interpreter C-API', + _interpreters.WHENCE_STDLIB: '_interpreters module', + } + + def __new__(cls, id, /, _whence=None, _ownsref=None): # There is only one instance for any given ID. if not isinstance(id, int): raise TypeError(f'id must be an int, got {id!r}') id = int(id) + if _whence is None: + if _ownsref: + _whence = _interpreters.WHENCE_STDLIB + else: + _whence = _interpreters.whence(id) + assert _whence in cls._WHENCE_TO_STR, repr(_whence) + if _ownsref is None: + _ownsref = (_whence == _interpreters.WHENCE_STDLIB) try: self = _known[id] assert hasattr(self, '_ownsref') except KeyError: - # This may raise InterpreterNotFoundError: - _interpreters.incref(id) - try: - self = super().__new__(cls) - self._id = id - self._ownsref = True - except BaseException: - _interpreters.decref(id) - raise + self = super().__new__(cls) _known[id] = self + self._id = id + self._whence = _whence + self._ownsref = _ownsref + if _ownsref: + # This may raise InterpreterNotFoundError: + _interpreters.incref(id) return self def __repr__(self): @@ -143,7 +169,7 @@ class Interpreter: return self._ownsref = False try: - _interpreters.decref(self.id) + _interpreters.decref(self._id) except InterpreterNotFoundError: pass @@ -151,17 +177,24 @@ class Interpreter: def id(self): return self._id + @property + def whence(self): + return self._WHENCE_TO_STR[self._whence] + def is_running(self): """Return whether or not the identified interpreter is running.""" return _interpreters.is_running(self._id) + # Everything past here is available only to interpreters created by + # interpreters.create(). + def close(self): """Finalize and destroy the interpreter. Attempting to destroy the current interpreter results in an InterpreterError. """ - return _interpreters.destroy(self._id) + return _interpreters.destroy(self._id, restrict=True) def prepare_main(self, ns=None, /, **kwargs): """Bind the given values into the interpreter's __main__. @@ -169,7 +202,7 @@ class Interpreter: The values must be shareable. """ ns = dict(ns, **kwargs) if ns is not None else kwargs - _interpreters.set___main___attrs(self._id, ns) + _interpreters.set___main___attrs(self._id, ns, restrict=True) def exec(self, code, /): """Run the given source code in the interpreter. @@ -189,7 +222,7 @@ class Interpreter: that time, the previous interpreter is allowed to run in other threads. """ - excinfo = _interpreters.exec(self._id, code) + excinfo = _interpreters.exec(self._id, code, restrict=True) if excinfo is not None: raise ExecutionFailed(excinfo) @@ -209,7 +242,7 @@ class Interpreter: # XXX Support args and kwargs. # XXX Support arbitrary callables. # XXX Support returning the return value (e.g. via pickle). - excinfo = _interpreters.call(self._id, callable) + excinfo = _interpreters.call(self._id, callable, restrict=True) if excinfo is not None: raise ExecutionFailed(excinfo) diff --git a/Lib/test/test_interpreters/test_api.py b/Lib/test/test_interpreters/test_api.py index 769bb7b..9a7c7f2 100644 --- a/Lib/test/test_interpreters/test_api.py +++ b/Lib/test/test_interpreters/test_api.py @@ -21,6 +21,14 @@ from .utils import ( ) +WHENCE_STR_UNKNOWN = 'unknown' +WHENCE_STR_RUNTIME = 'runtime init' +WHENCE_STR_LEGACY_CAPI = 'legacy C-API' +WHENCE_STR_CAPI = 'C-API' +WHENCE_STR_XI = 'cross-interpreter C-API' +WHENCE_STR_STDLIB = '_interpreters module' + + class ModuleTests(TestBase): def test_queue_aliases(self): @@ -173,10 +181,11 @@ class GetCurrentTests(TestBase): text = self.run_temp_from_capi(f""" import {interpreters.__name__} as interpreters interp = interpreters.get_current() - print(interp.id) + print((interp.id, interp.whence)) """) - interpid = eval(text) + interpid, whence = eval(text) self.assertEqual(interpid, expected) + self.assertEqual(whence, WHENCE_STR_CAPI) class ListAllTests(TestBase): @@ -228,22 +237,22 @@ class ListAllTests(TestBase): interpid4 = interpid3 + 1 interpid5 = interpid4 + 1 expected = [ - (mainid,), - (interpid1,), - (interpid2,), - (interpid3,), - (interpid4,), - (interpid5,), + (mainid, WHENCE_STR_RUNTIME), + (interpid1, WHENCE_STR_STDLIB), + (interpid2, WHENCE_STR_STDLIB), + (interpid3, WHENCE_STR_STDLIB), + (interpid4, WHENCE_STR_CAPI), + (interpid5, WHENCE_STR_STDLIB), ] expected2 = expected[:-2] text = self.run_temp_from_capi(f""" import {interpreters.__name__} as interpreters interp = interpreters.create() print( - [(i.id,) for i in interpreters.list_all()]) + [(i.id, i.whence) for i in interpreters.list_all()]) """) res = eval(text) - res2 = [(i.id,) for i in interpreters.list_all()] + res2 = [(i.id, i.whence) for i in interpreters.list_all()] self.assertEqual(res, expected) self.assertEqual(res2, expected2) @@ -299,6 +308,38 @@ class InterpreterObjectTests(TestBase): with self.assertRaises(AttributeError): interp.id = 1_000_000 + def test_whence(self): + main = interpreters.get_main() + interp = interpreters.create() + + with self.subTest('main'): + self.assertEqual(main.whence, WHENCE_STR_RUNTIME) + + with self.subTest('from _interpreters'): + self.assertEqual(interp.whence, WHENCE_STR_STDLIB) + + with self.subTest('from C-API'): + text = self.run_temp_from_capi(f""" + import {interpreters.__name__} as interpreters + interp = interpreters.get_current() + print(repr(interp.whence)) + """) + whence = eval(text) + self.assertEqual(whence, WHENCE_STR_CAPI) + + with self.subTest('readonly'): + for value in [ + None, + WHENCE_STR_UNKNOWN, + WHENCE_STR_RUNTIME, + WHENCE_STR_STDLIB, + WHENCE_STR_CAPI, + ]: + with self.assertRaises(AttributeError): + interp.whence = value + with self.assertRaises(AttributeError): + main.whence = value + def test_hashable(self): interp = interpreters.create() expected = hash(interp.id) @@ -568,41 +609,42 @@ class TestInterpreterClose(TestBase): with self.subTest('running __main__ (from self)'): with self.interpreter_from_capi() as interpid: with self.assertRaisesRegex(ExecutionFailed, - 'InterpreterError.*current'): + 'InterpreterError.*unrecognized'): self.run_from_capi(interpid, script, main=True) with self.subTest('running, but not __main__ (from self)'): with self.assertRaisesRegex(ExecutionFailed, - 'InterpreterError.*current'): + 'InterpreterError.*unrecognized'): self.run_temp_from_capi(script) with self.subTest('running __main__ (from other)'): with self.interpreter_obj_from_capi() as (interp, interpid): with self.running_from_capi(interpid, main=True): - with self.assertRaisesRegex(InterpreterError, 'running'): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): interp.close() # Make sure it wssn't closed. self.assertTrue( - interp.is_running()) + self.interp_exists(interpid)) - # The rest must be skipped until we deal with running threads when - # interp.close() is called. - return + # The rest would be skipped until we deal with running threads when + # interp.close() is called. However, the "whence" restrictions + # trigger first. with self.subTest('running, but not __main__ (from other)'): with self.interpreter_obj_from_capi() as (interp, interpid): with self.running_from_capi(interpid, main=False): - with self.assertRaisesRegex(InterpreterError, 'not managed'): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): interp.close() # Make sure it wssn't closed. - self.assertFalse(interp.is_running()) + self.assertTrue( + self.interp_exists(interpid)) with self.subTest('not running (from other)'): - with self.interpreter_obj_from_capi() as (interp, _): - with self.assertRaisesRegex(InterpreterError, 'not managed'): + with self.interpreter_obj_from_capi() as (interp, interpid): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): interp.close() - # Make sure it wssn't closed. - self.assertFalse(interp.is_running()) + self.assertTrue( + self.interp_exists(interpid)) class TestInterpreterPrepareMain(TestBase): @@ -671,12 +713,11 @@ class TestInterpreterPrepareMain(TestBase): @requires_test_modules def test_created_with_capi(self): - with self.interpreter_from_capi() as interpid: - interp = interpreters.Interpreter(interpid) - interp.prepare_main({'spam': True}) - rc = _testinternalcapi.exec_interpreter(interpid, - 'assert spam is True') - assert rc == 0, rc + with self.interpreter_obj_from_capi() as (interp, interpid): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + interp.prepare_main({'spam': True}) + with self.assertRaisesRegex(ExecutionFailed, 'NameError'): + self.run_from_capi(interpid, 'assert spam is True') class TestInterpreterExec(TestBase): @@ -835,7 +876,7 @@ class TestInterpreterExec(TestBase): def test_created_with_capi(self): with self.interpreter_obj_from_capi() as (interp, _): - with self.assertRaisesRegex(ExecutionFailed, 'it worked'): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): interp.exec('raise Exception("it worked!")') # test_xxsubinterpreters covers the remaining @@ -1114,14 +1155,6 @@ class LowLevelTests(TestBase): # encountered by the high-level module, thus they # mostly shouldn't matter as much. - def interp_exists(self, interpid): - try: - _interpreters.is_running(interpid) - except InterpreterNotFoundError: - return False - else: - return True - def test_new_config(self): # This test overlaps with # test.test_capi.test_misc.InterpreterConfigTests. @@ -1245,32 +1278,35 @@ class LowLevelTests(TestBase): _interpreters.new_config(gil=value) def test_get_main(self): - interpid, = _interpreters.get_main() + interpid, whence = _interpreters.get_main() self.assertEqual(interpid, 0) + self.assertEqual(whence, _interpreters.WHENCE_RUNTIME) + self.assertEqual( + _interpreters.whence(interpid), + _interpreters.WHENCE_RUNTIME) def test_get_current(self): with self.subTest('main'): main, *_ = _interpreters.get_main() - interpid, = _interpreters.get_current() + interpid, whence = _interpreters.get_current() self.assertEqual(interpid, main) + self.assertEqual(whence, _interpreters.WHENCE_RUNTIME) script = f""" import {_interpreters.__name__} as _interpreters - interpid, = _interpreters.get_current() - print(interpid) + interpid, whence = _interpreters.get_current() + print((interpid, whence)) """ def parse_stdout(text): - parts = text.split() - assert len(parts) == 1, parts - interpid, = parts - interpid = int(interpid) - return interpid, + interpid, whence = eval(text) + return interpid, whence with self.subTest('from _interpreters'): orig = _interpreters.create() text = self.run_and_capture(orig, script) - interpid, = parse_stdout(text) + interpid, whence = parse_stdout(text) self.assertEqual(interpid, orig) + self.assertEqual(whence, _interpreters.WHENCE_STDLIB) with self.subTest('from C-API'): last = 0 @@ -1278,8 +1314,9 @@ class LowLevelTests(TestBase): last = max(last, id) expected = last + 1 text = self.run_temp_from_capi(script) - interpid, = parse_stdout(text) + interpid, whence = parse_stdout(text) self.assertEqual(interpid, expected) + self.assertEqual(whence, _interpreters.WHENCE_CAPI) def test_list_all(self): mainid, *_ = _interpreters.get_main() @@ -1287,17 +1324,17 @@ class LowLevelTests(TestBase): interpid2 = _interpreters.create() interpid3 = _interpreters.create() expected = [ - (mainid,), - (interpid1,), - (interpid2,), - (interpid3,), + (mainid, _interpreters.WHENCE_RUNTIME), + (interpid1, _interpreters.WHENCE_STDLIB), + (interpid2, _interpreters.WHENCE_STDLIB), + (interpid3, _interpreters.WHENCE_STDLIB), ] with self.subTest('main'): res = _interpreters.list_all() self.assertEqual(res, expected) - with self.subTest('from _interpreters'): + with self.subTest('via interp from _interpreters'): text = self.run_and_capture(interpid2, f""" import {_interpreters.__name__} as _interpreters print( @@ -1307,15 +1344,15 @@ class LowLevelTests(TestBase): res = eval(text) self.assertEqual(res, expected) - with self.subTest('from C-API'): + with self.subTest('via interp from C-API'): interpid4 = interpid3 + 1 interpid5 = interpid4 + 1 expected2 = expected + [ - (interpid4,), - (interpid5,), + (interpid4, _interpreters.WHENCE_CAPI), + (interpid5, _interpreters.WHENCE_STDLIB), ] expected3 = expected + [ - (interpid5,), + (interpid5, _interpreters.WHENCE_STDLIB), ] text = self.run_temp_from_capi(f""" import {_interpreters.__name__} as _interpreters @@ -1380,6 +1417,12 @@ class LowLevelTests(TestBase): with self.assertRaises(ValueError): _interpreters.create(orig) + with self.subTest('whence'): + interpid = _interpreters.create() + self.assertEqual( + _interpreters.whence(interpid), + _interpreters.WHENCE_STDLIB) + @requires_test_modules def test_destroy(self): with self.subTest('from _interpreters'): @@ -1401,6 +1444,10 @@ class LowLevelTests(TestBase): with self.subTest('from C-API'): interpid = _testinternalcapi.create_interpreter() + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + _interpreters.destroy(interpid, restrict=True) + self.assertTrue( + self.interp_exists(interpid)) _interpreters.destroy(interpid) self.assertFalse( self.interp_exists(interpid)) @@ -1433,6 +1480,8 @@ class LowLevelTests(TestBase): with self.subTest('from C-API'): orig = _interpreters.new_config('isolated') with self.interpreter_from_capi(orig) as interpid: + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + _interpreters.get_config(interpid, restrict=True) config = _interpreters.get_config(interpid) self.assert_ns_equal(config, orig) @@ -1446,10 +1495,10 @@ class LowLevelTests(TestBase): with self.subTest('stdlib'): interpid = _interpreters.create() whence = _interpreters.whence(interpid) - self.assertEqual(whence, _interpreters.WHENCE_XI) + self.assertEqual(whence, _interpreters.WHENCE_STDLIB) for orig, name in { - # XXX Also check WHENCE_UNKNOWN. + _interpreters.WHENCE_UNKNOWN: 'not ready', _interpreters.WHENCE_LEGACY_CAPI: 'legacy C-API', _interpreters.WHENCE_CAPI: 'C-API', _interpreters.WHENCE_XI: 'cross-interpreter C-API', @@ -1481,38 +1530,40 @@ class LowLevelTests(TestBase): self.assertEqual(whence, _interpreters.WHENCE_LEGACY_CAPI) def test_is_running(self): - with self.subTest('main'): - interpid, *_ = _interpreters.get_main() + def check(interpid, expected): + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + _interpreters.is_running(interpid, restrict=True) running = _interpreters.is_running(interpid) - self.assertTrue(running) + self.assertIs(running, expected) with self.subTest('from _interpreters (running)'): interpid = _interpreters.create() with self.running(interpid): running = _interpreters.is_running(interpid) - self.assertTrue(running) + self.assertTrue(running) with self.subTest('from _interpreters (not running)'): interpid = _interpreters.create() running = _interpreters.is_running(interpid) self.assertFalse(running) + with self.subTest('main'): + interpid, *_ = _interpreters.get_main() + check(interpid, True) + with self.subTest('from C-API (running __main__)'): with self.interpreter_from_capi() as interpid: with self.running_from_capi(interpid, main=True): - running = _interpreters.is_running(interpid) - self.assertTrue(running) + check(interpid, True) with self.subTest('from C-API (running, but not __main__)'): with self.interpreter_from_capi() as interpid: with self.running_from_capi(interpid, main=False): - running = _interpreters.is_running(interpid) - self.assertFalse(running) + check(interpid, False) with self.subTest('from C-API (not running)'): with self.interpreter_from_capi() as interpid: - running = _interpreters.is_running(interpid) - self.assertFalse(running) + check(interpid, False) def test_exec(self): with self.subTest('run script'): @@ -1549,6 +1600,9 @@ class LowLevelTests(TestBase): with self.subTest('from C-API'): with self.interpreter_from_capi() as interpid: + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + _interpreters.exec(interpid, 'raise Exception("it worked!")', + restrict=True) exc = _interpreters.exec(interpid, 'raise Exception("it worked!")') self.assertIsNot(exc, None) self.assertEqual(exc.msg, 'it worked!') @@ -1574,6 +1628,7 @@ class LowLevelTests(TestBase): errdisplay=exc.errdisplay, )) + @requires_test_modules def test_set___main___attrs(self): with self.subTest('from _interpreters'): interpid = _interpreters.create() @@ -1595,9 +1650,15 @@ class LowLevelTests(TestBase): with self.subTest('from C-API'): with self.interpreter_from_capi() as interpid: + with self.assertRaisesRegex(InterpreterError, 'unrecognized'): + _interpreters.set___main___attrs(interpid, {'spam': True}, + restrict=True) _interpreters.set___main___attrs(interpid, {'spam': True}) - exc = _interpreters.exec(interpid, 'assert spam is True') - self.assertIsNone(exc) + rc = _testinternalcapi.exec_interpreter( + interpid, + 'assert spam is True', + ) + self.assertEqual(rc, 0) if __name__ == '__main__': diff --git a/Lib/test/test_interpreters/utils.py b/Lib/test/test_interpreters/utils.py index d921794..08768c0 100644 --- a/Lib/test/test_interpreters/utils.py +++ b/Lib/test/test_interpreters/utils.py @@ -509,6 +509,14 @@ class TestBase(unittest.TestCase): else: return text + def interp_exists(self, interpid): + try: + _interpreters.whence(interpid) + except _interpreters.InterpreterNotFoundError: + return False + else: + return True + @requires_test_modules @contextlib.contextmanager def interpreter_from_capi(self, config=None, whence=None): @@ -545,7 +553,12 @@ class TestBase(unittest.TestCase): @contextlib.contextmanager def interpreter_obj_from_capi(self, config='legacy'): with self.interpreter_from_capi(config) as interpid: - yield interpreters.Interpreter(interpid), interpid + interp = interpreters.Interpreter( + interpid, + _whence=_interpreters.WHENCE_CAPI, + _ownsref=False, + ) + yield interp, interpid @contextlib.contextmanager def capturing(self, script): |