Throw OSError when unsupported mode is used

rhpvorderman · rhpvorderman · commit 865cb983c77d · 2024-02-20T09:30:40.000+01:00
diff --git a/src/xopen/__init__.py b/src/xopen/__init__.py
@@ -192,8 +192,9 @@ def __init__(  # noqa: C901
         if isinstance(self._infile, io.IOBase):
             # not supported (yet) for reading.
             if "r" in mode:
-                raise ValueError(
-                    "File object not supported for reading through Piped Program."
+                raise OSError(
+                    f"File object not supported for reading through "
+                    f"{self.__class__.__name__}."
                 )
             path = path.name
         if (
diff --git a/tests/test_piped.py b/tests/test_piped.py
@@ -378,3 +378,12 @@ def test_piped_tool_fails_on_close(tmp_path):
         ) as f:
             f.write(b"Hello")
     assert "exit code 5" in e.value.args[0]
+
+
+def test_piped_tool_fails_read_mode_with_filehandle(tmp_path):
+    test_file = tmp_path / "test.txt.gz"
+    test_file.write_bytes(gzip.compress(b"test"))
+    with open(test_file, "rb") as filehandle:
+        with pytest.raises(OSError) as error:
+            _PipedCompressionProgram(filehandle, "rb")
+        error.match("File object not supported for reading")
diff --git a/tests/test_xopen.py b/tests/test_xopen.py
@@ -578,24 +578,6 @@ def test_pass_file_object_for_reading_no_threads(ext):
             assert f.readline() == CONTENT_LINES[0].encode("utf-8")
 
 
-@pytest.mark.parametrize("ext", extensions)
-def test_pass_file_object_for_reading(ext):
-    if ext == ".zst" and zstandard is None:
-        return
-    # non-compressed file do not raise error
-    if not ext:
-        return
-    # todo : disable the gz modules to force piped compression.
-    with open(TEST_DIR / f"file.txt{ext}", "rb") as fh:
-        with pytest.raises(ValueError) as e:
-            with xopen(fh, mode="rb", threads=1) as f:
-                f.readline()  # pragma: no cover
-        assert (
-            "File object not supported for reading through Piped Program"
-            in e.value.args[0]
-        )
-
-
 @pytest.mark.parametrize("ext", extensions)
 def test_pass_file_object_for_writing(tmp_path, ext):
     if ext == ".zst" and zstandard is None:
