1
0
forked from brl/citadel
citadel/poky/meta/lib/oeqa/selftest/cases/runcmd.py

135 lines
4.9 KiB
Python
Raw Normal View History

from oeqa.selftest.case import OESelftestTestCase
from oeqa.utils.commands import runCmd
from oeqa.utils import CommandError
from oeqa.core.decorator.oeid import OETestID
import subprocess
import threading
import time
import signal
class MemLogger(object):
def __init__(self):
self.info_msgs = []
self.error_msgs = []
def info(self, msg):
self.info_msgs.append(msg)
def error(self, msg):
self.error_msgs.append(msg)
class RunCmdTests(OESelftestTestCase):
""" Basic tests for runCmd() utility function """
# The delta is intentionally smaller than the timeout, to detect cases where
# we incorrectly apply the timeout more than once.
TIMEOUT = 2
DELTA = 1
@OETestID(1916)
def test_result_okay(self):
result = runCmd("true")
self.assertEqual(result.status, 0)
@OETestID(1915)
def test_result_false(self):
result = runCmd("false", ignore_status=True)
self.assertEqual(result.status, 1)
@OETestID(1917)
def test_shell(self):
# A shell is used for all string commands.
result = runCmd("false; true", ignore_status=True)
self.assertEqual(result.status, 0)
@OETestID(1910)
def test_no_shell(self):
self.assertRaises(FileNotFoundError,
runCmd, "false; true", shell=False)
@OETestID(1906)
def test_list_not_found(self):
self.assertRaises(FileNotFoundError,
runCmd, ["false; true"])
@OETestID(1907)
def test_list_okay(self):
result = runCmd(["true"])
self.assertEqual(result.status, 0)
@OETestID(1913)
def test_result_assertion(self):
self.assertRaisesRegexp(AssertionError, "Command 'echo .* false' returned non-zero exit status 1:\nfoobar",
runCmd, "echo foobar >&2; false", shell=True)
@OETestID(1914)
def test_result_exception(self):
self.assertRaisesRegexp(CommandError, "Command 'echo .* false' returned non-zero exit status 1 with output: foobar",
runCmd, "echo foobar >&2; false", shell=True, assert_error=False)
@OETestID(1911)
def test_output(self):
result = runCmd("echo stdout; echo stderr >&2", shell=True)
self.assertEqual("stdout\nstderr", result.output)
self.assertEqual("", result.error)
@OETestID(1912)
def test_output_split(self):
result = runCmd("echo stdout; echo stderr >&2", shell=True, stderr=subprocess.PIPE)
self.assertEqual("stdout", result.output)
self.assertEqual("stderr", result.error)
@OETestID(1920)
def test_timeout(self):
numthreads = threading.active_count()
start = time.time()
# Killing a hanging process only works when not using a shell?!
result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True)
self.assertEqual(result.status, -signal.SIGTERM)
end = time.time()
self.assertLess(end - start, self.TIMEOUT + self.DELTA)
self.assertEqual(numthreads, threading.active_count())
@OETestID(1921)
def test_timeout_split(self):
numthreads = threading.active_count()
start = time.time()
# Killing a hanging process only works when not using a shell?!
result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True, stderr=subprocess.PIPE)
self.assertEqual(result.status, -signal.SIGTERM)
end = time.time()
self.assertLess(end - start, self.TIMEOUT + self.DELTA)
self.assertEqual(numthreads, threading.active_count())
@OETestID(1918)
def test_stdin(self):
numthreads = threading.active_count()
result = runCmd("cat", data=b"hello world", timeout=self.TIMEOUT)
self.assertEqual("hello world", result.output)
self.assertEqual(numthreads, threading.active_count())
@OETestID(1919)
def test_stdin_timeout(self):
numthreads = threading.active_count()
start = time.time()
result = runCmd(['sleep', '60'], data=b"hello world", timeout=self.TIMEOUT, ignore_status=True)
self.assertEqual(result.status, -signal.SIGTERM)
end = time.time()
self.assertLess(end - start, self.TIMEOUT + self.DELTA)
self.assertEqual(numthreads, threading.active_count())
@OETestID(1908)
def test_log(self):
log = MemLogger()
result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log)
self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout", "stderr"], log.info_msgs)
self.assertEqual([], log.error_msgs)
@OETestID(1909)
def test_log_split(self):
log = MemLogger()
result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log, stderr=subprocess.PIPE)
self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout"], log.info_msgs)
self.assertEqual(["stderr"], log.error_msgs)