summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorColin Watson <cjwatson@canonical.com>2022-02-09 10:51:44 +0000
committerStephen Finucane <stephenfinucane@hotmail.com>2022-02-09 15:50:27 +0000
commit0ebc44f67c1bf14c73af6d3e908e207a3f7ce3f0 (patch)
tree8df779fdc6c223e129238d5747e8a14532c95ad8
parent7b25135ffdecd37e803ff860a6ec4128d255f9c6 (diff)
downloadfixtures-git-0ebc44f67c1bf14c73af6d3e908e207a3f7ce3f0.tar.gz
Support all Popen arguments up to Python 3.10
Fixes #53.
-rw-r--r--NEWS1
-rw-r--r--fixtures/_fixtures/popen.py30
-rw-r--r--fixtures/tests/_fixtures/test_popen.py57
3 files changed, 71 insertions, 17 deletions
diff --git a/NEWS b/NEWS
index af2c00c..1439d53 100644
--- a/NEWS
+++ b/NEWS
@@ -7,6 +7,7 @@ NEXT
* Dropped support for Python 2.7, Python 3.4 and Python 3.5 (EOL).
* Added support for Python 3.7-3.10.
+* Support all ``subprocess.Popen`` arguments up to Python 3.10.
3.0.0
~~~~~
diff --git a/fixtures/_fixtures/popen.py b/fixtures/_fixtures/popen.py
index c35ed5e..ffa9bf4 100644
--- a/fixtures/_fixtures/popen.py
+++ b/fixtures/_fixtures/popen.py
@@ -20,6 +20,7 @@ __all__ = [
import random
import subprocess
+import sys
from fixtures import Fixture
@@ -126,13 +127,38 @@ class FakePopen(Fixture):
stdin=_unpassed, stdout=_unpassed, stderr=_unpassed,
preexec_fn=_unpassed, close_fds=_unpassed, shell=_unpassed,
cwd=_unpassed, env=_unpassed, universal_newlines=_unpassed,
- startupinfo=_unpassed, creationflags=_unpassed):
+ startupinfo=_unpassed, creationflags=_unpassed,
+ restore_signals=_unpassed, start_new_session=_unpassed,
+ pass_fds=_unpassed, *, group=_unpassed, extra_groups=_unpassed,
+ user=_unpassed, umask=_unpassed, encoding=_unpassed,
+ errors=_unpassed, text=_unpassed, pipesize=_unpassed):
+ # Reject arguments introduced by newer versions of Python in older
+ # versions; this makes it harder to accidentally hide compatibility
+ # problems using test doubles.
+ if sys.version_info < (3, 7) and text is not FakePopen._unpassed:
+ raise TypeError(
+ "FakePopen.__call__() got an unexpected keyword argument "
+ "'text'")
+ if sys.version_info < (3, 9):
+ for arg_name in "group", "extra_groups", "user", "umask":
+ if locals()[arg_name] is not FakePopen._unpassed:
+ raise TypeError(
+ "FakePopen.__call__() got an unexpected keyword "
+ "argument '{}'".format(arg_name))
+ if sys.version_info < (3, 10) and pipesize is not FakePopen._unpassed:
+ raise TypeError(
+ "FakePopen.__call__() got an unexpected keyword argument "
+ "'pipesize'")
+
proc_args = dict(args=args)
local = locals()
for param in [
"bufsize", "executable", "stdin", "stdout", "stderr",
"preexec_fn", "close_fds", "shell", "cwd", "env",
- "universal_newlines", "startupinfo", "creationflags"]:
+ "universal_newlines", "startupinfo", "creationflags",
+ "restore_signals", "start_new_session", "pass_fds", "group",
+ "extra_groups", "user", "umask", "encoding", "errors", "text",
+ "pipesize"]:
if local[param] is not FakePopen._unpassed:
proc_args[param] = local[param]
proc_info = self.get_info(proc_args)
diff --git a/fixtures/tests/_fixtures/test_popen.py b/fixtures/tests/_fixtures/test_popen.py
index c5dc141..cafd98e 100644
--- a/fixtures/tests/_fixtures/test_popen.py
+++ b/fixtures/tests/_fixtures/test_popen.py
@@ -15,6 +15,7 @@
import io
import subprocess
+import sys
import testtools
@@ -48,33 +49,59 @@ class TestFakePopen(testtools.TestCase, TestWithFixtures):
proc = fixture(['foo'])
self.assertEqual('stdout', proc.stdout)
- def test_handles_all_2_7_args(self):
- all_args = dict(
- args="args", bufsize="bufsize", executable="executable",
- stdin="stdin", stdout="stdout", stderr="stderr",
- preexec_fn="preexec_fn", close_fds="close_fds", shell="shell",
- cwd="cwd", env="env", universal_newlines="universal_newlines",
- startupinfo="startupinfo", creationflags="creationflags")
- def get_info(proc_args):
- self.assertEqual(all_args, proc_args)
- return {}
- fixture = self.useFixture(FakePopen(get_info))
- fixture(**all_args)
-
- def test_handles_all_3_7_args(self):
+ def test_handles_all_Popen_args(self):
all_args = dict(
args="args", bufsize="bufsize", executable="executable",
stdin="stdin", stdout="stdout", stderr="stderr",
preexec_fn="preexec_fn", close_fds="close_fds", shell="shell",
cwd="cwd", env="env", universal_newlines="universal_newlines",
startupinfo="startupinfo", creationflags="creationflags",
- text="text")
+ restore_signals="restore_signals",
+ start_new_session="start_new_session", pass_fds="pass_fds",
+ encoding="encoding", errors="errors")
+ if sys.version_info >= (3, 7):
+ all_args["text"] = "text"
+ if sys.version_info >= (3, 9):
+ all_args["group"] = "group"
+ all_args["extra_groups"] = "extra_groups"
+ all_args["user"] = "user"
+ all_args["umask"] = "umask"
+ if sys.version_info >= (3, 10):
+ all_args["pipesize"] = "pipesize"
def get_info(proc_args):
self.assertEqual(all_args, proc_args)
return {}
fixture = self.useFixture(FakePopen(get_info))
fixture(**all_args)
+ @testtools.skipUnless(
+ sys.version_info < (3, 7), "only relevant on Python <3.7")
+ def test_rejects_3_7_args_on_older_versions(self):
+ fixture = self.useFixture(FakePopen(lambda proc_args: {}))
+ with testtools.ExpectedException(
+ TypeError, r".* got an unexpected keyword argument 'text'"):
+ fixture(args="args", text=True)
+
+ @testtools.skipUnless(
+ sys.version_info < (3, 9), "only relevant on Python <3.9")
+ def test_rejects_3_9_args_on_older_versions(self):
+ fixture = self.useFixture(FakePopen(lambda proc_args: {}))
+ for arg_name in ("group", "extra_groups", "user", "umask"):
+ kwargs = {arg_name: arg_name}
+ expected_message = (
+ r".* got an unexpected keyword argument '{}'".format(arg_name))
+ with testtools.ExpectedException(TypeError, expected_message):
+ fixture(args="args", **kwargs)
+
+ @testtools.skipUnless(
+ sys.version_info < (3, 10), "only relevant on Python <3.10")
+ def test_rejects_3_10_args_on_older_versions(self):
+ fixture = self.useFixture(FakePopen(lambda proc_args: {}))
+ with testtools.ExpectedException(
+ TypeError,
+ r".* got an unexpected keyword argument 'pipesize'"):
+ fixture(args="args", pipesize=1024)
+
def test_custom_returncode(self):
def get_info(proc_args):
return dict(returncode=1)