packages/Python/lldbsuite/pre_kill_hook/tests/test_linux.py

   1 """Test the pre-kill hook on Linux."""
   2 from __future__ import print_function
   3
   4 # system imports
   5 from multiprocessing import Process, Queue
   6 import platform
   7 import re
   8 import subprocess
   9 from unittest import main, TestCase
  10
  11 # third party
  12 from six import StringIO
  13
  14
  15 def do_child_thread():
  16     import os
  17     x = 0
  18     while True:
  19         x = x + 42 * os.getpid()
  20     return x
  21
  22
  23 def do_child_process(child_work_queue, parent_work_queue, verbose):
  24     import os
  25
  26     pid = os.getpid()
  27     if verbose:
  28         print("child: pid {} started, sending to parent".format(pid))
  29     parent_work_queue.put(pid)
  30
  31     # Spin up a daemon thread to do some "work", which will show
  32     # up in a sample of this process.
  33     import threading
  34     worker = threading.Thread(target=do_child_thread)
  35     worker.daemon = True
  36     worker.start()
  37
  38     if verbose:
  39         print("child: waiting for shut-down request from parent")
  40     child_work_queue.get()
  41     if verbose:
  42         print("child: received shut-down request.  Child exiting.")
  43
  44
  45 class LinuxPreKillTestCase(TestCase):
  46
  47     def __init__(self, methodName):
  48         super(LinuxPreKillTestCase, self).__init__(methodName)
  49         self.process = None
  50         self.child_work_queue = None
  51         self.verbose = False
  52         # self.verbose = True
  53
  54     def tearDown(self):
  55         if self.verbose:
  56             print("parent: sending shut-down request to child")
  57         if self.process:
  58             self.child_work_queue.put("hello, child")
  59             self.process.join()
  60         if self.verbose:
  61             print("parent: child is fully shut down")
  62
  63     def test_sample(self):
  64         # Ensure we're Darwin.
  65         if platform.system() != 'Linux':
  66             self.skipTest("requires a Linux-based OS")
  67
  68         # Ensure we have the 'perf' tool.  If not, skip the test.
  69         try:
  70             perf_version = subprocess.check_output(["perf", "version"])
  71             if perf_version is None or not (
  72                     perf_version.startswith("perf version")):
  73                 raise Exception("The perf executable doesn't appear"
  74                                 " to be the Linux perf tools perf")
  75         except Exception:
  76             self.skipTest("requires the Linux perf tools 'perf' command")
  77
  78         # Start the child process.
  79         self.child_work_queue = Queue()
  80         parent_work_queue = Queue()
  81         self.process = Process(target=do_child_process,
  82                                args=(self.child_work_queue, parent_work_queue,
  83                                      self.verbose))
  84         if self.verbose:
  85             print("parent: starting child")
  86         self.process.start()
  87
  88         # Wait for the child to report its pid.  Then we know we're running.
  89         if self.verbose:
  90             print("parent: waiting for child to start")
  91         child_pid = parent_work_queue.get()
  92
  93         # Sample the child process.
  94         from linux import do_pre_kill
  95         context_dict = {
  96             "archs": [platform.machine()],
  97             "platform_name": None,
  98             "platform_url": None,
  99             "platform_working_dir": None
 100         }
 101
 102         if self.verbose:
 103             print("parent: running pre-kill action on child")
 104         output_io = StringIO()
 105         do_pre_kill(child_pid, context_dict, output_io)
 106         output = output_io.getvalue()
 107
 108         if self.verbose:
 109             print("parent: do_pre_kill() wrote the following output:", output)
 110         self.assertIsNotNone(output)
 111
 112         # We should have a samples count entry.
 113         # Samples:
 114         self.assertTrue("Samples:" in output, "should have found a 'Samples:' "
 115                         "field in the sampled process output")
 116
 117         # We should see an event count entry
 118         event_count_re = re.compile(r"Event count[^:]+:\s+(\d+)")
 119         match = event_count_re.search(output)
 120         self.assertIsNotNone(match, "should have found the event count entry "
 121                              "in sample output")
 122         if self.verbose:
 123             print("cpu-clock events:", match.group(1))
 124
 125         # We should see some percentages in the file.
 126         percentage_re = re.compile(r"\d+\.\d+%")
 127         match = percentage_re.search(output)
 128         self.assertIsNotNone(match, "should have found at least one percentage "
 129                              "in the sample output")
 130
 131
 132 if __name__ == "__main__":
 133     main()