runcmd.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  1. from oeqa.selftest.case import OESelftestTestCase
  2. from oeqa.utils.commands import runCmd
  3. from oeqa.utils import CommandError
  4. import subprocess
  5. import threading
  6. import time
  7. import signal
  8. class MemLogger(object):
  9. def __init__(self):
  10. self.info_msgs = []
  11. self.error_msgs = []
  12. def info(self, msg):
  13. self.info_msgs.append(msg)
  14. def error(self, msg):
  15. self.error_msgs.append(msg)
  16. class RunCmdTests(OESelftestTestCase):
  17. """ Basic tests for runCmd() utility function """
  18. # The delta is intentionally smaller than the timeout, to detect cases where
  19. # we incorrectly apply the timeout more than once.
  20. TIMEOUT = 5
  21. DELTA = 3
  22. def test_result_okay(self):
  23. result = runCmd("true")
  24. self.assertEqual(result.status, 0)
  25. def test_result_false(self):
  26. result = runCmd("false", ignore_status=True)
  27. self.assertEqual(result.status, 1)
  28. def test_shell(self):
  29. # A shell is used for all string commands.
  30. result = runCmd("false; true", ignore_status=True)
  31. self.assertEqual(result.status, 0)
  32. def test_no_shell(self):
  33. self.assertRaises(FileNotFoundError,
  34. runCmd, "false; true", shell=False)
  35. def test_list_not_found(self):
  36. self.assertRaises(FileNotFoundError,
  37. runCmd, ["false; true"])
  38. def test_list_okay(self):
  39. result = runCmd(["true"])
  40. self.assertEqual(result.status, 0)
  41. def test_result_assertion(self):
  42. self.assertRaisesRegexp(AssertionError, "Command 'echo .* false' returned non-zero exit status 1:\nfoobar",
  43. runCmd, "echo foobar >&2; false", shell=True)
  44. def test_result_exception(self):
  45. self.assertRaisesRegexp(CommandError, "Command 'echo .* false' returned non-zero exit status 1 with output: foobar",
  46. runCmd, "echo foobar >&2; false", shell=True, assert_error=False)
  47. def test_output(self):
  48. result = runCmd("echo stdout; echo stderr >&2", shell=True)
  49. self.assertEqual("stdout\nstderr", result.output)
  50. self.assertEqual("", result.error)
  51. def test_output_split(self):
  52. result = runCmd("echo stdout; echo stderr >&2", shell=True, stderr=subprocess.PIPE)
  53. self.assertEqual("stdout", result.output)
  54. self.assertEqual("stderr", result.error)
  55. def test_timeout(self):
  56. numthreads = threading.active_count()
  57. start = time.time()
  58. # Killing a hanging process only works when not using a shell?!
  59. result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True)
  60. self.assertEqual(result.status, -signal.SIGTERM)
  61. end = time.time()
  62. self.assertLess(end - start, self.TIMEOUT + self.DELTA)
  63. self.assertEqual(numthreads, threading.active_count())
  64. def test_timeout_split(self):
  65. numthreads = threading.active_count()
  66. start = time.time()
  67. # Killing a hanging process only works when not using a shell?!
  68. result = runCmd(['sleep', '60'], timeout=self.TIMEOUT, ignore_status=True, stderr=subprocess.PIPE)
  69. self.assertEqual(result.status, -signal.SIGTERM)
  70. end = time.time()
  71. self.assertLess(end - start, self.TIMEOUT + self.DELTA)
  72. self.assertEqual(numthreads, threading.active_count())
  73. def test_stdin(self):
  74. numthreads = threading.active_count()
  75. result = runCmd("cat", data=b"hello world", timeout=self.TIMEOUT)
  76. self.assertEqual("hello world", result.output)
  77. self.assertEqual(numthreads, threading.active_count())
  78. def test_stdin_timeout(self):
  79. numthreads = threading.active_count()
  80. start = time.time()
  81. result = runCmd(['sleep', '60'], data=b"hello world", timeout=self.TIMEOUT, ignore_status=True)
  82. self.assertEqual(result.status, -signal.SIGTERM)
  83. end = time.time()
  84. self.assertLess(end - start, self.TIMEOUT + self.DELTA)
  85. self.assertEqual(numthreads, threading.active_count())
  86. def test_log(self):
  87. log = MemLogger()
  88. result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log)
  89. self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout", "stderr"], log.info_msgs)
  90. self.assertEqual([], log.error_msgs)
  91. def test_log_split(self):
  92. log = MemLogger()
  93. result = runCmd("echo stdout; echo stderr >&2", shell=True, output_log=log, stderr=subprocess.PIPE)
  94. self.assertEqual(["Running: echo stdout; echo stderr >&2", "stdout"], log.info_msgs)
  95. self.assertEqual(["stderr"], log.error_msgs)