Add extensive tests to read from pipes

rhpvorderman · rhpvorderman · commit a6f4b9eddea6 · 2024-03-27T16:50:34.000+01:00
diff --git a/src/xopen/__init__.py b/src/xopen/__init__.py
@@ -721,7 +721,11 @@ def _filepath_from_path_or_filelike(fileorpath: FileOrPath) -> str:
     except TypeError:
         pass
     if hasattr(fileorpath, "name"):
-        return fileorpath.name
+        name = fileorpath.name
+        if isinstance(name, str):
+            return name
+        elif isinstance(name, bytes):
+            return name.decode()
     return ""
 
 
@@ -831,7 +835,7 @@ def xopen(  # noqa: C901
     if filename == "-":
         filename = _open_stdin_or_out(binary_mode)
     elif _file_is_a_socket_or_pipe(filename):
-        filename = open(filename, binary_mode)
+        filename = open(filename, binary_mode)  # type: ignore
 
     if format not in (None, "gz", "xz", "bz2", "zst"):
         raise ValueError(
diff --git a/tests/test_xopen.py b/tests/test_xopen.py
@@ -2,6 +2,7 @@
 Tests for the xopen.xopen function
 """
 import bz2
+import subprocess
 import sys
 import tempfile
 from contextlib import contextmanager
@@ -656,3 +657,32 @@ def test_xopen_stdout(monkeypatch):
         raw.seek(0)
         data = raw.read()
     assert data == "Hello world!"
+
+
+@pytest.mark.parametrize("threads", (0, 1))
+def test_xopen_read_from_pipe(ext, threads):
+    if ext == ".zst" and zstandard is None:
+        return
+    in_file = TEST_DIR / f"file.txt{ext}"
+    process = subprocess.Popen(("cat", str(in_file)), stdout=subprocess.PIPE)
+    with xopen(process.stdout, "rt", threads=threads) as f:
+        data = f.read()
+    process.wait()
+    assert data == CONTENT
+
+
+@pytest.mark.parametrize("threads", (0, 1))
+def test_xopen_write_to_pipe(threads, ext):
+    if ext == ".zst" and zstandard is None:
+        return
+    format = ext.lstrip(".")
+    if format == "":
+        format = None
+    process = subprocess.Popen(("cat",), stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+    with xopen(process.stdin, "wt", threads=threads, format=format) as f:
+        f.write(CONTENT)
+    process.stdin.close()
+    with xopen(process.stdout, "rt", threads=threads) as f:
+        data = f.read()
+    process.wait()
+    assert data == CONTENT
