fix: handle file stdout/stderr (#186)

* fix: handle file stdout/stderr

* fix: handle stdin=subprocess.PIPE

Allows tests to write to the standard input of a mocked process

* test: fix failing test

- ignore `PytestDeprecationWarning` in `pytest.ini`.

- suppress `docutils` errors in `tests/test_examples.py` by setting `report_level` to 5.

* Test/fix windows test (#2)

* test: convert to lowercase in whoami test

* test: use a more cross platform command

or else you'd get  FileNotFoundError errors

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix/mac test (#3)

* temp: test on push

* fix: increase timeout for mac test

* fix: ignore warning in universal newlines test

* fix: add another filterwarning tag

* remove: temp on push test

* Add changelog, minor refactor in tests.

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: aklajnert <github@aklajnert.pl>

Author: 3 people

.gitignore

@@ -78,3 +78,6 @@ target/
 # pyenv
 .python-version
 
+# Visual Studio Code
+.vs/
+

README.rst

@@ -129,17 +129,18 @@ processes with real ``subprocess``, or use
 .. code-block:: python
 
     def test_real_process(fp):
+        command = ["python", "-c", "pass"]
         with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
-            # this will fail, as "ls" command is not registered
-            subprocess.call("ls")
+            # this will fail, as the command is not registered
+            subprocess.call(command)
 
-        fp.pass_command("ls")
+        fp.pass_command(command)
         # now it should be fine
-        assert subprocess.call("ls") == 0
+        assert subprocess.call(command) == 0
 
         # allow all commands to be called by real subprocess
         fp.allow_unregistered(True)
-        assert subprocess.call(["ls", "-l"]) == 0
+        assert subprocess.call(command) == 0
 
 
 Differing results
@@ -272,12 +273,12 @@ if the subprocess command will be called with a string argument.
 
     def test_non_exact_matching(fp):
         # define a command that will take any number of arguments
-        fp.register(["ls", fp.any()])
-        assert subprocess.check_call("ls -lah") == 0
+        fp.register(["python", fp.any()])
+        assert subprocess.check_call(["python", "-c", "pass"]) == 0
 
         # `fake_subprocess.any()` is OK even with no arguments
-        fp.register(["ls", fp.any()])
-        assert subprocess.check_call("ls") == 0
+        fp.register(["python", fp.any()])
+        assert subprocess.check_call(["python"]) == 0
 
         # but it can force a minimum amount of arguments
         fp.register(["cp", fp.any(min=2)])
@@ -310,8 +311,8 @@ the same name, regardless of the location. This is accomplished with
 
     def test_any_matching_program(fp):
         # define a command that can come from anywhere
-        fp.register([fp.program("ls")])
-        assert subprocess.check_call("/bin/ls") == 0
+        fp.register([fp.program("python")])
+        assert subprocess.check_call(sys.executable) == 0
 
 
 Check if process was called

changelog.d/bug.af83d745.entry.yaml

@@ -0,0 +1,5 @@
+message: Support file handles in stdout and stderr.
+pr_ids:
+- '186'
+timestamp: 1756146621
+type: bug

pytest.ini

@@ -1,5 +1,7 @@
 [pytest]
 junit_family=legacy
+asyncio_default_fixture_loop_scope = function
 filterwarnings =
     error
     ignore::pytest.PytestUnraisableExceptionWarning
+    ignore::pytest.PytestDeprecationWarning

pytest_subprocess/fake_popen.py

@@ -39,6 +39,7 @@ class FakePopen:
 
     stdout: Optional[BUFFER] = None
     stderr: Optional[BUFFER] = None
+    stdin: Optional[BUFFER] = None
     returncode: Optional[int] = None
     text_mode: bool = False
     pid: int = 0
@@ -168,6 +169,11 @@ def configure(self, **kwargs: Optional[Dict]) -> None:
         """Setup the FakePopen instance based on a real Popen arguments."""
         self.__kwargs = self.safe_copy(kwargs)
         self.__universal_newlines = kwargs.get("universal_newlines", None)
+
+        stdin = kwargs.get("stdin")
+        if stdin == subprocess.PIPE:
+            self.stdin = self._get_empty_buffer(False)
+
         text = kwargs.get("text", None)
         encoding = kwargs.get("encoding", None)
         errors = kwargs.get("errors", None)
@@ -288,7 +294,7 @@ def _write_to_buffer(self, data: OPTIONAL_TEXT_OR_ITERABLE, buffer: IO) -> None:
         )
         if isinstance(data, (list, tuple)):
             buffer.writelines([data_type(line + "\n") for line in data])
-        else:
+        elif data is not None:
             buffer.write(data_type(data))
 
     def _convert(self, input: Union[str, bytes]) -> Union[str, bytes]:

tests/example_script.py

@@ -9,7 +9,7 @@
     print("Stderr line 1", file=sys.stderr)
 
 if "wait" in sys.argv:
-    time.sleep(0.5)
+    time.sleep(1)
 
 if "non-zero" in sys.argv:
     sys.exit(1)

tests/test_examples.py

@@ -15,7 +15,9 @@ def get_code_blocks(file_path):
     with file_path.open() as file_handle:
         content = file_handle.read()
 
-    code_blocks = publish_doctree(content).findall(condition=is_code_block)
+    code_blocks = publish_doctree(
+        content, settings_overrides={"report_level": 5}
+    ).findall(condition=is_code_block)
     return [block.astext() for block in code_blocks]
 
 

tests/test_subprocess.py

@@ -7,6 +7,7 @@
 import sys
 import time
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 
 import pytest
 
@@ -347,6 +348,7 @@ def test_run(fp, fake):
     assert process.stderr is None
 
 
+@pytest.mark.filterwarnings("ignore:unclosed file:ResourceWarning")
 @pytest.mark.parametrize("fake", [False, True])
 def test_universal_newlines(fp, fake):
     fp.allow_unregistered(not fake)
@@ -363,6 +365,7 @@ def test_universal_newlines(fp, fake):
     assert process.stdout.read() == "Stdout line 1\nStdout line 2\n"
 
 
+@pytest.mark.filterwarnings("ignore:unclosed file:ResourceWarning")
 @pytest.mark.parametrize("fake", [False, True])
 def test_text(fp, fake):
     fp.allow_unregistered(not fake)
@@ -922,12 +925,14 @@ def test_encoding(fp, fake, argument):
     if fake:
         fp.register(["whoami"], stdout=username)
 
-    output = subprocess.check_output(
-        ["whoami"], **{argument: values.get(argument)}
-    ).strip()
+    output = (
+        subprocess.check_output(["whoami"], **{argument: values.get(argument)})
+        .strip()
+        .lower()
+    )
 
     assert isinstance(output, str)
-    assert output.endswith(username)
+    assert output.endswith(username.lower())
 
 
 @pytest.mark.parametrize("command", ["ls -lah", ["ls", "-lah"]])
@@ -1280,3 +1285,73 @@ def spawn_process() -> subprocess.Popen[str]:
     proc.wait()
 
     assert proc.stdout.read() == "Stdout line 1\nStdout line 2\n"
+
+
+def test_stdin_pipe(fp):
+    """
+    Test that stdin is a writable buffer when using subprocess.PIPE.
+    """
+    fp.register(["my-command"])
+
+    process = subprocess.Popen(
+        ["my-command"],
+        stdin=subprocess.PIPE,
+    )
+
+    assert process.stdin is not None
+    assert process.stdin.writable()
+
+    # We can write to the buffer.
+    process.stdin.write(b"some data")
+    process.stdin.flush()
+
+    # The data can be read back from the buffer for inspection.
+    process.stdin.seek(0)
+    assert process.stdin.read() == b"some data"
+
+    # After closing, it should raise a ValueError.
+    process.stdin.close()
+    with pytest.raises(ValueError):
+        process.stdin.write(b"more data")
+
+
+def test_stdout_stderr_as_file_bug(fp):
+    """
+    Test that no TypeError is raised when stdout/stderr is a file
+    and the stream is not registered.
+
+    From GitHub #144
+    """
+    # register process with stdout but no stderr
+    fp.register(
+        ["test-no-stderr"],
+        stdout="test",
+    )
+    # register process with stderr but no stdout
+    fp.register(
+        ["test-no-stdout"],
+        stderr="test",
+    )
+    # register process with no streams
+    fp.register(
+        ["test-no-streams"],
+    )
+
+    with NamedTemporaryFile("wb") as temp_file:
+        # test with stderr not registered
+        process = subprocess.Popen(
+            "test-no-stderr", stdout=temp_file.file, stderr=temp_file.file
+        )
+        process.wait()
+
+        # test with stdout not registered
+        process = subprocess.Popen(
+            "test-no-stdout", stdout=temp_file.file, stderr=temp_file.file
+        )
+        process.wait()
+
+        # test with no streams registered
+        process = subprocess.Popen(
+            "test-no-streams", stdout=temp_file.file, stderr=temp_file.file
+        )
+        process.wait()