diff --git a/src/pyproject_api/_backend.py b/src/pyproject_api/_backend.py index f0be631..0583d12 100644 --- a/src/pyproject_api/_backend.py +++ b/src/pyproject_api/_backend.py @@ -112,17 +112,18 @@ def run(argv): return 0 -def read_line(): +def read_line(fd=0): # for some reason input() seems to break (hangs forever) so instead we read byte by byte the unbuffered stream content = bytearray() while True: - try: - char = os.read(0, 1) - except EOFError: # pragma: no cover # when the stdout is closed without exit - break # pragma: no cover - if char == b"\n": # pragma: no cover + char = os.read(fd, 1) + if not char: + if not content: + raise EOFError("EOF without reading anything") # we didn't get a line at all, let the caller know + break + if char == b"\n": break - if char != b"\r": # pragma: win32 cover + if char != b"\r": content += char return content diff --git a/src/pyproject_api/_backend.pyi b/src/pyproject_api/_backend.pyi index 45f3b75..1bf7ae6 100644 --- a/src/pyproject_api/_backend.pyi +++ b/src/pyproject_api/_backend.pyi @@ -12,5 +12,5 @@ class BackendProxy: def _optional_commands(self) -> dict[str, bool]: ... def run(argv: Sequence[str]) -> int: ... -def read_line() -> bytearray: ... +def read_line(fd: int = 0) -> bytearray: ... def flush() -> None: ... diff --git a/tests/test_backend.py b/tests/test_backend.py index 58cb731..f4d9e46 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -1,11 +1,12 @@ from __future__ import annotations import json +import os from typing import TYPE_CHECKING, Any import pytest -from pyproject_api._backend import BackendProxy, run +from pyproject_api._backend import BackendProxy, read_line, run if TYPE_CHECKING: from pathlib import Path @@ -127,3 +128,37 @@ def fake_backend(name: str, *args: Any, **kwargs: Any) -> Any: # noqa: ARG001 assert "Backend: run command dummy_command_b with args {'baz': 'qux'}" in captured.out assert "Backend: run command dummy_command_c with args {'win': 'wow'}" in captured.out assert "SystemExit: 2" in captured.err + + +def test_read_line_success() -> None: + r, w = os.pipe() + try: + line_in = b"this is a line\r\n" + os.write(w, line_in) + line_out = read_line(fd=r) + assert line_out == bytearray(b"this is a line") + finally: + os.close(r) + os.close(w) + + +def test_read_line_eof_before_newline() -> None: + r, w = os.pipe() + try: + line_in = b"this is a line" + os.write(w, line_in) + os.close(w) + line_out = read_line(fd=r) + assert line_out == bytearray(b"this is a line") + finally: + os.close(r) + + +def test_read_line_eof_at_the_beginning() -> None: + r, w = os.pipe() + try: + os.close(w) + with pytest.raises(EOFError): + read_line(fd=r) + finally: + os.close(r)