Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pypeline.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ pipeline:
module: pypeline.steps.create_venv
config:
package_manager: uv>=0
python_version: 3.13
python_version: "3.13"
- step: PreCommit
run: pre-commit run --all-files
- step: PyTest
Expand Down
8 changes: 6 additions & 2 deletions src/py_app_dev/core/subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,10 @@ class SubprocessExecutor:
"""
Execute a command in a subprocess.

stderr is merged into stdout, so the progress lines and the error lines of the command stay in
chronological order. The returned CompletedProcess therefore has stderr set to None ("not
captured", as in the standard library). Read all output from stdout.

Args:
----
capture_output: If True, the output of the command will be captured.
Expand Down Expand Up @@ -96,11 +100,11 @@ def _finalize_process(
pipe.close()

def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any] | None:
"""Execute the command and return the CompletedProcess object if handle_errors is False."""
"""Execute the command and return the CompletedProcess object if handle_errors is False (stdout holds the merged output, stderr is always None)."""
start_time = time.monotonic()
completed_process: subprocess.CompletedProcess[Any] | None = None
stdout = ""
stderr = ""
stderr: str | None = None
self.logger.info(f"Running command: {self.command_str}")
cwd_path = (self.current_working_directory or Path.cwd()).as_posix()
process: subprocess.Popen[str] | None = None
Expand Down
19 changes: 19 additions & 0 deletions tests/test_subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,25 @@ def test_command_execution_scenarios(command: list[str | Path], exp_stdout: str,
assert result.returncode == exp_returncode


@pytest.mark.parametrize(
"capture_output, print_output, exp_stdout",
[
(True, True, "boom"),
(True, False, "boom"),
(False, True, ""),
],
)
def test_stderr_is_none_because_it_is_merged_into_stdout(capture_output: bool, print_output: bool, exp_stdout: str) -> None:
result = SubprocessExecutor(
["python", "-c", "import sys; sys.stderr.write('boom')"],
capture_output=capture_output,
print_output=print_output,
).execute(handle_errors=False)
assert result is not None
assert result.stdout == exp_stdout
assert result.stderr is None


@pytest.mark.skipif(platform.system() != "Windows", reason="Junction creation test is Windows-specific")
def test_junction_creation(tmp_path: Path) -> None:
test_path = tmp_path / "test"
Expand Down
Loading