Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Change env property to kwargs with all Popen arguments. #178

Merged
merged 1 commit into from
Jan 1, 2025
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
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: Dict[str, AnyType]) -> Dict[str, AnyType]:
"""
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
5 changes: 3 additions & 2 deletions tests/test_asyncio.py
Original file line number Diff line number Diff line change
Expand Up @@ -383,7 +383,7 @@ async def my_async_func():


@pytest.mark.asyncio
async def test_process_recorder_env(fp):
async def test_process_recorder_args(fp):
fp.keep_last_process(True)
recorder = fp.register(["test_script", fp.any()])
await asyncio.create_subprocess_exec(
Expand All @@ -393,7 +393,8 @@ async def test_process_recorder_env(fp):
)

assert recorder.call_count() == 1
assert recorder.calls[0].env.get("foo") == "bar"
assert recorder.calls[0].args == ["test_script", "arg1"]
assert recorder.calls[0].kwargs == {"env": {"foo": "bar"}}


@pytest.fixture(autouse=True)
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_args(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
Loading