Port subinterpreter tests to higher-level API

While still not a fully approved API, this looks more like PEP 734, and
what we'll hopefully get in the future.

Signed-off-by: Rodrigo Tobar <[email protected]>

tests/test_subinterpreter.py

@@ -9,77 +9,48 @@
 import pytest
 
 try:
-    import _interpreters
-    import _interpqueues as _queues
+    from test.support import interpreters
+    from test.support.interpreters import queues
 except ImportError:
     pytest.skip("No sub-interpreter support", allow_module_level=True)
 
-@pytest.fixture(name="interpreter_id")
-def interpreter_id_fixture():
-    intp_id = _interpreters.create()
-    yield intp_id
-    _interpreters.destroy(intp_id)
 
+@pytest.fixture(name="interpreter")
+def interpreter_fixture():
+    interpreter = interpreters.create()
+    yield interpreter
+    interpreter.close()
 
-_UNBOUND_ERROR = 2
-_PICKLED = 1
 
-@pytest.fixture(name="queue_id")
-def queue_id_fixture():
-    maxsize = 0
-    queue_id = _queues.create(maxsize, _PICKLED, _UNBOUND_ERROR)
-    _queues.bind(queue_id)
-    yield queue_id
-    _queues.release(queue_id)
+def test_ijson_can_be_loaded(interpreter):
+    interpreter.exec('import ijson')
 
 
-def test_ijson_can_be_loaded(interpreter_id):
-    execinfo = _interpreters.exec(interpreter_id, 'import ijson')
-    assert execinfo is None
-
-
-def test_ijson_yajl2_backend_can_be_loaded(interpreter_id):
+def test_ijson_yajl2_backend_can_be_loaded(interpreter):
     spec = importlib.util.find_spec("ijson.backends._yajl2")
     if spec is None:
         pytest.skip("yajl2_c is not built")
-    execinfo = _interpreters.exec(interpreter_id, 'import ijson')
-    assert execinfo is None
-    execinfo = _interpreters.exec(interpreter_id, 'ijson.get_backend("yajl2_c")')
-    assert execinfo is None
+    interpreter.exec('import ijson')
+    interpreter.exec('ijson.get_backend("yajl2_c")')
 
 
-SIMPLE_IJSON_ITEMS_USAGE = f"""
-import ijson
-import pickle
-import _interpqueues as _queues
-
-value_out = next(ijson.items(f'{{value_in}}', prefix=''))
-_queues.put(queue_id, pickle.dumps(value_out), {_PICKLED}, {_UNBOUND_ERROR})
-"""
-
-def test_ijson_can_run(interpreter_id, queue_id):
-
-    if sys.platform == "win32" and os.getenv("CIBUILDWHEEL", "0") == "1":
-        pytest.xfail("Currently failing under cibuildwheel@win32")
+def test_ijson_can_run(interpreter):
 
+    queue = queues.create()
     VALUE = 43
-    _interpreters.set___main___attrs(interpreter_id, {"queue_id": queue_id, "value_in": VALUE}, restrict=True)
+    interpreter.prepare_main(queue_id=queue.id, value_in=VALUE)
 
     def ijson_in_subinterpreter():
-        execinfo = _interpreters.exec(interpreter_id, SIMPLE_IJSON_ITEMS_USAGE)
-        assert execinfo is None
+        from test.support.interpreters.queues import Queue
 
-    thread = threading.Thread(target=ijson_in_subinterpreter)
-    thread.start()
+        import ijson
 
-    delay = 10 / 1000
-    while True:
-        try:
-            result_tuple = _queues.get(queue_id)
-            result = pickle.loads(result_tuple[0])
-            break
-        except queue.Empty:
-            time.sleep(delay)
+        value_out = next(ijson.items(str(value_in).encode('ascii'), prefix=''))
+        queue = Queue(queue_id)
+        queue.put(value_out)
 
+    thread = interpreter.call_in_thread(ijson_in_subinterpreter)
+    value_out = queue.get(timeout=5)
     thread.join()
-    assert result == VALUE
+
+    assert VALUE == value_out