devedeng-extended/tests/test_executor_io.py

"""Tests for the hardened subprocess I/O in executor.py.

The executor drains child stdout/stderr with background threads (not the GLib
main loop), so a stalled/wedged GUI can never deadlock a child on a full pipe.
"""

import time
import os
import tempfile

import gi
gi.require_version("Gtk", "3.0")
from gi.repository import GLib

import devedeng.executor as ex


class CaptureExecutor(ex.executor):
    """Executor subclass that records the lines handed to process_*."""

    def __init__(self):
        super().__init__()
        self.out_lines = []
        self.err_lines = []

    def process_stdout(self, data):
        self.out_lines += data

    def process_stderr(self, data):
        self.err_lines += data


def _run_to_end(executor, timeout_s=20):
    """Drive a GLib main loop until the executor emits 'ended' (or times out).
    Returns the process return value, or raises on timeout (a hang)."""
    loop = GLib.MainLoop()
    result = {}

    def on_ended(_obj, ret):
        result["ret"] = ret
        loop.quit()

    executor.connect("ended", on_ended)
    executor.launch_process(executor.command_var)
    GLib.timeout_add(int(timeout_s * 1000),
                     lambda: (result.setdefault("timeout", True), loop.quit())[1])
    loop.run()
    assert not result.get("timeout"), "executor hung (no 'ended' within timeout)"
    return result["ret"]


def test_line_parsing_stdout_and_stderr_with_cr():
    e = CaptureExecutor()
    e.command_var = ["python3", "-c",
                     "import sys\n"
                     "sys.stdout.write('o1\\no2\\n'); sys.stdout.flush()\n"
                     "sys.stderr.write('p 1\\rp 2\\rdone\\n'); sys.stderr.flush()\n"]
    ret = _run_to_end(e)
    assert ret == 0
    assert e.out_lines == ["o1", "o2"]
    # carriage-return progress lines are split like newlines
    assert e.err_lines == ["p 1", "p 2", "done"]


def test_large_output_no_deadlock_no_loss():
    # ~2 MB of stderr, far beyond the ~64 KB pipe buffer
    e = CaptureExecutor()
    e.command_var = ["python3", "-c",
                     "import sys\n"
                     "for i in range(20000): sys.stderr.write('e%d\\n'%i)\n"
                     "for i in range(10000): sys.stdout.write('o%d\\n'%i)\n"]
    ret = _run_to_end(e)
    assert ret == 0
    assert len(e.err_lines) == 20000   # nothing lost
    assert len(e.out_lines) == 10000


def test_completes_with_no_main_loop_running():
    # the wedged-GUI scenario: never service the GLib loop; the reader threads
    # must still drain the pipe so the child can finish instead of deadlocking
    e = CaptureExecutor()
    e.command_var = ["python3", "-c",
                     "import sys\n"
                     "for i in range(20000): sys.stderr.write('x'*100+'\\n')\n"]
    e.launch_process(e.command_var)
    deadline = time.time() + 15
    while time.time() < deadline:
        if e.handle is None or e.handle.poll() is not None:
            break
        time.sleep(0.05)
    rc = "reaped" if e.handle is None else e.handle.poll()
    assert rc is not None, "child deadlocked with no draining main loop"


def test_cancel_kills_and_ends():
    e = CaptureExecutor()
    e.command_var = ["python3", "-c", "import time; time.sleep(60)"]
    loop = GLib.MainLoop()
    result = {}
    e.connect("ended", lambda o, r: (result.update(ret=r), loop.quit()))
    e.launch_process(e.command_var)
    # cancel shortly after start
    GLib.timeout_add(300, lambda: (e.cancel(), False)[1])
    GLib.timeout_add(15000, lambda: (result.setdefault("timeout", True),
                                     loop.quit())[1])
    loop.run()
    assert not result.get("timeout"), "cancel did not end the process"
    # a killed process reports success (retval forced to 0 by design)
    assert result["ret"] == 0
    assert e.killed is True


def test_stdout_redirected_to_file():
    tmp = tempfile.mktemp(suffix=".out")
    e = CaptureExecutor()
    e.stdout_file = tmp
    e.command_var = ["python3", "-c",
                     "import sys\n"
                     "sys.stdout.write('FILE CONTENT\\n')\n"
                     "sys.stderr.write('status line\\n')\n"]
    try:
        ret = _run_to_end(e)
        assert ret == 0
        with open(tmp) as f:
            assert "FILE CONTENT" in f.read()
        # stderr is still parsed while stdout goes to the file
        assert "status line" in e.err_lines
        # stdout was not also dispatched as lines
        assert e.out_lines == []
    finally:
        if os.path.exists(tmp):
            os.remove(tmp)


def test_nonzero_exit_code_propagates():
    e = CaptureExecutor()
    e.command_var = ["python3", "-c", "import sys; sys.exit(3)"]
    ret = _run_to_end(e)
    assert ret == 3