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