run_cmd_with_pipe
Wrapper for the default use case of the cumbersome "subprocess" library.
- Acceps a list of arguments that describe the command invocation. Returns
+ Accepts a list of arguments that describe the command invocation. Returns
``True`` and the contents of ``stdout`` if the pipe returned sucessfully,
``False`` plus ``stderr`` and the exit status otherwise. For example::
import re
import sysmisc
def parse(line):
- if re.match('\d', line):
+ if re.match(r'\d', line):
print('found digits in line!')
sysmisc.read_linewise('dump_db', parse)
hash_file
- Return a hash of a flie.
+ Return a hash of a file.
cheat_reboot
Replace the reboot binary with a fake one.
"""
-from __future__ import print_function
-
import re
import subprocess
import hashlib
llog = logging.getLogger('pyi2ncommon.sysmisc')
-__all__ = ("inf", "run_cmd_with_pipe", "get_mountpoints_by_type", "read_linewise", "hash_file", "cheat_reboot", "RUN_RESULT_OK", "RUN_RESULT_TIMEDOUT", "RUN_RESULT_FAIL", "RUN_RESULT_NAME", "cmd_block_till")
+__all__ = ("inf", "run_cmd_with_pipe", "get_mountpoints_by_type",
+ "read_linewise", "hash_file", "cheat_reboot",
+ "RUN_RESULT_OK", "RUN_RESULT_TIMEDOUT", "RUN_RESULT_FAIL",
+ "RUN_RESULT_NAME", "cmd_block_till")
###############################################################################
http://code.activestate.com/recipes/576620-changedirectory-context-manager/
(MIT license)
- :arg str path: paht to temporarily switch to
+ :arg str path: path to temporarily switch to
"""
orig_wd = os.getcwd()
os.chdir(path)
try:
with open(procmounts, "r") as m:
lines = list(m)
- pat = re.compile("^[^\s]+\s+([^\s]+)\s+" + fstype + "\s+.*$")
+ pat = re.compile(r"^\S+\s+(\S+)\s+" + fstype + r"\s+.*$")
mps = [mp.group(1)
for mp in map(lambda l: re.match(pat, l), lines)
if mp]
- except IOError as e:
- raise IOError("Failed to read %s." % procmounts)
+ except IOError:
+ raise IOError(f"Failed to read {procmounts}")
if not mps:
return None
return mps
# if proc.poll() is not None:
# break
- #rest_output,_ = proc.communicate()
- #for line in rest_output:
- # func(line)
+ # rest_output,_ = proc.communicate()
+ # for line in rest_output:
+ # func(line)
return proc.wait()
This replaces the ``reboot-intranator`` executable by script which
replaces itself by the backed up executable upon the next invocation.
"""
- #path = utils.system_output("which reboot")
+ # path = utils.system_output("which reboot")
path = "/usr/intranator/bin/reboot-intranator"
suffix = uuid.uuid1()
backup = backup_fmt % (path, backup_infix, suffix)
:param cmd: Command line or callback to execute. Function arguments must
have the same signature as :py:func:`run_cmd_with_pipe`.
- :type cmd: [str] | types.FunctionType
+ :type cmd: [str] | types.FunctionType
:param int timeout: Blocking timeout
-
- :returns: Pair of result and error message if appropriate or
- :py:value:`None`.
- :rtype: (run_result, str | None)
+ :param cond: Function to call; code will wait for this to return something
+ other than `False`
+ :param interval: Time (in seconds) to sleep between each attempt at `cond`
+ :returns: Pair of result and error message if appropriate or :py:value:`None`.
+ :rtype: (run_result, str | None)
"""
llog.debug("cmd_block_till: %r, %d s, %r", cmd, timeout, cond)
if isinstance(cmd, types.FunctionType):
else:
label = ""
name = CURRENT_TEST_NAME if isinstance(CURRENT_TEST_NAME, str) else ""
- fmt, label = str(fmt), str(label) # yes
+ fmt, label = str(fmt), str(label) # yes
llog.info("[%s%s] %s" % (name, label, fmt), *args)
# TODO: this method is more dynamic
# llog.info("[%s%s] %s%s" % (LOG_TAG, "", LOG_INDENT*indent, fmt), *args)