From adffd9e34815b52d7210abe6c55ab79b79277bd5 Mon Sep 17 00:00:00 2001
From: Andrzej Klajnert <github@aklajnert.pl>
Date: Wed, 1 Jan 2025 13:08:53 +0100
Subject: [PATCH] Change `env` property to `kwargs` with all Popen arguments.

---
 README.rst                              | 34 +++++++++++++++++++++++++
 changelog.d/feature.8899db6c.entry.yaml |  3 ++-
 pytest_subprocess/fake_popen.py         | 19 +++++++++++---
 tests/test_subprocess.py                | 24 ++++++++++++-----
 4 files changed, 69 insertions(+), 11 deletions(-)

diff --git a/README.rst b/README.rst
index cc33a8a..5e0925b 100644
--- a/README.rst
+++ b/README.rst
@@ -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
 ----------------
 
diff --git a/changelog.d/feature.8899db6c.entry.yaml b/changelog.d/feature.8899db6c.entry.yaml
index da25e12..9201f49 100644
--- a/changelog.d/feature.8899db6c.entry.yaml
+++ b/changelog.d/feature.8899db6c.entry.yaml
@@ -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
diff --git a/pytest_subprocess/fake_popen.py b/pytest_subprocess/fake_popen.py
index ebf068b..c308359 100644
--- a/pytest_subprocess/fake_popen.py
+++ b/pytest_subprocess/fake_popen.py
@@ -3,6 +3,7 @@
 import asyncio
 import collections.abc
 import concurrent.futures
+import copy
 import io
 import os
 import signal
@@ -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
@@ -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
@@ -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)
@@ -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,
diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py
index fdd9126..2d8cd1c 100644
--- a/tests/test_subprocess.py
+++ b/tests/test_subprocess.py
@@ -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):