Use _PyCode_GetScriptXIData().

Author: ericsnowcurrently

Lib/test/support/interpreters/channels.py

@@ -69,7 +69,7 @@ def list_all():
         if not hasattr(send, '_unboundop'):
             send._set_unbound(unboundop)
         else:
-            assert send._unbound[0] == op
+            assert send._unbound[0] == unboundop
         channels.append(chan)
     return channels
 

Lib/test/test__interpreters.py

@@ -474,13 +474,15 @@ def setUp(self):
 
     def test_signatures(self):
         # See https://github.com/python/cpython/issues/126654
-        msg = "expected 'shared' to be a dict"
+        msg = r'_interpreters.exec\(\) argument 3 must be dict, not int'
         with self.assertRaisesRegex(TypeError, msg):
             _interpreters.exec(self.id, 'a', 1)
         with self.assertRaisesRegex(TypeError, msg):
             _interpreters.exec(self.id, 'a', shared=1)
+        msg = r'_interpreters.run_string\(\) argument 3 must be dict, not int'
         with self.assertRaisesRegex(TypeError, msg):
             _interpreters.run_string(self.id, 'a', shared=1)
+        msg = r'_interpreters.run_func\(\) argument 3 must be dict, not int'
         with self.assertRaisesRegex(TypeError, msg):
             _interpreters.run_func(self.id, lambda: None, shared=1)
 
@@ -952,7 +954,8 @@ def test_invalid_syntax(self):
             """)
 
         with self.subTest('script'):
-            self.assert_run_failed(SyntaxError, script)
+            with self.assertRaises(SyntaxError):
+                _interpreters.run_string(self.id, script)
 
         with self.subTest('module'):
             modname = 'spam_spam_spam'
@@ -1019,12 +1022,19 @@ def script():
             with open(w, 'w', encoding="utf-8") as spipe:
                 with contextlib.redirect_stdout(spipe):
                     print('it worked!', end='')
+        failed = None
         def f():
-            _interpreters.set___main___attrs(self.id, dict(w=w))
-            _interpreters.run_func(self.id, script)
+            nonlocal failed
+            try:
+                _interpreters.set___main___attrs(self.id, dict(w=w))
+                _interpreters.run_func(self.id, script)
+            except Exception as exc:
+                failed = exc
         t = threading.Thread(target=f)
         t.start()
         t.join()
+        if failed:
+            raise Exception from failed
 
         with open(r, encoding="utf-8") as outfile:
             out = outfile.read()
@@ -1053,12 +1063,9 @@ def test_closure(self):
         spam = True
         def script():
             assert spam
-
-        with self.assertRaises(TypeError):
+        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'

Lib/test/test_interpreters/test_api.py

@@ -839,9 +839,16 @@ def test_bad_script(self):
             interp.exec(10)
 
     def test_bytes_for_script(self):
+        r, w = self.pipe()
+        RAN = b'R'
+        DONE = b'D'
         interp = interpreters.create()
-        with self.assertRaises(TypeError):
-            interp.exec(b'print("spam")')
+        interp.exec(f"""if True:
+            import os
+            os.write({w}, {RAN!r})
+            """)
+        os.write(w, DONE)
+        self.assertEqual(os.read(r, 1), RAN)
 
     def test_with_background_threads_still_running(self):
         r_interp, w_interp = self.pipe()
@@ -1010,8 +1017,8 @@ def test_call(self):
 
         for i, (callable, args, kwargs) in enumerate([
             (call_func_noop, (), {}),
-            (call_func_return_shareable, (), {}),
-            (call_func_return_not_shareable, (), {}),
+            #(call_func_return_shareable, (), {}),
+            #(call_func_return_not_shareable, (), {}),
             (Spam.noop, (), {}),
         ]):
             with self.subTest(f'success case #{i+1}'):
@@ -1036,6 +1043,8 @@ def test_call(self):
             (call_func_complex, ('custom', 'spam!'), {}),
             (call_func_complex, ('custom-inner', 'eggs!'), {}),
             (call_func_complex, ('???',), {'exc': ValueError('spam')}),
+            (call_func_return_shareable, (), {}),
+            (call_func_return_not_shareable, (), {}),
         ]):
             with self.subTest(f'invalid case #{i+1}'):
                 with self.assertRaises(Exception):
@@ -1051,8 +1060,8 @@ def test_call_in_thread(self):
 
         for i, (callable, args, kwargs) in enumerate([
             (call_func_noop, (), {}),
-            (call_func_return_shareable, (), {}),
-            (call_func_return_not_shareable, (), {}),
+            #(call_func_return_shareable, (), {}),
+            #(call_func_return_not_shareable, (), {}),
             (Spam.noop, (), {}),
         ]):
             with self.subTest(f'success case #{i+1}'):
@@ -1079,6 +1088,8 @@ def test_call_in_thread(self):
             (call_func_complex, ('custom', 'spam!'), {}),
             (call_func_complex, ('custom-inner', 'eggs!'), {}),
             (call_func_complex, ('???',), {'exc': ValueError('spam')}),
+            (call_func_return_shareable, (), {}),
+            (call_func_return_not_shareable, (), {}),
         ]):
             with self.subTest(f'invalid case #{i+1}'):
                 if args or kwargs:
@@ -1618,8 +1629,10 @@ def test_exec(self):
     def test_call(self):
         with self.subTest('no args'):
             interpid = _interpreters.create()
-            exc = _interpreters.call(interpid, call_func_return_shareable)
-            self.assertIs(exc, None)
+            with self.assertRaises(ValueError):
+                _interpreters.call(interpid, call_func_return_shareable)
+#            exc = _interpreters.call(interpid, call_func_return_shareable)
+#            self.assertIs(exc, None)
 
         with self.subTest('uncaught exception'):
             interpid = _interpreters.create()