Fix read_line to raise EOFError if nothing was read (#86)

Author: sk1p

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
 

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: ...

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)