Sketch out tests.

python · ericsnowcurrently · Apr 11, 2024 · Apr 8, 2024 · Apr 5, 2024 · Apr 5, 2024
commit 9b7bdc4d096a1d6b7e049f6dc6a772a24caf0aab
@@ -157,6 +157,20 @@ def test_idempotent(self):
             id2 = id(interp)
             self.assertNotEqual(id1, id2)
 
+    def test_unmanaged(self):
+        with self.unmanaged_interpreter() as unmanaged:
+            print(unmanaged)
+            err = unmanaged.exec(dedent(f"""
+                import {interpreters.__name__} as interpreters
+                err = None
+                try:
+                    interpreters.get_current()
+                except ValueError as exc:
+                    err = exc
+                assert exc is not None
+                """))
+            self.assertEqual(err, '')
+
 
 class ListAllTests(TestBase):
 
@@ -199,6 +213,9 @@ def test_idempotent(self):
         for interp1, interp2 in zip(actual, expected):
             self.assertIs(interp1, interp2)
 
+    def test_unmanaged(self):
+        ...
+
 
 class InterpreterObjectTests(TestBase):
 
@@ -1071,7 +1088,7 @@ def test_get_config(self):
         with self.subTest('main'):
             expected = _interpreters.new_config('legacy')
             expected.gil = 'own'
-            interpid = _interpreters.get_main()
+            interpid, _ = _interpreters.get_main()
             config = _interpreters.get_config(interpid)
             self.assert_ns_equal(config, expected)
 
@@ -1138,6 +1155,12 @@ def test_create(self):
             with self.assertRaises(ValueError):
                 _interpreters.create(orig)
 
+    def test_current(self):
+        ...
+
+    def test_list_all(self):
+        ...
+
 
 if __name__ == '__main__':
     # Test needs to be a package, so we can do relative imports.

@@ -1,6 +1,7 @@
 import contextlib
 import os
 import os.path
+import pickle
 import subprocess
 import sys
 import tempfile
@@ -66,6 +67,111 @@ def run():
     t.join()
 
 
+class UnmanagedInterpreter:
+
+    @classmethod
+    def start(cls, create_pipe, *, legacy=True):
+        r_in, _ = inpipe = create_pipe()
+        r_out, w_out = outpipe = create_pipe()
+
+        script = dedent(f"""
+            import pickle, os, traceback, _testcapi
+
+            # Send the interpreter ID.
+            interpid = _testcapi.get_current_interpid()
+            os.write({w_out}, pickle.dumps(interpid))
+
+            # Run exec requests until "done".
+            script = b''
+            while True:
+                ch = os.read({r_in}, 1)
+                if ch == b'\0':
+                    if not script:
+                        # done!
+                        break
+
+                    # Run the provided script.
+                    try:
+                        exec(script)
+                    except Exception as exc:
+                        traceback.print_exc()
+                        err = traceback.format_exception_only(exc)
+                        os.write(w_out, err.encode('utf-8'))
+                    os.write(w_out, b'\0')
+                    script = b''
+                else:
+                    script += ch
+            """)
+        def run():
+            try:
+                if legacy:
+                    rc = support.run_in_subinterp(script)
+                else:
+                    rc = support.run_in_subinterp_with_config(script)
+                assert rc == 0, rc
+            except BaseException:
+                os.write(w_out, b'\0')
+                raise  # re-raise
+        t = threading.Thread(target=run)
+        t.start()
+        try:
+            # Get the interpreter ID.
+            data = os.read(r_out, 10)
+            assert len(data) < 10, repr(data) 
+            assert data != b'\0'
+            interpid = pickle.loads(data)
+
+            return cls(interpid, t, inpipe, outpipe)
+        except BaseException:
+            os.write(w_out, b'\0')
+            raise  # re-raise
+
+    def __init__(self, id, thread, inpipe, outpipe):
+        self._id = id
+        self._thread = thread
+        self._r_in, self._w_in = inpipe
+        self._r_out, self._w_out = outpipe
+
+    def __repr__(self):
+        return f'<{type(self).__name__} {self._id}>'
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.shutdown()
+
+    @property
+    def id(self):
+        return self._id
+
+    def exec(self, code):
+        if isinstance(code, str):
+            code = code.encode('utf-8')
+        elif not isinstance(code, bytes):
+            raise TypeError(f'expected str/bytes, got {code!r}')
+        if not code:
+            raise ValueError('empty code')
+        os.write(self._w_in, code)
+        os.write(self._w_in, b'\0')
+
+        out = b''
+        while True:
+            ch = os.read(self._r_out, 1)
+            if ch == b'\0':
+                break
+            out += ch
+        return out.decode('utf-8')
+
+    def shutdown(self, *, wait=True):
+        t = self._thread
+        self._thread = None
+        if t is not None:
+            os.write(self._w_in, b'\0')
+            if wait:
+                t.join()
+
+
 class TestBase(unittest.TestCase):
 
     def tearDown(self):
@@ -175,3 +281,40 @@ def assert_ns_equal(self, ns1, ns2, msg=None):
         diff = f'namespace({diff})'
         standardMsg = self._truncateMessage(standardMsg, diff)
         self.fail(self._formatMessage(msg, standardMsg))
+
+    def unmanaged_interpreter(self, *, legacy=True):
+        return UnmanagedInterpreter.start(self.pipe, legacy=legacy)
+#    @contextlib.contextmanager
+#    def unmanaged_interpreter(self, *, legacy=True):
+#        r_id, w_id = self.pipe()
+#        r_done, w_done = self.pipe()
+#        script = dedent(f"""
+#            import marshal, os, _testcapi
+#
+#            # Send the interpreter ID.
+#            interpid = _testcapi.get_current_interpid()
+#            data = marshal.dumps(interpid)
+#            os.write({w_id}, data)
+#
+#            # Wait for "done".
+#            os.read({r_done}, 1)
+#            """)
+#        rc = None
+#        def task():
+#            nonlocal rc
+#            if legacy:
+#                rc = support.run_in_subinterp(script)
+#            else:
+#                rc = support.run_in_subinterp_with_config(script)
+#        t = threading.Thread(target=task)
+#        t.start()
+#        try:
+#            # Get the interpreter ID.
+#            data = os.read(r_id, 10)
+#            assert len(data) < 10, repr(data) 
+#            yield marshal.loads(data)
+#        finally:
+#            # Send "done".
+#            os.write(w_done, b'\0')
+#            t.join()
+#        self.assertEqual(rc, 0)