add non-functional validation

refactors

validation/raptor_validation/subprocess_utils.py

@@ -0,0 +1,119 @@
+import errno
+import os
+import pty
+import selectors
+import subprocess
+import time
+
+MAX_ERROR_OUTPUT_BYTES = 8192
+
+
+def _read_chunk(fd, treat_eio_as_eof=False):
+    try:
+        return os.read(fd, 4096)
+    except OSError as exc:
+        if treat_eio_as_eof and exc.errno == errno.EIO:
+            return b""
+        raise
+
+
+def _stream_output(fd, process, reporter, treat_eio_as_eof=False, stream_output=True, timeout_sec=None):
+    selector = selectors.DefaultSelector()
+    recent_output = bytearray()
+    captured_output = bytearray()
+    deadline = None if timeout_sec is None else time.monotonic() + timeout_sec
+
+    try:
+        selector.register(fd, selectors.EVENT_READ)
+
+        while selector.get_map():
+            select_timeout = None
+            if deadline is not None:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    process.kill()
+                    process.wait()
+                    raise subprocess.TimeoutExpired(process.args, timeout_sec, output=bytes(captured_output))
+                select_timeout = min(1.0, remaining)
+
+            for key, _ in selector.select(select_timeout):
+                data = _read_chunk(key.fileobj, treat_eio_as_eof=treat_eio_as_eof)
+                if not data:
+                    selector.unregister(key.fileobj)
+                    os.close(key.fileobj)
+                    continue
+
+                if stream_output:
+                    reporter._clear()
+                    os.write(1, data)
+                    reporter._render()
+                captured_output.extend(data)
+                if stream_output:
+                    recent_output.extend(data)
+                    if len(recent_output) > MAX_ERROR_OUTPUT_BYTES:
+                        del recent_output[:-MAX_ERROR_OUTPUT_BYTES]
+    finally:
+        selector.close()
+
+    return_code = process.wait()
+    if return_code != 0:
+        error_output = captured_output if not stream_output else recent_output
+        exc = subprocess.CalledProcessError(return_code, process.args, output=bytes(error_output))
+        exc.output_already_streamed = stream_output and bool(captured_output)
+        raise exc
+    return bytes(captured_output)
+
+
+def run_command_with_reporter(cmd, cwd=None, reporter=None, capture_output=False, timeout_sec=None):
+    if reporter is None:
+        if capture_output:
+            completed = subprocess.run(
+                cmd,
+                cwd=cwd,
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                timeout=timeout_sec,
+            )
+            return completed.stdout.decode("utf-8", errors="replace")
+        subprocess.run(cmd, cwd=cwd, check=True, timeout=timeout_sec)
+        return None
+
+    stream_output = bool(getattr(reporter, "verbose", False))
+    if not stream_output:
+        completed = subprocess.run(
+            cmd,
+            cwd=cwd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            timeout=timeout_sec,
+        )
+        if completed.returncode != 0:
+            raise subprocess.CalledProcessError(completed.returncode, completed.args, output=completed.stdout)
+        return completed.stdout.decode("utf-8", errors="replace") if capture_output else None
+
+    try:
+        master_fd, slave_fd = pty.openpty()
+    except OSError:
+        process = subprocess.Popen(
+            cmd,
+            cwd=cwd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
+        assert process.stdout is not None
+        output = _stream_output(process.stdout.fileno(), process, reporter, timeout_sec=timeout_sec)
+        return output.decode("utf-8", errors="replace") if capture_output else None
+
+    try:
+        process = subprocess.Popen(
+            cmd,
+            cwd=cwd,
+            stdout=slave_fd,
+            stderr=slave_fd,
+        )
+    finally:
+        os.close(slave_fd)
+
+    output = _stream_output(master_fd, process, reporter, treat_eio_as_eof=True, timeout_sec=timeout_sec)
+    return output.decode("utf-8", errors="replace") if capture_output else None