diff options
author | Dave Cobbley <david.j.cobbley@linux.intel.com> | 2018-08-14 20:05:37 +0300 |
---|---|---|
committer | Brad Bishop <bradleyb@fuzziesquirrel.com> | 2018-08-23 04:26:31 +0300 |
commit | eb8dc40360f0cfef56fb6947cc817a547d6d9bc6 (patch) | |
tree | de291a73dc37168da6370e2cf16c347d1eba9df8 /poky/meta/lib/oeqa/selftest/cases/runcmd.py | |
parent | 9c3cf826d853102535ead04cebc2d6023eff3032 (diff) | |
download | openbmc-eb8dc40360f0cfef56fb6947cc817a547d6d9bc6.tar.xz |
[Subtree] Removing import-layers directory
As part of the move to subtrees, need to bring all the import layers
content to the top level.
Change-Id: I4a163d10898cbc6e11c27f776f60e1a470049d8f
Signed-off-by: Dave Cobbley <david.j.cobbley@linux.intel.com>
Signed-off-by: Brad Bishop <bradleyb@fuzziesquirrel.com>
Diffstat (limited to 'poky/meta/lib/oeqa/selftest/cases/runcmd.py')
-rw-r--r-- | poky/meta/lib/oeqa/selftest/cases/runcmd.py | 134 |
1 files changed, 134 insertions, 0 deletions
diff --git a/poky/meta/lib/oeqa/selftest/cases/runcmd.py b/poky/meta/lib/oeqa/selftest/cases/runcmd.py new file mode 100644 index 000000000..d76d7063c --- /dev/null +++ b/poky/meta/lib/oeqa/selftest/cases/runcmd.py @@ -0,0 +1,134 @@ +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) |