blob: 3755764ee71dad162b8500af7e4f0ff402e5bc58 [file] [log] [blame]
Brad Bishopc342db32019-05-15 21:57:59 -04001#
2# SPDX-License-Identifier: MIT
3#
4
Brad Bishopd7bf8c12018-02-25 22:55:05 -05005from oeqa.selftest.case import OESelftestTestCase
6from oeqa.utils.commands import runCmd
7from oeqa.utils import CommandError
Brad Bishopd7bf8c12018-02-25 22:55:05 -05008
9import subprocess
10import threading
11import time
12import signal
13
14class MemLogger(object):
15 def __init__(self):
16 self.info_msgs = []
17 self.error_msgs = []
18
19 def info(self, msg):
20 self.info_msgs.append(msg)
21
22 def error(self, msg):
23 self.error_msgs.append(msg)
24
25class RunCmdTests(OESelftestTestCase):
26 """ Basic tests for runCmd() utility function """
27
28 # The delta is intentionally smaller than the timeout, to detect cases where
29 # we incorrectly apply the timeout more than once.
Brad Bishop19323692019-04-05 15:28:33 -040030 TIMEOUT = 5
31 DELTA = 3
Brad Bishopd7bf8c12018-02-25 22:55:05 -050032
Brad Bishopd7bf8c12018-02-25 22:55:05 -050033 def test_result_okay(self):
34 result = runCmd("true")
35 self.assertEqual(result.status, 0)
36
Brad Bishopd7bf8c12018-02-25 22:55:05 -050037 def test_result_false(self):
38 result = runCmd("false", ignore_status=True)
39 self.assertEqual(result.status, 1)
40
Brad Bishopd7bf8c12018-02-25 22:55:05 -050041 def test_shell(self):
42 # A shell is used for all string commands.
43 result = runCmd("false; true", ignore_status=True)
44 self.assertEqual(result.status, 0)
45
Brad Bishopd7bf8c12018-02-25 22:55:05 -050046 def test_no_shell(self):
47 self.assertRaises(FileNotFoundError,
48 runCmd, "false; true", shell=False)
49
Brad Bishopd7bf8c12018-02-25 22:55:05 -050050 def test_list_not_found(self):
51 self.assertRaises(FileNotFoundError,
52 runCmd, ["false; true"])
53
Brad Bishopd7bf8c12018-02-25 22:55:05 -050054 def test_list_okay(self):
55 result = runCmd(["true"])
56 self.assertEqual(result.status, 0)
57
Brad Bishopd7bf8c12018-02-25 22:55:05 -050058 def test_result_assertion(self):
59 self.assertRaisesRegexp(AssertionError, "Command 'echo .* false' returned non-zero exit status 1:\nfoobar",
60 runCmd, "echo foobar >&2; false", shell=True)
61
Brad Bishopd7bf8c12018-02-25 22:55:05 -050062 def test_result_exception(self):
63 self.assertRaisesRegexp(CommandError, "Command 'echo .* false' returned non-zero exit status 1 with output: foobar",
64 runCmd, "echo foobar >&2; false", shell=True, assert_error=False)
65
Brad Bishopd7bf8c12018-02-25 22:55:05 -050066 def test_output(self):
67 result = runCmd("echo stdout; echo stderr >&2", shell=True)
68 self.assertEqual("stdout\nstderr", result.output)
69 self.assertEqual("", result.error)
70
Brad Bishopd7bf8c12018-02-25 22:55:05 -050071 def test_output_split(self):
72 result = runCmd("echo stdout; echo stderr >&2", shell=True, stderr=subprocess.PIPE)
73 self.assertEqual("stdout", result.output)
74 self.assertEqual("stderr", result.error)
75
Brad Bishopd7bf8c12018-02-25 22:55:05 -050076 def test_timeout(self):
77 numthreads = threading.active_count()
78 start = time.time()
79 # Killing a hanging process only works when not using a shell?!
80 result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True)
81 self.assertEqual(result.status, -signal.SIGTERM)
82 end = time.time()
83 self.assertLess(end - start, self.TIMEOUT + self.DELTA)
84 self.assertEqual(numthreads, threading.active_count())
85
Brad Bishopd7bf8c12018-02-25 22:55:05 -050086 def test_timeout_split(self):
87 numthreads = threading.active_count()
88 start = time.time()
89 # Killing a hanging process only works when not using a shell?!
90 result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True, stderr=subprocess.PIPE)
91 self.assertEqual(result.status, -signal.SIGTERM)
92 end = time.time()
93 self.assertLess(end - start, self.TIMEOUT + self.DELTA)
94 self.assertEqual(numthreads, threading.active_count())
95
Brad Bishopd7bf8c12018-02-25 22:55:05 -050096 def test_stdin(self):
97 numthreads = threading.active_count()
98 result = runCmd("cat", data=b"hello world", timeout=self.TIMEOUT)
99 self.assertEqual("hello world", result.output)
100 self.assertEqual(numthreads, threading.active_count())
101
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500102 def test_stdin_timeout(self):
103 numthreads = threading.active_count()
104 start = time.time()
105 result = runCmd(['sleep', '60'], data=b"hello world", timeout=self.TIMEOUT, ignore_status=True)
106 self.assertEqual(result.status, -signal.SIGTERM)
107 end = time.time()
108 self.assertLess(end - start, self.TIMEOUT + self.DELTA)
109 self.assertEqual(numthreads, threading.active_count())
110
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500111 def test_log(self):
112 log = MemLogger()
113 result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log)
114 self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout", "stderr"], log.info_msgs)
115 self.assertEqual([], log.error_msgs)
116
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500117 def test_log_split(self):
118 log = MemLogger()
119 result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log, stderr=subprocess.PIPE)
120 self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout"], log.info_msgs)
121 self.assertEqual(["stderr"], log.error_msgs)