1 files changed, 77 insertions, 14 deletions

diff --git a/Lib/test/test_capi.py b/Lib/test/test_capi.py
index 3f1d63c..c2c633f 100644
--- a/Lib/test/test_capi.py
+++ b/Lib/test/test_capi.py
@@ -2,12 +2,18 @@
 # these are all functions _testcapi exports whose name begins with 'test_'.
 
 from __future__ import with_statement
+import os
+import pickle
+import random
+import subprocess
 import sys
 import time
-import random
 import unittest
-import threading
 from test import support
+try:
+    import threading
+except ImportError:
+    threading = None
 import _testcapi
 
 
@@ -32,7 +38,24 @@ class CAPITest(unittest.TestCase):
         self.assertEqual(testfunction.attribute, "test")
         self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute", "test")
 
-
+    @unittest.skipUnless(threading, 'Threading required for this test.')
+    def test_no_FatalError_infinite_loop(self):
+        p = subprocess.Popen([sys.executable, "-c",
+                              'import _testcapi;'
+                              '_testcapi.crash_no_current_thread()'],
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        (out, err) = p.communicate()
+        self.assertEqual(out, b'')
+        # This used to cause an infinite loop.
+        self.assertEqual(err.rstrip(),
+                         b'Fatal Python error:'
+                         b' PyThreadState_Get: no current thread')
+
+    def test_memoryview_from_NULL_pointer(self):
+        self.assertRaises(ValueError, _testcapi.make_memoryview_from_NULL_pointer)
+
+@unittest.skipUnless(threading, 'Threading required for this test.')
 class TestPendingCalls(unittest.TestCase):
 
     def pendingcalls_submit(self, l, n):
@@ -115,13 +138,62 @@ class TestPendingCalls(unittest.TestCase):
         self.pendingcalls_submit(l, n)
         self.pendingcalls_wait(l, n)
 
+    def test_subinterps(self):
+        # XXX this test leaks in refleak runs
+        import builtins
+        r, w = os.pipe()
+        code = """if 1:
+            import sys, builtins, pickle
+            with open({:d}, "wb") as f:
+                pickle.dump(id(sys.modules), f)
+                pickle.dump(id(builtins), f)
+            """.format(w)
+        with open(r, "rb") as f:
+            ret = _testcapi.run_in_subinterp(code)
+            self.assertEqual(ret, 0)
+            self.assertNotEqual(pickle.load(f), id(sys.modules))
+            self.assertNotEqual(pickle.load(f), id(builtins))
+
 # Bug #6012
 class Test6012(unittest.TestCase):
     def test(self):
         self.assertEqual(_testcapi.argparsing("Hello", "World"), 1)
 
+
+class EmbeddingTest(unittest.TestCase):
+
+    @unittest.skipIf(
+        sys.platform.startswith('win'),
+        "test doesn't work under Windows")
+    def test_subinterps(self):
+        # XXX only tested under Unix checkouts
+        basepath = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+        oldcwd = os.getcwd()
+        # This is needed otherwise we get a fatal error:
+        # "Py_Initialize: Unable to get the locale encoding
+        # LookupError: no codec search functions registered: can't find encoding"
+        os.chdir(basepath)
+        try:
+            exe = os.path.join(basepath, "Modules", "_testembed")
+            if not os.path.exists(exe):
+                self.skipTest("%r doesn't exist" % exe)
+            p = subprocess.Popen([exe],
+                                 stdout=subprocess.PIPE,
+                                 stderr=subprocess.PIPE)
+            (out, err) = p.communicate()
+            self.assertEqual(p.returncode, 0,
+                             "bad returncode %d, stderr is %r" %
+                             (p.returncode, err))
+            if support.verbose:
+                print()
+                print(out.decode('latin1'))
+                print(err.decode('latin1'))
+        finally:
+            os.chdir(oldcwd)
+
+
 def test_main():
-    support.run_unittest(CAPITest)
+    support.run_unittest(CAPITest, TestPendingCalls, Test6012, EmbeddingTest)
 
     for name in dir(_testcapi):
         if name.startswith('test_'):
@@ -148,23 +220,14 @@ def test_main():
             raise support.TestFailed(
                         "Couldn't find main thread correctly in the list")
 
-    try:
-        _testcapi._test_thread_state
-        have_thread_state = True
-    except AttributeError:
-        have_thread_state = False
-
-    if have_thread_state:
+    if threading:
         import _thread
         import time
         TestThreadState()
-        import threading
         t = threading.Thread(target=TestThreadState)
         t.start()
         t.join()
 
-    support.run_unittest(TestPendingCalls, Test6012)
-
 
 if __name__ == "__main__":
     test_main()