debug harder...

Author: A5rocks

ci.sh

@@ -9,6 +9,7 @@ export PYRIGHT_PYTHON_IGNORE_WARNINGS=1
 # Log some general info about the environment
 echo "::group::Environment"
 uname -a
+uname -r
 env | sort
 echo "::endgroup::"
 
@@ -124,6 +125,8 @@ perl -i -pe 's/-p trio\._tests\.pytest_plugin//' "$INSTALLDIR/pyproject.toml"
 
 echo "::endgroup::"
 echo "::group:: Run Tests"
+python test.py
+
 if PYTHONPATH=../tests COVERAGE_PROCESS_START=$(pwd)/../pyproject.toml \
         coverage run --rcfile=../pyproject.toml -m \
         pytest -s -ra --junitxml=../test-results.xml \

test.py

@@ -0,0 +1,63 @@
+import os
+import signal
+import sys
+import time
+
+
+def test_ki_newline_injection() -> None:
+    assert sys.platform != "win32"
+
+    import pty
+
+    # NOTE: this cannot be subprocess.Popen because pty.fork
+    #       does some magic to set the controlling terminal.
+    # (which I don't know how to replicate... so I copied this
+    # structure from pty.spawn...)
+    pid, pty_fd = pty.fork()  # type: ignore[attr-defined,unused-ignore]
+    if pid == 0:
+        os.execlp(sys.executable, *[sys.executable, "-u", "-m", "trio"])
+
+    # setup:
+    buffer = b""
+    while not buffer.endswith(b"import trio\r\n>>> "):
+        buffer += os.read(pty_fd, 4096)
+
+    # sanity check:
+    print("AAAAAAAAAAAAAAAAAA")
+    print(buffer.decode())
+    buffer = b""
+    os.write(pty_fd, b'print("hello!")\n')
+    while not buffer.endswith(b">>> "):
+        buffer += os.read(pty_fd, 4096)
+
+    assert buffer.count(b"hello!") == 2
+
+    # press ctrl+c
+    time.sleep(5)  # hypothesis: things are slow...
+    print("BBBBBBBBBBBBBBBB")
+    print(buffer.decode())
+    buffer = b""
+    os.kill(pid, signal.SIGINT)
+    while not buffer.endswith(b">>> "):
+        buffer += os.read(pty_fd, 4096)
+        print(5, buffer)
+
+    assert b"KeyboardInterrupt" in buffer
+
+    # press ctrl+c later
+    print("CCCCCCCCCCCCCCCC")
+    print(buffer.decode())
+    buffer = b""
+    os.write(pty_fd, b'print("hello!")')
+    os.kill(pid, signal.SIGINT)
+    while not buffer.endswith(b">>> "):
+        buffer += os.read(pty_fd, 4096)
+
+    assert b"KeyboardInterrupt" in buffer
+    print("DDDDDDDDDDDDDDDDDDD")
+    print(buffer.decode())
+    os.close(pty_fd)
+    os.waitpid(pid, 0)[1]
+
+
+test_ki_newline_injection()