Skip to content

Commit

Permalink
Change env property to kwargs with all Popen arguments.
Browse files Browse the repository at this point in the history
  • Loading branch information
aklajnert committed Jan 1, 2025
1 parent 038b35d commit adffd9e
Show file tree
Hide file tree
Showing 4 changed files with 69 additions and 11 deletions.
34 changes: 34 additions & 0 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,40 @@ how many a command has been called. The latter supports ``fp.any()``.
assert fp.call_count(["cp", fp.any()]) == 3
Check Popen arguments
---------------------

You can use the recorded calls functionality to introspect the keyword
arguments that were passed to `Popen`.

.. code-block:: python
def test_process_recorder_kwargs(fp):
fp.keep_last_process(True)
recorder = fp.register(["test_script", fp.any()])
subprocess.run(
("test_script", "arg1"), env={"foo": "bar"}, cwd="/home/user"
)
subprocess.Popen(
["test_script", "arg2"],
env={"foo": "bar1"},
executable="test_script",
shell=True,
)
assert recorder.calls[0].args == ("test_script", "arg1")
assert recorder.calls[0].kwargs == {
"cwd": "/home/user",
"env": {"foo": "bar"},
}
assert recorder.calls[1].args == ["test_script", "arg2"]
assert recorder.calls[1].kwargs == {
"env": {"foo": "bar1"},
"executable": "test_script",
"shell": True,
}
Handling signals
----------------

Expand Down
3 changes: 2 additions & 1 deletion changelog.d/feature.8899db6c.entry.yaml
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
message: Add env property to FakePopen.
message: Allow to access keyword arguments passed to Popen.
pr_ids:
- '171'
- '178'
timestamp: 1728114595
type: feature
19 changes: 15 additions & 4 deletions pytest_subprocess/fake_popen.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import asyncio
import collections.abc
import concurrent.futures
import copy
import io
import os
import signal
Expand Down Expand Up @@ -68,7 +69,7 @@ def __init__(
msg = f"argument of type {arg.__class__.__name__!r} is not iterable"
raise TypeError(msg)
self.args = command
self.__env: Optional[Dict[str, AnyType]] = None
self.__kwargs: Optional[Dict[str, AnyType]] = None
self.__stdout: OPTIONAL_TEXT_OR_ITERABLE = stdout
self.__stderr: OPTIONAL_TEXT_OR_ITERABLE = stderr
self.__thread: Optional[Thread] = None
Expand All @@ -82,8 +83,8 @@ def __init__(
self._callback_kwargs: Optional[Dict[str, AnyType]] = callback_kwargs

@property
def env(self) -> Optional[Dict[str, AnyType]]:
return self.__env
def kwargs(self) -> Optional[Dict[str, AnyType]]:
return self.__kwargs

def __enter__(self) -> "FakePopen":
return self
Expand Down Expand Up @@ -164,8 +165,8 @@ def kill(self) -> None:

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)
self.__env = kwargs.get("env")
text = kwargs.get("text", None)
encoding = kwargs.get("encoding", None)
errors = kwargs.get("errors", None)
Expand Down Expand Up @@ -201,6 +202,16 @@ def configure(self, **kwargs: Optional[Dict]) -> None:
elif isinstance(stderr, (io.BufferedWriter, io.TextIOWrapper)):
self._write_to_buffer(self.__stderr, stderr)

@staticmethod
def safe_copy(kwargs):
"""
Deepcopy can fail if the value is not serializable, fallback to shallow copy.
"""
try:
return copy.deepcopy(kwargs)
except TypeError:
return dict(**kwargs)

def _prepare_buffer(
self,
input: OPTIONAL_TEXT_OR_ITERABLE,
Expand Down
24 changes: 18 additions & 6 deletions tests/test_subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -1234,18 +1234,30 @@ def test_process_recorder(fp):
assert not recorder.was_called()


def test_process_recorder_env(fp):
def test_process_recorder_kwargs(fp):
fp.keep_last_process(True)
recorder = fp.register(["test_script", fp.any()])

subprocess.call(("test_script", "arg1"))
subprocess.run(("test_script", "arg2"), env={"foo": "bar"})
subprocess.Popen(["test_script", "arg3"], env={"foo": "bar1"})
subprocess.run(("test_script", "arg2"), env={"foo": "bar"}, cwd="/home/user")
subprocess.Popen(
["test_script", "arg3"],
env={"foo": "bar1"},
executable="test_script",
shell=True,
)

assert recorder.call_count() == 3
assert recorder.calls[0].env is None
assert recorder.calls[1].env.get("foo") == "bar"
assert recorder.calls[2].env.get("foo") == "bar1"
assert recorder.calls[0].args == ("test_script", "arg1")
assert recorder.calls[0].kwargs == {}
assert recorder.calls[1].args == ("test_script", "arg2")
assert recorder.calls[1].kwargs == {"cwd": "/home/user", "env": {"foo": "bar"}}
assert recorder.calls[2].args == ["test_script", "arg3"]
assert recorder.calls[2].kwargs == {
"env": {"foo": "bar1"},
"executable": "test_script",
"shell": True,
}


def test_fake_popen_is_typed(fp):
Expand Down

0 comments on commit adffd9e

Please sign in to comment.