StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

utils: accept sequences of strings as commands 

We used to only accept lists of strings for commands. We now accept
sequences of strings instead ; which are lists of strings or tuple of
strings.

Signed-off-by: Olivier Gayot <olivier.gayot@canonical.com>
This commit is contained in:
Olivier Gayot 2022-04-05 10:24:57 +02:00
parent 9f5d6d9c08
commit a63a3ef753
1 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
subiquitycore/utils.py
View File

@ -19,7 +19,7 @@ import logging
import os import os
import random import random
import subprocess import subprocess
from typing import List from typing import List, Sequence
log = logging.getLogger("subiquitycore.utils") log = logging.getLogger("subiquitycore.utils")
@ -35,7 +35,7 @@ def _clean_env(env):
return env return env
def run_command(cmd: List[str], *, input=None, stdout=subprocess.PIPE, def run_command(cmd: Sequence[str], *, input=None, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, encoding='utf-8', errors='replace', stderr=subprocess.PIPE, encoding='utf-8', errors='replace',
env=None, **kw) -> subprocess.CompletedProcess: env=None, **kw) -> subprocess.CompletedProcess:
"""A wrapper around subprocess.run with logging and different defaults. """A wrapper around subprocess.run with logging and different defaults.
@ -63,7 +63,7 @@ def run_command(cmd: List[str], *, input=None, stdout=subprocess.PIPE,
return cp return cp
async def arun_command(cmd: List[str], *, async def arun_command(cmd: Sequence[str], *,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
encoding='utf-8', input=None, errors='replace', encoding='utf-8', input=None, errors='replace',
env=None, check=False, **kw) \ env=None, check=False, **kw) \
@ -93,7 +93,7 @@ async def arun_command(cmd: List[str], *,
cmd, proc.returncode, stdout, stderr) cmd, proc.returncode, stdout, stderr)
async def astart_command(cmd: List[str], *, stdout=subprocess.PIPE, async def astart_command(cmd: Sequence[str], *, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL,
env=None, **kw) -> asyncio.subprocess.Process: env=None, **kw) -> asyncio.subprocess.Process:
log.debug("astart_command called: %s", cmd) log.debug("astart_command called: %s", cmd)
@ -102,12 +102,12 @@ async def astart_command(cmd: List[str], *, stdout=subprocess.PIPE,
env=_clean_env(env), **kw) env=_clean_env(env), **kw)
async def split_cmd_output(cmd: List[str], split_on: str) -> List[str]: async def split_cmd_output(cmd: Sequence[str], split_on: str) -> List[str]:
cp = await arun_command(cmd, check=True) cp = await arun_command(cmd, check=True)
return cp.stdout.split(split_on) return cp.stdout.split(split_on)
def start_command(cmd: List[str], *, def start_command(cmd: Sequence[str], *,
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, encoding='utf-8', errors='replace', stderr=subprocess.PIPE, encoding='utf-8', errors='replace',
env=None, **kw) -> subprocess.Popen: env=None, **kw) -> subprocess.Popen: