1import re
2
3import gdbremote_testcase
4from lldbsuite.test.decorators import *
5from lldbsuite.test.lldbtest import *
6
7
8class TestPartialResume(gdbremote_testcase.GdbRemoteTestCaseBase):
9    THREAD_MATCH_RE = re.compile(r"thread ([0-9a-f]+) running")
10
11    def start_vCont_run_subset_of_threads_test(self):
12        self.build()
13        self.set_inferior_startup_launch()
14
15        procs = self.prep_debug_monitor_and_inferior(inferior_args=["3"])
16        # grab the main thread id
17        self.add_threadinfo_collection_packets()
18        main_thread = self.parse_threadinfo_packets(
19            self.expect_gdbremote_sequence())
20        self.assertEqual(len(main_thread), 1)
21        self.reset_test_sequence()
22
23        # run until threads start, then grab full thread list
24        self.test_sequence.add_log_lines([
25            "read packet: $c#63",
26            {"direction": "send", "regex": "[$]T.*;reason:signal.*"},
27        ], True)
28        self.add_threadinfo_collection_packets()
29
30        all_threads = self.parse_threadinfo_packets(
31            self.expect_gdbremote_sequence())
32        self.assertEqual(len(all_threads), 4)
33        self.assertIn(main_thread[0], all_threads)
34        self.reset_test_sequence()
35
36        all_subthreads = set(all_threads) - set(main_thread)
37        self.assertEqual(len(all_subthreads), 3)
38
39        return (main_thread[0], list(all_subthreads))
40
41    def continue_and_get_threads_running(self, main_thread, vCont_req):
42        self.test_sequence.add_log_lines(
43            ["read packet: $vCont;c:{:x};{}#00".format(main_thread, vCont_req),
44             "send packet: $W00#00",
45             ], True)
46        exp = self.expect_gdbremote_sequence()
47        self.reset_test_sequence()
48        found = set()
49        for line in exp["O_content"].decode().splitlines():
50            m = self.THREAD_MATCH_RE.match(line)
51            if m is not None:
52                found.add(int(m.group(1), 16))
53        return found
54
55    @skipIfWindows
56    @add_test_categories(["llgs"])
57    def test_vCont_cxcx(self):
58        main_thread, all_subthreads_list = (
59            self.start_vCont_run_subset_of_threads_test())
60        # resume two threads explicitly, stop the third one implicitly
61        self.assertEqual(
62            self.continue_and_get_threads_running(
63                main_thread,
64                "c:{:x};c:{:x}".format(*all_subthreads_list[:2])),
65            set(all_subthreads_list[:2]))
66
67    @skipIfWindows
68    @add_test_categories(["llgs"])
69    def test_vCont_cxcxt(self):
70        main_thread, all_subthreads_list = (
71            self.start_vCont_run_subset_of_threads_test())
72        # resume two threads explicitly, stop others explicitly
73        self.assertEqual(
74            self.continue_and_get_threads_running(
75                main_thread,
76                "c:{:x};c:{:x};t".format(*all_subthreads_list[:2])),
77            set(all_subthreads_list[:2]))
78
79    @skipIfWindows
80    @add_test_categories(["llgs"])
81    def test_vCont_txc(self):
82        main_thread, all_subthreads_list = (
83            self.start_vCont_run_subset_of_threads_test())
84        # stop one thread explicitly, resume others
85        self.assertEqual(
86            self.continue_and_get_threads_running(
87                main_thread,
88                "t:{:x};c".format(all_subthreads_list[-1])),
89            set(all_subthreads_list[:2]))
90
91    @skipIfWindows
92    @add_test_categories(["llgs"])
93    def test_vCont_cxtxc(self):
94        main_thread, all_subthreads_list = (
95            self.start_vCont_run_subset_of_threads_test())
96        # resume one thread explicitly, stop one explicitly,
97        # resume others
98        self.assertEqual(
99            self.continue_and_get_threads_running(
100                main_thread,
101                "c:{:x};t:{:x};c".format(*all_subthreads_list[-2:])),
102            set(all_subthreads_list[:2]))
103
104    @skipIfWindows
105    @add_test_categories(["llgs"])
106    def test_vCont_txcx(self):
107        main_thread, all_subthreads_list = (
108            self.start_vCont_run_subset_of_threads_test())
109        # resume one thread explicitly, stop one explicitly,
110        # stop others implicitly
111        self.assertEqual(
112            self.continue_and_get_threads_running(
113                main_thread,
114                "t:{:x};c:{:x}".format(*all_subthreads_list[:2])),
115            set(all_subthreads_list[1:2]))
116
117    @skipIfWindows
118    @add_test_categories(["llgs"])
119    def test_vCont_txcxt(self):
120        main_thread, all_subthreads_list = (
121            self.start_vCont_run_subset_of_threads_test())
122        # resume one thread explicitly, stop one explicitly,
123        # stop others explicitly
124        self.assertEqual(
125            self.continue_and_get_threads_running(
126                main_thread,
127                "t:{:x};c:{:x};t".format(*all_subthreads_list[:2])),
128            set(all_subthreads_list[1:2]))
129