Commit e623a122 authored by Victor Stinner's avatar Victor Stinner

asyncio: subprocess_shell() and subprocess_exec() now raise ValueError instead of assert.

Moreover, bufsize different than 0 is now considered as an error.
parent 73f10fd2
......@@ -552,9 +552,14 @@ class BaseEventLoop(events.AbstractEventLoop):
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=False, shell=True, bufsize=0,
**kwargs):
assert not universal_newlines, "universal_newlines must be False"
assert shell, "shell must be True"
assert isinstance(cmd, str), cmd
if not isinstance(cmd, str):
raise ValueError("cmd must be a string")
if universal_newlines:
raise ValueError("universal_newlines must be False")
if not shell:
raise ValueError("shell must be False")
if bufsize != 0:
raise ValueError("bufsize must be 0")
protocol = protocol_factory()
transport = yield from self._make_subprocess_transport(
protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs)
......@@ -565,8 +570,12 @@ class BaseEventLoop(events.AbstractEventLoop):
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=False, shell=False, bufsize=0,
**kwargs):
assert not universal_newlines, "universal_newlines must be False"
assert not shell, "shell must be False"
if universal_newlines:
raise ValueError("universal_newlines must be False")
if shell:
raise ValueError("shell must be False")
if bufsize != 0:
raise ValueError("bufsize must be 0")
protocol = protocol_factory()
transport = yield from self._make_subprocess_transport(
protocol, args, False, stdin, stdout, stderr, bufsize, **kwargs)
......
......@@ -1491,6 +1491,38 @@ class SubprocessTestsMixin:
self.loop.run_until_complete(proto.completed)
self.assertEqual(7, proto.returncode)
def test_subprocess_exec_invalid_args(self):
@asyncio.coroutine
def connect(**kwds):
yield from self.loop.subprocess_exec(
asyncio.SubprocessProtocol,
'pwd', **kwds)
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(universal_newlines=True))
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(bufsize=4096))
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(shell=True))
def test_subprocess_shell_invalid_args(self):
@asyncio.coroutine
def connect(cmd=None, **kwds):
if not cmd:
cmd = 'pwd'
yield from self.loop.subprocess_shell(
asyncio.SubprocessProtocol,
cmd, **kwds)
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(['ls', '-l']))
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(universal_newlines=True))
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(bufsize=4096))
with self.assertRaises(ValueError):
self.loop.run_until_complete(connect(shell=False))
if sys.platform == 'win32':
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment