test_env_unittest.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. #!/usr/bin/env python3
  2. # Copyright (c) 2019 The Chromium Authors. All rights reserved.
  3. # Use of this source code is governed by a BSD-style license that can be
  4. # found in the LICENSE file.
  5. """Unit tests for test_env.py functionality.
  6. Each unit test is launches python process that uses test_env.py
  7. to launch another python process. Then signal handling and
  8. propagation is tested. This similates how Swarming uses test_env.py.
  9. """
  10. import os
  11. import signal
  12. import subprocess
  13. import sys
  14. import time
  15. import unittest
  16. HERE = os.path.dirname(os.path.abspath(__file__))
  17. TEST_SCRIPT = os.path.join(HERE, 'test_env_user_script.py')
  18. def launch_process_windows(args):
  19. # The `universal_newlines` option is equivalent to `text` in Python 3.
  20. return subprocess.Popen(
  21. [sys.executable, TEST_SCRIPT] + args,
  22. stdout=subprocess.PIPE,
  23. stderr=subprocess.STDOUT,
  24. env=os.environ.copy(),
  25. creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
  26. universal_newlines=True)
  27. def launch_process_nonwindows(args):
  28. # The `universal_newlines` option is equivalent to `text` in Python 3.
  29. return subprocess.Popen(
  30. [sys.executable, TEST_SCRIPT] + args,
  31. stdout=subprocess.PIPE,
  32. stderr=subprocess.STDOUT,
  33. env=os.environ.copy(),
  34. universal_newlines=True)
  35. # pylint: disable=inconsistent-return-statements
  36. def read_subprocess_message(proc, starts_with):
  37. """Finds the value after first line prefix condition."""
  38. for line in proc.stdout:
  39. if line.startswith(starts_with):
  40. return line.rstrip().replace(starts_with, '')
  41. # pylint: enable=inconsistent-return-statements
  42. def send_and_wait(proc, sig, sleep_time=0.6):
  43. """Sends a signal to subprocess."""
  44. time.sleep(sleep_time) # gives process time to launch.
  45. os.kill(proc.pid, sig)
  46. proc.wait()
  47. class SignalingWindowsTest(unittest.TestCase):
  48. def setUp(self):
  49. super().setUp()
  50. if sys.platform != 'win32':
  51. self.skipTest('test only runs on Windows')
  52. def test_send_ctrl_break_event(self):
  53. proc = launch_process_windows([])
  54. send_and_wait(proc, signal.CTRL_BREAK_EVENT) # pylint: disable=no-member
  55. sig = read_subprocess_message(proc, 'Signal :')
  56. # This test is flaky because it relies on the child process starting quickly
  57. # "enough", which it fails to do sometimes. This is tracked by
  58. # https://crbug.com/1335123 and it is hoped that increasing the timeout will
  59. # reduce the flakiness.
  60. self.assertEqual(sig, str(int(signal.SIGBREAK))) # pylint: disable=no-member
  61. class SignalingNonWindowsTest(unittest.TestCase):
  62. def setUp(self):
  63. super().setUp()
  64. if sys.platform == 'win32':
  65. self.skipTest('test does not run on Windows')
  66. def test_send_sigterm(self):
  67. proc = launch_process_nonwindows([])
  68. send_and_wait(proc, signal.SIGTERM)
  69. sig = read_subprocess_message(proc, 'Signal :')
  70. self.assertEqual(sig, str(int(signal.SIGTERM)))
  71. def test_send_sigint(self):
  72. proc = launch_process_nonwindows([])
  73. send_and_wait(proc, signal.SIGINT)
  74. sig = read_subprocess_message(proc, 'Signal :')
  75. self.assertEqual(sig, str(int(signal.SIGINT)))
  76. if __name__ == '__main__':
  77. unittest.main()