forked from clan/clan-core
remove unused ssh.run method
This commit is contained in:
parent
c66c25aeb7
commit
a2f0d077c8
@ -20,9 +20,7 @@ from typing import (
|
||||
IO,
|
||||
Any,
|
||||
Generic,
|
||||
Literal,
|
||||
TypeVar,
|
||||
overload,
|
||||
)
|
||||
|
||||
# https://no-color.org
|
||||
@ -786,74 +784,3 @@ def parse_deployment_address(
|
||||
meta=meta,
|
||||
ssh_options=options,
|
||||
)
|
||||
|
||||
|
||||
@overload
|
||||
def run(
|
||||
cmd: list[str] | str,
|
||||
text: Literal[True] = ...,
|
||||
stdout: FILE = ...,
|
||||
stderr: FILE = ...,
|
||||
extra_env: dict[str, str] = ...,
|
||||
cwd: None | str | Path = ...,
|
||||
check: bool = ...,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
...
|
||||
|
||||
|
||||
@overload
|
||||
def run(
|
||||
cmd: list[str] | str,
|
||||
text: Literal[False],
|
||||
stdout: FILE = ...,
|
||||
stderr: FILE = ...,
|
||||
extra_env: dict[str, str] = ...,
|
||||
cwd: None | str | Path = ...,
|
||||
check: bool = ...,
|
||||
) -> subprocess.CompletedProcess[bytes]:
|
||||
...
|
||||
|
||||
|
||||
def run(
|
||||
cmd: list[str] | str,
|
||||
text: bool = True,
|
||||
stdout: FILE = None,
|
||||
stderr: FILE = None,
|
||||
extra_env: dict[str, str] = {},
|
||||
cwd: None | str | Path = None,
|
||||
check: bool = True,
|
||||
) -> subprocess.CompletedProcess[Any]:
|
||||
"""
|
||||
Run command locally
|
||||
|
||||
@cmd if this parameter is a string the command is interpreted as a shell command,
|
||||
otherwise if it is a list, than the first list element is the command
|
||||
and the remaining list elements are passed as arguments to the
|
||||
command.
|
||||
@text when true, file objects for stdout and stderr are opened in text mode.
|
||||
@stdout if not None stdout of the command will be redirected to this file i.e. stdout=subprocss.PIPE
|
||||
@stderr if not None stderr of the command will be redirected to this file i.e. stderr=subprocess.PIPE
|
||||
@extra_env environment variables to override whe running the command
|
||||
@cwd current working directory to run the process in
|
||||
@check If check is true, and the process exits with a non-zero exit code, a
|
||||
CalledProcessError exception will be raised. Attributes of that exception
|
||||
hold the arguments, the exit code, and stdout and stderr if they were
|
||||
captured.
|
||||
"""
|
||||
if isinstance(cmd, list):
|
||||
info("$ " + " ".join(cmd))
|
||||
else:
|
||||
info(f"$ {cmd}")
|
||||
env = os.environ.copy()
|
||||
env.update(extra_env)
|
||||
|
||||
return subprocess.run(
|
||||
cmd,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
env=env,
|
||||
cwd=cwd,
|
||||
check=check,
|
||||
shell=not isinstance(cmd, list),
|
||||
text=text,
|
||||
)
|
||||
|
@ -1,32 +1,11 @@
|
||||
import subprocess
|
||||
|
||||
from clan_cli.ssh import Host, HostGroup, run
|
||||
|
||||
|
||||
def test_run() -> None:
|
||||
p = run("echo hello")
|
||||
assert p.stdout is None
|
||||
|
||||
|
||||
def test_run_failure() -> None:
|
||||
p = run("exit 1", check=False)
|
||||
assert p.returncode == 1
|
||||
|
||||
try:
|
||||
p = run("exit 1")
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
assert False, "Command should have raised an error"
|
||||
|
||||
from clan_cli.ssh import Host, HostGroup
|
||||
|
||||
hosts = HostGroup([Host("some_host")])
|
||||
|
||||
|
||||
def test_run_environment() -> None:
|
||||
p1 = run("echo $env_var", stdout=subprocess.PIPE, extra_env=dict(env_var="true"))
|
||||
assert p1.stdout == "true\n"
|
||||
|
||||
p2 = hosts.run_local(
|
||||
"echo $env_var", extra_env=dict(env_var="true"), stdout=subprocess.PIPE
|
||||
)
|
||||
@ -38,17 +17,6 @@ def test_run_environment() -> None:
|
||||
assert "env_var=true" in p3[0].result.stdout
|
||||
|
||||
|
||||
def test_run_non_shell() -> None:
|
||||
p = run(["echo", "$hello"], stdout=subprocess.PIPE)
|
||||
assert p.stdout == "$hello\n"
|
||||
|
||||
|
||||
def test_run_stderr_stdout() -> None:
|
||||
p = run("echo 1; echo 2 >&2", stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
assert p.stdout == "1\n"
|
||||
assert p.stderr == "2\n"
|
||||
|
||||
|
||||
def test_run_local() -> None:
|
||||
hosts.run_local("echo hello")
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user