1import json
2import re
3
4import gdbremote_testcase
5from lldbsuite.test.decorators import *
6from lldbsuite.test.lldbtest import *
7from lldbsuite.test import lldbutil
8
9class TestGdbRemote_vContThreads(gdbremote_testcase.GdbRemoteTestCaseBase):
10    mydir = TestBase.compute_mydir(__file__)
11
12    def start_threads(self, num):
13        procs = self.prep_debug_monitor_and_inferior(inferior_args=[str(num)])
14        # start the process and wait for output
15        self.test_sequence.add_log_lines([
16            "read packet: $c#63",
17            {"type": "output_match", "regex": self.maybe_strict_output_regex(
18                r"@started\r\n")},
19        ], True)
20        # then interrupt it
21        self.add_interrupt_packets()
22        self.add_threadinfo_collection_packets()
23
24        context = self.expect_gdbremote_sequence()
25        self.assertIsNotNone(context)
26        threads = self.parse_threadinfo_packets(context)
27        self.assertIsNotNone(threads)
28        self.assertEqual(len(threads), num + 1)
29
30        self.reset_test_sequence()
31        return threads
32
33    def send_and_check_signal(self, vCont_data, threads):
34        self.test_sequence.add_log_lines([
35            "read packet: $vCont;{0}#00".format(vCont_data),
36            {"type": "output_match",
37             "regex": self.maybe_strict_output_regex(
38                 len(threads) *
39                 r"received SIGUSR1 on thread id: ([0-9a-f]+)\r\n"),
40             "capture": dict((i, "tid{0}".format(i)) for i
41                             in range(1, len(threads)+1)),
42             },
43        ], True)
44
45        context = self.expect_gdbremote_sequence()
46        self.assertIsNotNone(context)
47        tids = sorted(int(context["tid{0}".format(x)], 16)
48                      for x in range(1, len(threads)+1))
49        self.assertEqual(tids, sorted(threads))
50
51    def get_pid(self):
52        self.add_process_info_collection_packets()
53        context = self.expect_gdbremote_sequence()
54        self.assertIsNotNone(context)
55        procinfo = self.parse_process_info_response(context)
56        return int(procinfo['pid'], 16)
57
58    @skipIfWindows
59    @expectedFailureNetBSD
60    @expectedFailureDarwin # No signals delivered
61    @skipIfAsan # Times out under asan
62    def test_signal_process_without_tid(self):
63        self.build()
64        self.set_inferior_startup_launch()
65
66        threads = self.start_threads(1)
67        self.send_and_check_signal(
68            "C{0:x}".format(lldbutil.get_signal_number('SIGUSR1')),
69            threads)
70
71    @skipUnlessPlatform(["netbsd"])
72    @expectedFailureNetBSD
73    @skipIfAsan # Times out under asan
74    def test_signal_one_thread(self):
75        self.build()
76        self.set_inferior_startup_launch()
77
78        threads = self.start_threads(1)
79        # try sending a signal to one of the two threads
80        self.send_and_check_signal(
81            "C{0:x}:{1:x};c".format(lldbutil.get_signal_number('SIGUSR1')),
82            threads[:1])
83
84    @skipIfWindows
85    @expectedFailureNetBSD
86    @expectedFailureDarwin # Only one signal delivered
87    @skipIfAsan # Times out under asan
88    def test_signal_all_threads(self):
89        self.build()
90        self.set_inferior_startup_launch()
91
92        threads = self.start_threads(1)
93        # try sending a signal to two threads (= the process)
94        self.send_and_check_signal(
95            "C{0:x}:{1:x};C{0:x}:{2:x}".format(
96                lldbutil.get_signal_number('SIGUSR1'),
97                *threads),
98            threads)
99
100    @skipIfWindows
101    @expectedFailureNetBSD
102    @add_test_categories(["llgs"])
103    @skipIfAsan # Times out under asan
104    def test_signal_process_by_pid(self):
105        self.build()
106        self.set_inferior_startup_launch()
107
108        threads = self.start_threads(1)
109        self.send_and_check_signal(
110            "C{0:x}:p{1:x}".format(
111                lldbutil.get_signal_number('SIGUSR1'),
112                self.get_pid()),
113            threads)
114
115    @skipIfWindows
116    @expectedFailureNetBSD
117    @add_test_categories(["llgs"])
118    @skipIfAsan # Times out under asan
119    def test_signal_process_minus_one(self):
120        self.build()
121        self.set_inferior_startup_launch()
122
123        threads = self.start_threads(1)
124        self.send_and_check_signal(
125            "C{0:x}:p-1".format(
126                lldbutil.get_signal_number('SIGUSR1')),
127            threads)
128
129    @skipIfWindows
130    @expectedFailureNetBSD
131    @add_test_categories(["llgs"])
132    def test_signal_minus_one(self):
133        self.build()
134        self.set_inferior_startup_launch()
135
136        threads = self.start_threads(1)
137        self.send_and_check_signal(
138            "C{0:x}:-1".format(lldbutil.get_signal_number('SIGUSR1')),
139            threads)
140
141    @skipIfWindows
142    @expectedFailureNetBSD
143    @add_test_categories(["llgs"])
144    @skipIfAsan # Times out under asan
145    def test_signal_all_threads_by_pid(self):
146        self.build()
147        self.set_inferior_startup_launch()
148
149        threads = self.start_threads(1)
150        # try sending a signal to two threads (= the process)
151        self.send_and_check_signal(
152            "C{0:x}:p{1:x}.{2:x};C{0:x}:p{1:x}.{3:x}".format(
153                lldbutil.get_signal_number('SIGUSR1'),
154                self.get_pid(),
155                *threads),
156            threads)
157
158    @skipIfWindows
159    @expectedFailureNetBSD
160    @add_test_categories(["llgs"])
161    @skipIfAsan # Times out under asan
162    def test_signal_minus_one_by_pid(self):
163        self.build()
164        self.set_inferior_startup_launch()
165
166        threads = self.start_threads(1)
167        self.send_and_check_signal(
168            "C{0:x}:p{1:x}.-1".format(
169                lldbutil.get_signal_number('SIGUSR1'),
170                self.get_pid()),
171            threads)
172
173    @skipIfWindows
174    @expectedFailureNetBSD
175    @add_test_categories(["llgs"])
176    @skipIfAsan # Times out under asan
177    def test_signal_minus_one_by_minus_one(self):
178        self.build()
179        self.set_inferior_startup_launch()
180
181        threads = self.start_threads(1)
182        self.send_and_check_signal(
183            "C{0:x}:p-1.-1".format(
184                lldbutil.get_signal_number('SIGUSR1')),
185            threads)
186
187    @skipUnlessPlatform(["netbsd"])
188    def test_signal_two_of_three_threads(self):
189        self.build()
190        self.set_inferior_startup_launch()
191
192        threads = self.start_threads(2)
193        # try sending a signal to 2 out of 3 threads
194        self.test_sequence.add_log_lines([
195            "read packet: $vCont;C{0:x}:{1:x};C{0:x}:{2:x};c#00".format(
196                lldbutil.get_signal_number('SIGUSR1'),
197                threads[1], threads[2]),
198            {"direction": "send", "regex": r"^\$E1e#db$"},
199        ], True)
200
201        context = self.expect_gdbremote_sequence()
202        self.assertIsNotNone(context)
203
204    @skipUnlessPlatform(["netbsd"])
205    def test_signal_two_signals(self):
206        self.build()
207        self.set_inferior_startup_launch()
208
209        threads = self.start_threads(1)
210        # try sending two different signals to two threads
211        self.test_sequence.add_log_lines([
212            "read packet: $vCont;C{0:x}:{1:x};C{2:x}:{3:x}#00".format(
213                lldbutil.get_signal_number('SIGUSR1'), threads[0],
214                lldbutil.get_signal_number('SIGUSR2'), threads[1]),
215            {"direction": "send", "regex": r"^\$E1e#db$"},
216        ], True)
217
218        context = self.expect_gdbremote_sequence()
219        self.assertIsNotNone(context)
220