Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

QOL: Context run command accepts either str or list #987

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
4 changes: 3 additions & 1 deletion .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -3,4 +3,6 @@ branch = True
include =
invoke/*
tests/*
omit = invoke/vendor/*
omit =
invoke/shims.py
invoke/vendor/*
6 changes: 3 additions & 3 deletions invoke/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Optional
from typing import Any, List, Optional, Union

from ._version import __version_info__, __version__ # noqa
from .collection import Collection # noqa
Expand Down Expand Up @@ -31,7 +31,7 @@
from .watchers import FailingResponder, Responder, StreamWatcher # noqa


def run(command: str, **kwargs: Any) -> Optional[Result]:
def run(command: Union[str, List[str]], **kwargs: Any) -> Optional[Result]:
"""
Run ``command`` in a subprocess and return a `.Result` object.

Expand All @@ -50,7 +50,7 @@ def run(command: str, **kwargs: Any) -> Optional[Result]:
return Context().run(command, **kwargs)


def sudo(command: str, **kwargs: Any) -> Optional[Result]:
def sudo(command: Union[str, List[str]], **kwargs: Any) -> Optional[Result]:
"""
Run ``command`` in a ``sudo`` subprocess and return a `.Result` object.

Expand Down
25 changes: 21 additions & 4 deletions invoke/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from .config import Config, DataProxy
from .exceptions import Failure, AuthFailure, ResponseNotAccepted
from .runners import Result
from .shims import shlex_join
from .watchers import FailingResponder

if TYPE_CHECKING:
Expand Down Expand Up @@ -87,7 +88,9 @@ def config(self, value: Config) -> None:
# runtime.
self._set(_config=value)

def run(self, command: str, **kwargs: Any) -> Optional[Result]:
def run(
self, command: Union[str, List[str]], **kwargs: Any
) -> Optional[Result]:
"""
Execute a local shell command, honoring config options.

Expand All @@ -101,6 +104,8 @@ def run(self, command: str, **kwargs: Any) -> Optional[Result]:
.. versionadded:: 1.0
"""
runner = self.config.runners.local(self)
if isinstance(command, list):
command = shlex_join(command)
return self._run(runner, command, **kwargs)

# NOTE: broken out of run() to allow for runner class injection in
Expand All @@ -112,7 +117,9 @@ def _run(
command = self._prefix_commands(command)
return runner.run(command, **kwargs)

def sudo(self, command: str, **kwargs: Any) -> Optional[Result]:
def sudo(
self, command: Union[str, List[str]], **kwargs: Any
) -> Optional[Result]:
"""
Execute a shell command via ``sudo`` with password auto-response.

Expand Down Expand Up @@ -182,6 +189,8 @@ def sudo(self, command: str, **kwargs: Any) -> Optional[Result]:
.. versionadded:: 1.0
"""
runner = self.config.runners.local(self)
if isinstance(command, list):
command = shlex_join(command)
return self._sudo(runner, command, **kwargs)

# NOTE: this is for runner injection; see NOTE above _run().
Expand Down Expand Up @@ -548,18 +557,26 @@ def _yield_result(self, attname: str, command: str) -> Result:
# raise_from(NotImplementedError(command), None)
raise NotImplementedError(command)

def run(self, command: str, *args: Any, **kwargs: Any) -> Result:
def run(
self, command: Union[str, List[str]], *args: Any, **kwargs: Any
) -> Result:
# TODO: perform more convenience stuff associating args/kwargs with the
# result? E.g. filling in .command, etc? Possibly useful for debugging
# if one hits unexpected-order problems with what they passed in to
# __init__.
if isinstance(command, list):
command = shlex_join(command)
return self._yield_result("__run", command)

def sudo(self, command: str, *args: Any, **kwargs: Any) -> Result:
def sudo(
self, command: Union[str, List[str]], *args: Any, **kwargs: Any
) -> Result:
# TODO: this completely nukes the top-level behavior of sudo(), which
# could be good or bad, depending. Most of the time I think it's good.
# No need to supply dummy password config, etc.
# TODO: see the TODO from run() re: injecting arg/kwarg values
if isinstance(command, list):
command = shlex_join(command)
return self._yield_result("__sudo", command)

def set_result_for(
Expand Down
16 changes: 16 additions & 0 deletions invoke/shims.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
import shlex
import sys
from typing import List


if sys.version_info >= (3, 8):

def shlex_join(split_command: List) -> str:
"""Convert command from list to str."""
return shlex.join(split_command)

else:

def shlex_join(split_command: List) -> str:
"""Convert command from list to str."""
return shlex.quote(" ".join(split_command))[1:-1]
59 changes: 46 additions & 13 deletions tests/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,27 @@ def honors_runner_config_setting(self):
c.run("foo")
assert runner_class.mock_calls == [call(c), call().run("foo")]

def sudo(self):
self._expect_attr("sudo")
@patch(local_path)
def converts_command_list_to_str(self, Local):
runner = Local.return_value
c = Context()
c.run(["foo", "bar", "baz"])
cmd = "foo bar baz"
assert runner.run.called, "run() never called runner.run()!"
assert runner.run.call_args[0][0] == cmd

class sudo:
def exists(self):
self._expect_attr("sudo")

@patch(local_path)
def converts_command_list_to_str(self, Local):
runner = Local.return_value
c = Context()
c.sudo(["foo", "bar", "baz"])
cmd = "sudo -S -p '[sudo] password: ' foo bar baz"
assert runner.run.called, "sudo() never called runner.run()!"
assert runner.run.call_args[0][0] == cmd

class configuration_proxy:
"Dict-like proxy for self.config"
Expand Down Expand Up @@ -735,17 +754,31 @@ def sudo(self):
with raises(TypeError):
mc.set_result_for("sudo", "whatever", Result("bar"))

def run(self):
mc = MockContext(run={"foo": Result("bar")})
assert mc.run("foo").stdout == "bar"
mc.set_result_for("run", "foo", Result("biz"))
assert mc.run("foo").stdout == "biz"

def sudo(self):
mc = MockContext(sudo={"foo": Result("bar")})
assert mc.sudo("foo").stdout == "bar"
mc.set_result_for("sudo", "foo", Result("biz"))
assert mc.sudo("foo").stdout == "biz"
class run:
def sets_result_for_command_str(self):
mc = MockContext(run={"foo": Result("bar")})
assert mc.run("foo").stdout == "bar"
mc.set_result_for("run", "foo", Result("biz"))
assert mc.run("foo").stdout == "biz"

def sets_result_for_command_list(self):
mc = MockContext(run={"foo fpp fqq": Result("bar")})
assert mc.run(["foo", "fpp", "fqq"]).stdout == "bar"
mc.set_result_for("run", "foo fpp fqq", Result("biz"))
assert mc.run(["foo", "fpp", "fqq"]).stdout == "biz"

class sudo:
def sets_result_for_command_str(self):
mc = MockContext(sudo={"foo": Result("bar")})
assert mc.sudo("foo").stdout == "bar"
mc.set_result_for("sudo", "foo", Result("biz"))
assert mc.sudo("foo").stdout == "biz"

def sets_result_for_command_list(self):
mc = MockContext(sudo={"foo fpp fqq": Result("bar")})
assert mc.sudo(["foo", "fpp", "fqq"]).stdout == "bar"
mc.set_result_for("sudo", "foo fpp fqq", Result("biz"))
assert mc.sudo(["foo", "fpp", "fqq"]).stdout == "biz"

def wraps_run_and_sudo_with_Mock(self, clean_sys_modules):
sys.modules["mock"] = None # legacy
Expand Down