diff --git a/changelog.d/bug.c797773a.entry.yaml b/changelog.d/bug.c797773a.entry.yaml new file mode 100644 index 0000000..2bb6e68 --- /dev/null +++ b/changelog.d/bug.c797773a.entry.yaml @@ -0,0 +1,6 @@ +message: Wrapped ProcessDispatcher.dispatch into FakePopenWrapper as it was causing + TypeError when Popen is used as a type. +pr_ids: +- '170' +timestamp: 1727883418 +type: bug diff --git a/pytest_subprocess/process_dispatcher.py b/pytest_subprocess/process_dispatcher.py index 35f3da2..7613e13 100644 --- a/pytest_subprocess/process_dispatcher.py +++ b/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 diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py index da0d8d9..29a7dad 100644 --- a/tests/test_subprocess.py +++ b/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"