return searchval == val
-def sh(command):
+def sh(command, log_limit=128):
"""
Run the shell command and return the output in ascii (stderr and
stdout). If the command fails, raise an exception. The command
shell=True)
output = proc.communicate()[0]
if output.strip():
- log.debug(command + " output " + str(output))
+ if len(output) > log_limit:
+ log.debug(command + " output " + str(output)[:log_limit] +
+ "... (truncated to the first " + str(log_limit) +
+ " characters)")
+ else:
+ log.debug(command + " output " + str(output))
if proc.returncode != 0:
+ log.debug(command + " failed with " + str(output))
raise subprocess.CalledProcessError(
returncode=proc.returncode,
cmd=command,
class FakeRemote(object):
pass
+def test_sh_normal(caplog):
+ assert misc.sh("echo ABC") == "ABC\n"
+ assert "truncated" not in caplog.text()
+
+def test_sh_truncate(caplog):
+ assert misc.sh("echo -n AB ; echo C", 2) == "ABC\n"
+ assert "truncated" in caplog.text()
+ assert "ABC" not in caplog.text()
+
+def test_sh_fail(caplog):
+ with pytest.raises(Exception) as excinfo:
+ misc.sh("echo -n AB ; echo C ; exit 111", 2) == "ABC\n"
+ assert excinfo.value.returncode == 111
+ assert "failed with ABC" in caplog.text()
def test_wait_until_osds_up():
ctx = argparse.Namespace()