run echo_watch test in a subprocess

to avoid messing with pytest output capture

I'm pretty sure it's _possible_ to do this in-process, but I can't seem to figure it out

ipykernel/tests/test_io.py

@@ -2,7 +2,9 @@
 
 import io
 import os
+import subprocess
 import sys
+import time
 import warnings
 from unittest import mock
 
@@ -113,37 +115,101 @@ def test_outstream(iopub_thread):
         assert stream.writable()
 
 
+def subprocess_test_echo_watch():
+    # handshake Pub subscription
+    session = Session(key=b'abc')
+
+    with zmq.Context() as ctx:
+        # use PUSH socket to avoid subscription issues
+        with ctx.socket(zmq.PUSH) as pub:
+            pub.connect(os.environ["IOPUB_URL"])
+            iopub_thread = IOPubThread(pub)
+            stdout_fd = sys.stdout.fileno()
+            sys.stdout.flush()
+            stream = OutStream(
+                session,
+                iopub_thread,
+                "stdout",
+                isatty=True,
+                echo=sys.stdout,
+                watchfd="force",
+            )
+            save_stdout = sys.stdout
+            with stream, mock.patch.object(sys, "stdout", stream):
+                # write to low-level FD
+                os.write(stdout_fd, b"fd\n")
+                # print (writes to stream)
+                print("print")
+                sys.stdout.flush()
+                # write to unwrapped __stdout__ (should also go to original FD)
+                sys.__stdout__.write("__stdout__\n")
+                sys.__stdout__.flush()
+                # write to original sys.stdout (should be the same as __stdout__)
+                save_stdout.write("stdout\n")
+                save_stdout.flush()
+                stream.flush()
+                # we don't have a sync flush on _reading_ from the watched pipe
+                time.sleep(0.5)
+            iopub_thread.stop()
+            iopub_thread.close()
+
+
 @pytest.mark.skipif(sys.platform.startswith("win"), reason="Windows")
-def test_echo_watch(capfd, iopub_thread):
+def test_echo_watch(ctx):
     """Test echo on underlying FD while capturing the same FD
 
-    If not careful, this
+    Test runs in a subprocess to avoid messing with pytest output capturing.
     """
-    session = Session()
-
-    stdout_fd = sys.stdout.fileno()
-    stream = OutStream(
-        session,
-        iopub_thread,
+    s = ctx.socket(zmq.PULL)
+    port = s.bind_to_random_port("tcp://127.0.0.1")
+    url = f"tcp://127.0.0.1:{port}"
+    session = Session(key=b'abc')
+    messages = []
+    stdout_chunks = []
+    with s:
+        env = dict(os.environ)
+        env["IOPUB_URL"] = url
+        env["PYTHONUNBUFFERED"] = "1"
+        env.pop("PYTEST_CURRENT_TEST", None)
+        p = subprocess.run(
+            [
+                sys.executable,
+                "-c",
+                f"import {__name__}; {__name__}.subprocess_test_echo_watch()",
+            ],
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        print(f"{p.stdout=}")
+        print(f"{p.stderr}=", file=sys.stderr)
+        assert p.returncode == 0
+        while s.poll(timeout=100):
+            ident, msg = session.recv(s)
+            if msg["header"]["msg_type"] == "stream" and msg["content"]["name"] == "stdout":
+                stdout_chunks.append(msg["content"]["text"])
+
+    # check outputs
+    # use sets of lines to ignore ordering issues with
+    # async flush and watchfd thread
+
+    # Check the stream output forwarded over zmq
+    zmq_stdout = "".join(stdout_chunks)
+    assert set(zmq_stdout.strip().splitlines()) == {
+        "fd",
+        "print",
+        # original stdout streams don't get captured,
+        # they write directly to the terminal
+        # "stdout",
+        # "__stdout__",
+    }
+
+    # Check what was written to the process stdout (kernel terminal)
+    # just check that each output source went to the terminal
+    assert set(p.stdout.strip().splitlines()) == {
+        "fd",
+        "print",
         "stdout",
-        isatty=True,
-        echo=sys.stdout,
-    )
-    save_stdout = sys.stdout
-    with stream, mock.patch.object(sys, "stdout", stream):
-        # write to low-level FD
-        os.write(stdout_fd, b"fd\n")
-        os.fsync(stdout_fd)
-        # write to unwrapped __stdout__ (should also go to original FD)
-        sys.__stdout__.write("__stdout__\n")
-        sys.__stdout__.flush()
-        # write to original sys.stdout (should be the same as __stdout__)
-        save_stdout.write("stdout\n")
-        save_stdout.flush()
-        # print (writes to stream)
-        print("print")
-        sys.stdout.flush()
-
-    out, err = capfd.readouterr()
-    print(out, err)
-    assert out.strip() == "fd\n__stdout__\nstdout\nprint"
+        "__stdout__",
+    }