diff --git a/.gitignore b/.gitignore index 23e0bec..a5b6080 100644 --- a/.gitignore +++ b/.gitignore @@ -78,3 +78,6 @@ target/ # pyenv .python-version +# Visual Studio Code +.vs/ + diff --git a/README.rst b/README.rst index 597de28..c023d44 100644 --- a/README.rst +++ b/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 diff --git a/changelog.d/bug.af83d745.entry.yaml b/changelog.d/bug.af83d745.entry.yaml new file mode 100644 index 0000000..85fd27a --- /dev/null +++ b/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 diff --git a/pytest.ini b/pytest.ini index 306cf5d..0382fa7 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,5 +1,7 @@ [pytest] junit_family=legacy +asyncio_default_fixture_loop_scope = function filterwarnings = error ignore::pytest.PytestUnraisableExceptionWarning + ignore::pytest.PytestDeprecationWarning diff --git a/pytest_subprocess/fake_popen.py b/pytest_subprocess/fake_popen.py index 5188634..5f6ed23 100644 --- a/pytest_subprocess/fake_popen.py +++ b/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]: diff --git a/tests/example_script.py b/tests/example_script.py index 4d5f1e5..88040a1 100644 --- a/tests/example_script.py +++ b/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) diff --git a/tests/test_examples.py b/tests/test_examples.py index df849f3..8969406 100644 --- a/tests/test_examples.py +++ b/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] diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py index 3b38bea..289dc62 100644 --- a/tests/test_subprocess.py +++ b/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()