add FakePopenWrapper

pytest_subprocess/process_dispatcher.py

@@ -8,10 +8,12 @@
 from copy import deepcopy
 from functools import partial
 from typing import Any as AnyType
+from typing import AnyStr
 from typing import Awaitable
 from typing import Callable
 from typing import Deque
 from typing import Dict
+from typing import Generic
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -29,6 +31,9 @@
     from .fake_process import FakeProcess
 
 
+__all__ = ["ProcessDispatcher"]
+
+
 class ProcessDispatcher:
     """Main class for handling processes."""
 
@@ -44,7 +49,7 @@ class ProcessDispatcher:
     def register(cls, process: "FakeProcess") -> None:
         if not cls.process_list:
             cls.built_in_popen = subprocess.Popen
-            subprocess.Popen = cls.dispatch  # type: ignore
+            subprocess.Popen = FakePopenWrapper  # type: ignore
 
             cls.built_in_async_subprocess = asyncio.subprocess
             asyncio.create_subprocess_shell = cls.async_shell  # type: ignore
@@ -238,3 +243,10 @@ def _get_process(
             if processes and isinstance(processes, deque):
                 return command_instance, processes, process_instance
         return None, None, None
+
+
+class FakePopenWrapper(Generic[AnyStr]):
+    def __new__(  # type: ignore
+        cls, command: COMMAND, **kwargs: Optional[Dict]
+    ) -> FakePopen:
+        return ProcessDispatcher.dispatch(command, **kwargs)  # type: ignore

tests/test_subprocess.py

@@ -1232,3 +1232,25 @@ def test_process_recorder(fp):
     recorder.clear()
 
     assert not recorder.was_called()
+
+
+def test_fake_popen_is_typed(fp):
+    fp.allow_unregistered(True)
+    fp.register(
+        [PYTHON, "example_script.py"],
+        stdout=b"Stdout line 1\r\nStdout line 2\r\n",
+    )
+
+    def spawn_process() -> subprocess.Popen[str]:
+        import subprocess
+
+        return subprocess.Popen(
+            (PYTHON, "example_script.py"),
+            universal_newlines=True,
+            stdout=subprocess.PIPE,
+        )
+
+    proc = spawn_process()
+    proc.wait()
+
+    assert proc.stdout.read() == "Stdout line 1\nStdout line 2\n"