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