packages/Python/lldbsuite/test/tools/lldb-server/TestGdbRemote_qThreadStopInfo.py

   1 from __future__ import print_function
   2
   3
   4
   5 import sys
   6
   7 import unittest2
   8 import gdbremote_testcase
   9 from lldbsuite.test.decorators import *
  10 from lldbsuite.test.lldbtest import *
  11 from lldbsuite.test import lldbutil
  12
  13 class TestGdbRemote_qThreadStopInfo(gdbremote_testcase.GdbRemoteTestCaseBase):
  14
  15     mydir = TestBase.compute_mydir(__file__)
  16     THREAD_COUNT = 5
  17
  18     def gather_stop_replies_via_qThreadStopInfo(self, thread_count):
  19         # Set up the inferior args.
  20         inferior_args=[]
  21         for i in range(thread_count - 1):
  22             inferior_args.append("thread:new")
  23         inferior_args.append("sleep:10")
  24         procs = self.prep_debug_monitor_and_inferior(inferior_args=inferior_args)
  25
  26         # Assumes test_sequence has anything added needed to setup the initial state.
  27         # (Like optionally enabling QThreadsInStopReply.)
  28         self.test_sequence.add_log_lines([
  29             "read packet: $c#63"
  30             ], True)
  31         context = self.expect_gdbremote_sequence()
  32         self.assertIsNotNone(context)
  33
  34         # Give threads time to start up, then break.
  35         time.sleep(1)
  36         self.reset_test_sequence()
  37         self.test_sequence.add_log_lines([
  38             "read packet: {}".format(chr(3)),
  39             {"direction":"send", "regex":r"^\$T([0-9a-fA-F]+)([^#]+)#[0-9a-fA-F]{2}$", "capture":{1:"stop_result", 2:"key_vals_text"} },
  40             ], True)
  41         context = self.expect_gdbremote_sequence()
  42         self.assertIsNotNone(context)
  43
  44         # Wait until all threads have started.
  45         threads = self.wait_for_thread_count(thread_count, timeout_seconds=3)
  46         self.assertIsNotNone(threads)
  47         self.assertEqual(len(threads), thread_count)
  48
  49         # Grab stop reply for each thread via qThreadStopInfo{tid:hex}.
  50         stop_replies = {}
  51         thread_dicts = {}
  52         for thread in threads:
  53             # Run the qThreadStopInfo command.
  54             self.reset_test_sequence()
  55             self.test_sequence.add_log_lines([
  56                 "read packet: $qThreadStopInfo{:x}#00".format(thread),
  57                 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]+)([^#]+)#[0-9a-fA-F]{2}$", "capture":{1:"stop_result", 2:"key_vals_text"} },
  58                 ], True)
  59             context = self.expect_gdbremote_sequence()
  60             self.assertIsNotNone(context)
  61
  62             # Parse stop reply contents.
  63             key_vals_text = context.get("key_vals_text")
  64             self.assertIsNotNone(key_vals_text)
  65             kv_dict = self.parse_key_val_dict(key_vals_text)
  66             self.assertIsNotNone(kv_dict)
  67
  68             # Verify there is a thread and that it matches the expected thread id.
  69             kv_thread = kv_dict.get("thread")
  70             self.assertIsNotNone(kv_thread)
  71             kv_thread_id = int(kv_thread, 16)
  72             self.assertEqual(kv_thread_id, thread)
  73
  74             # Grab the stop id reported.
  75             stop_result_text = context.get("stop_result")
  76             self.assertIsNotNone(stop_result_text)
  77             stop_replies[kv_thread_id] = int(stop_result_text, 16)
  78
  79             # Hang on to the key-val dictionary for the thread.
  80             thread_dicts[kv_thread_id] = kv_dict
  81
  82         return (stop_replies, thread_dicts)
  83
  84     def qThreadStopInfo_works_for_multiple_threads(self, thread_count):
  85         (stop_replies, _) = self.gather_stop_replies_via_qThreadStopInfo(thread_count)
  86         self.assertEqual(len(stop_replies), thread_count)
  87
  88     @debugserver_test
  89     def test_qThreadStopInfo_works_for_multiple_threads_debugserver(self):
  90         self.init_debugserver_test()
  91         self.build()
  92         self.set_inferior_startup_launch()
  93         self.qThreadStopInfo_works_for_multiple_threads(self.THREAD_COUNT)
  94
  95     @llgs_test
  96     def test_qThreadStopInfo_works_for_multiple_threads_llgs(self):
  97         self.init_llgs_test()
  98         self.build()
  99         self.set_inferior_startup_launch()
 100         self.qThreadStopInfo_works_for_multiple_threads(self.THREAD_COUNT)
 101
 102     def qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self, thread_count):
 103         (stop_replies, _) = self.gather_stop_replies_via_qThreadStopInfo(thread_count)
 104         self.assertIsNotNone(stop_replies)
 105
 106         no_stop_reason_count   = sum(1 for stop_reason in list(stop_replies.values()) if stop_reason == 0)
 107         with_stop_reason_count = sum(1 for stop_reason in list(stop_replies.values()) if stop_reason != 0)
 108
 109         # All but one thread should report no stop reason.
 110         self.assertEqual(no_stop_reason_count, thread_count - 1)
 111
 112         # Only one thread should should indicate a stop reason.
 113         self.assertEqual(with_stop_reason_count, 1)
 114
 115     @debugserver_test
 116     def test_qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt_debugserver(self):
 117         self.init_debugserver_test()
 118         self.build()
 119         self.set_inferior_startup_launch()
 120         self.qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self.THREAD_COUNT)
 121
 122     @llgs_test
 123     def test_qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt_llgs(self):
 124         self.init_llgs_test()
 125         self.build()
 126         self.set_inferior_startup_launch()
 127         self.qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self.THREAD_COUNT)
 128
 129     def qThreadStopInfo_has_valid_thread_names(self, thread_count, expected_thread_name):
 130         (_, thread_dicts) = self.gather_stop_replies_via_qThreadStopInfo(thread_count)
 131         self.assertIsNotNone(thread_dicts)
 132
 133         for thread_dict in list(thread_dicts.values()):
 134             name = thread_dict.get("name")
 135             self.assertIsNotNone(name)
 136             self.assertEqual(name, expected_thread_name)
 137
 138     @unittest2.skip("MacOSX doesn't have a default thread name")
 139     @debugserver_test
 140     def test_qThreadStopInfo_has_valid_thread_names_debugserver(self):
 141         self.init_debugserver_test()
 142         self.build()
 143         self.set_inferior_startup_launch()
 144         self.qThreadStopInfo_has_valid_thread_names(self.THREAD_COUNT, "a.out")
 145
 146     @skipUnlessPlatform(["linux"]) # test requires OS with set, equal thread names by default.
 147     @llgs_test
 148     def test_qThreadStopInfo_has_valid_thread_names_llgs(self):
 149         self.init_llgs_test()
 150         self.build()
 151         self.set_inferior_startup_launch()
 152         self.qThreadStopInfo_has_valid_thread_names(self.THREAD_COUNT, "a.out")