path: root/lldb/test/API/tools/lldb-server/TestGdbRemoteFork.py

import random

from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *

from fork_testbase import GdbRemoteForkTestBase


class TestGdbRemoteFork(GdbRemoteForkTestBase):
    def setUp(self):
        GdbRemoteForkTestBase.setUp(self)
        if self.getPlatform() == "linux" and self.getArchitecture() in ['arm', 'aarch64']:
            self.skipTest("Unsupported for Arm/AArch64 Linux")

    @add_test_categories(["fork"])
    def test_fork_multithreaded(self):
        _, _, child_pid, _ = self.start_fork_test(["thread:new"]*2 + ["fork"])

        # detach the forked child
        self.test_sequence.add_log_lines([
            "read packet: $D;{}#00".format(child_pid),
            "send packet: $OK#00",
            "read packet: $k#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_fork(self):
        parent_pid, _ = self.fork_and_detach_test("fork")

        # resume the parent
        self.test_sequence.add_log_lines([
            "read packet: $c#00",
            "send packet: $W00;process:{}#00".format(parent_pid),
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_vfork(self):
        parent_pid, parent_tid = self.fork_and_detach_test("vfork")

        # resume the parent
        self.test_sequence.add_log_lines([
            "read packet: $c#00",
            {"direction": "send",
             "regex": r"[$]T[0-9a-fA-F]{{2}}thread:p{}[.]{}.*vforkdone.*"
                      .format(parent_pid, parent_tid),
             },
            "read packet: $c#00",
            "send packet: $W00;process:{}#00".format(parent_pid),
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_fork_follow(self):
        self.fork_and_follow_test("fork")

    @add_test_categories(["fork"])
    def test_vfork_follow(self):
        self.fork_and_follow_test("vfork")

    @add_test_categories(["fork"])
    def test_select_wrong_pid(self):
        self.build()
        self.prep_debug_monitor_and_inferior()
        self.add_qSupported_packets(["multiprocess+"])
        ret = self.expect_gdbremote_sequence()
        self.assertIn("multiprocess+", ret["qSupported_response"])
        self.reset_test_sequence()

        # get process pid
        self.test_sequence.add_log_lines([
            "read packet: $qC#00",
            {"direction": "send", "regex": "[$]QCp([0-9a-f]+).([0-9a-f]+)#.*",
             "capture": {1: "pid", 2: "tid"}},
        ], True)
        ret = self.expect_gdbremote_sequence()
        pid, tid = (int(ret[x], 16) for x in ("pid", "tid"))
        self.reset_test_sequence()

        self.test_sequence.add_log_lines([
            # try switching to correct pid
            "read packet: $Hgp{:x}.{:x}#00".format(pid, tid),
            "send packet: $OK#00",
            "read packet: $Hcp{:x}.{:x}#00".format(pid, tid),
            "send packet: $OK#00",
            # try switching to invalid tid
            "read packet: $Hgp{:x}.{:x}#00".format(pid, tid+1),
            "send packet: $E15#00",
            "read packet: $Hcp{:x}.{:x}#00".format(pid, tid+1),
            "send packet: $E15#00",
            # try switching to invalid pid
            "read packet: $Hgp{:x}.{:x}#00".format(pid+1, tid),
            "send packet: $Eff#00",
            "read packet: $Hcp{:x}.{:x}#00".format(pid+1, tid),
            "send packet: $Eff#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_detach_current(self):
        self.build()
        self.prep_debug_monitor_and_inferior()
        self.add_qSupported_packets(["multiprocess+"])
        ret = self.expect_gdbremote_sequence()
        self.assertIn("multiprocess+", ret["qSupported_response"])
        self.reset_test_sequence()

        # get process pid
        self.test_sequence.add_log_lines([
            "read packet: $qC#00",
            {"direction": "send", "regex": "[$]QCp([0-9a-f]+).[0-9a-f]+#.*",
             "capture": {1: "pid"}},
        ], True)
        ret = self.expect_gdbremote_sequence()
        pid = ret["pid"]
        self.reset_test_sequence()

        # detach the process
        self.test_sequence.add_log_lines([
            "read packet: $D;{}#00".format(pid),
            "send packet: $OK#00",
            "read packet: $qC#00",
            "send packet: $E44#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_detach_all(self):
        self.detach_all_test()

    @add_test_categories(["fork"])
    def test_kill_all(self):
        parent_pid, _, child_pid, _ = self.start_fork_test(["fork"])

        exit_regex = "[$]X09;process:([0-9a-f]+)#.*"
        self.test_sequence.add_log_lines([
            # kill all processes
            "read packet: $k#00",
            {"direction": "send", "regex": exit_regex,
             "capture": {1: "pid1"}},
            {"direction": "send", "regex": exit_regex,
             "capture": {1: "pid2"}},
        ], True)
        ret = self.expect_gdbremote_sequence()
        self.assertEqual(set([ret["pid1"], ret["pid2"]]),
                         set([parent_pid, child_pid]))

    @add_test_categories(["fork"])
    def test_vkill_child(self):
        self.vkill_test(kill_child=True)

    @add_test_categories(["fork"])
    def test_vkill_parent(self):
        self.vkill_test(kill_parent=True)

    @add_test_categories(["fork"])
    def test_vkill_both(self):
        self.vkill_test(kill_parent=True, kill_child=True)

    @add_test_categories(["fork"])
    def test_c_parent(self):
        self.resume_one_test(run_order=["parent", "parent"])

    @add_test_categories(["fork"])
    def test_c_child(self):
        self.resume_one_test(run_order=["child", "child"])

    @add_test_categories(["fork"])
    def test_c_parent_then_child(self):
        self.resume_one_test(run_order=["parent", "parent", "child", "child"])

    @add_test_categories(["fork"])
    def test_c_child_then_parent(self):
        self.resume_one_test(run_order=["child", "child", "parent", "parent"])

    @add_test_categories(["fork"])
    def test_c_interspersed(self):
        self.resume_one_test(run_order=["parent", "child", "parent", "child"])

    @add_test_categories(["fork"])
    def test_vCont_parent(self):
        self.resume_one_test(run_order=["parent", "parent"], use_vCont=True)

    @add_test_categories(["fork"])
    def test_vCont_child(self):
        self.resume_one_test(run_order=["child", "child"], use_vCont=True)

    @add_test_categories(["fork"])
    def test_vCont_parent_then_child(self):
        self.resume_one_test(run_order=["parent", "parent", "child", "child"],
                             use_vCont=True)

    @add_test_categories(["fork"])
    def test_vCont_child_then_parent(self):
        self.resume_one_test(run_order=["child", "child", "parent", "parent"],
                             use_vCont=True)

    @add_test_categories(["fork"])
    def test_vCont_interspersed(self):
        self.resume_one_test(run_order=["parent", "child", "parent", "child"],
                             use_vCont=True)

    @add_test_categories(["fork"])
    def test_vCont_two_processes(self):
        parent_pid, parent_tid, child_pid, child_tid = (
            self.start_fork_test(["fork", "stop"]))

        self.test_sequence.add_log_lines([
            # try to resume both processes
            "read packet: $vCont;c:p{}.{};c:p{}.{}#00".format(
                parent_pid, parent_tid, child_pid, child_tid),
            "send packet: $E03#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_vCont_all_processes_explicit(self):
        self.start_fork_test(["fork", "stop"])

        self.test_sequence.add_log_lines([
            # try to resume all processes implicitly
            "read packet: $vCont;c:p-1.-1#00",
            "send packet: $E03#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_vCont_all_processes_implicit(self):
        self.start_fork_test(["fork", "stop"])

        self.test_sequence.add_log_lines([
            # try to resume all processes implicitly
            "read packet: $vCont;c#00",
            "send packet: $E03#00",
        ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_threadinfo(self):
        parent_pid, parent_tid, child_pid, child_tid = (
            self.start_fork_test(["fork", "thread:new", "stop"]))
        pidtids = [
            (parent_pid, parent_tid),
            (child_pid, child_tid),
        ]

        self.add_threadinfo_collection_packets()
        ret = self.expect_gdbremote_sequence()
        prev_pidtids = set(self.parse_threadinfo_packets(ret))
        self.assertEqual(prev_pidtids,
                         frozenset((int(pid, 16), int(tid, 16))
                                   for pid, tid in pidtids))
        self.reset_test_sequence()

        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hcp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $c#00",
                 {"direction": "send",
                  "regex": self.stop_regex.format(*pidtid),
                  },
                 ], True)
            self.add_threadinfo_collection_packets()
            ret = self.expect_gdbremote_sequence()
            self.reset_test_sequence()
            new_pidtids = set(self.parse_threadinfo_packets(ret))
            added_pidtid = new_pidtids - prev_pidtids
            prev_pidtids = new_pidtids

            # verify that we've got exactly one new thread, and that
            # the PID matches
            self.assertEqual(len(added_pidtid), 1)
            self.assertEqual(added_pidtid.pop()[0], int(pidtid[0], 16))

        for pidtid in new_pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_memory_read_write(self):
        self.build()
        INITIAL_DATA = "Initial message"
        self.prep_debug_monitor_and_inferior(
            inferior_args=["set-message:{}".format(INITIAL_DATA),
                           "get-data-address-hex:g_message",
                           "fork",
                           "print-message:",
                           "stop",
                           ])
        self.add_qSupported_packets(["multiprocess+",
                                     "fork-events+"])
        ret = self.expect_gdbremote_sequence()
        self.assertIn("fork-events+", ret["qSupported_response"])
        self.reset_test_sequence()

        # continue and expect fork
        self.test_sequence.add_log_lines([
            "read packet: $c#00",
            {"type": "output_match",
             "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"),
             "capture": {1: "addr"}},
            {"direction": "send", "regex": self.fork_regex.format("fork"),
             "capture": self.fork_capture},
        ], True)
        ret = self.expect_gdbremote_sequence()
        pidtids = {
            "parent": (ret["parent_pid"], ret["parent_tid"]),
            "child": (ret["child_pid"], ret["child_tid"]),
        }
        addr = ret["addr"]
        self.reset_test_sequence()

        for name, pidtid in pidtids.items():
            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 # read the current memory contents
                 "read packet: $m{},{:x}#00".format(addr,
                                                    len(INITIAL_DATA) + 1),
                 {"direction": "send",
                  "regex": r"^[$](.+)#.*$",
                  "capture": {1: "data"}},
                 # write a new value
                 "read packet: $M{},{:x}:{}#00".format(addr,
                                                       len(name) + 1,
                                                       seven.hexlify(
                                                           name + "\0")),
                 "send packet: $OK#00",
                 # resume the process and wait for the trap
                 "read packet: $Hcp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $c#00",
                 {"type": "output_match",
                  "regex": self.maybe_strict_output_regex(r"message: (.*)\r\n"),
                  "capture": {1: "printed_message"}},
                 {"direction": "send",
                  "regex": self.stop_regex.format(*pidtid),
                  },
                 ], True)
            ret = self.expect_gdbremote_sequence()
            data = seven.unhexlify(ret["data"])
            self.assertEqual(data, INITIAL_DATA + "\0")
            self.assertEqual(ret["printed_message"], name);
            self.reset_test_sequence()

        # we do the second round separately to make sure that initial data
        # is correctly preserved while writing into the first process

        for name, pidtid in pidtids.items():
            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 # read the current memory contents
                 "read packet: $m{},{:x}#00".format(addr,
                                                    len(name) + 1),
                 {"direction": "send",
                  "regex": r"^[$](.+)#.*$",
                  "capture": {1: "data"}},
                 ], True)
            ret = self.expect_gdbremote_sequence()
            self.assertIsNotNone(ret.get("data"))
            data = seven.unhexlify(ret.get("data"))
            self.assertEqual(data, name + "\0")
            self.reset_test_sequence()

    @add_test_categories(["fork"])
    def test_register_read_write(self):
        parent_pid, parent_tid, child_pid, child_tid = (
            self.start_fork_test(["fork", "thread:new", "stop"]))
        pidtids = [
            (parent_pid, parent_tid),
            (child_pid, child_tid),
        ]

        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hcp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $c#00",
                 {"direction": "send",
                  "regex": self.stop_regex.format(*pidtid),
                  },
                 ], True)

        self.add_threadinfo_collection_packets()
        ret = self.expect_gdbremote_sequence()
        self.reset_test_sequence()

        pidtids = set(self.parse_threadinfo_packets(ret))
        self.assertEqual(len(pidtids), 4)
        # first, save register values from all the threads
        thread_regs = {}
        for pidtid in pidtids:
            for regno in range(256):
                self.test_sequence.add_log_lines(
                    ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                     "send packet: $OK#00",
                     "read packet: $p{:x}#00".format(regno),
                     {"direction": "send",
                      "regex": r"^[$](.+)#.*$",
                      "capture": {1: "data"}},
                     ], True)
                ret = self.expect_gdbremote_sequence()
                data = ret.get("data")
                self.assertIsNotNone(data)
                # ignore registers shorter than 32 bits (this also catches
                # "Exx" errors)
                if len(data) >= 8:
                    break
            else:
                self.skipTest("no usable register found")
            thread_regs[pidtid] = (regno, data)

        vals = set(x[1] for x in thread_regs.values())
        # NB: cheap hack to make the loop below easier
        new_val = next(iter(vals))

        # then, start altering them and verify that we don't unexpectedly
        # change the value from another thread
        for pidtid in pidtids:
            old_val = thread_regs[pidtid]
            regno = old_val[0]
            old_val_length = len(old_val[1])
            # generate a unique new_val
            while new_val in vals:
                new_val = ('{{:0{}x}}'.format(old_val_length)
                           .format(random.getrandbits(old_val_length*4)))
            vals.add(new_val)

            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $p{:x}#00".format(regno),
                 {"direction": "send",
                  "regex": r"^[$](.+)#.*$",
                  "capture": {1: "data"}},
                 "read packet: $P{:x}={}#00".format(regno, new_val),
                 "send packet: $OK#00",
                 ], True)
            ret = self.expect_gdbremote_sequence()
            data = ret.get("data")
            self.assertIsNotNone(data)
            self.assertEqual(data, old_val[1])
            thread_regs[pidtid] = (regno, new_val)

        # finally, verify that new values took effect
        for pidtid in pidtids:
            old_val = thread_regs[pidtid]
            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $p{:x}#00".format(old_val[0]),
                 {"direction": "send",
                  "regex": r"^[$](.+)#.*$",
                  "capture": {1: "data"}},
                 ], True)
            ret = self.expect_gdbremote_sequence()
            data = ret.get("data")
            self.assertIsNotNone(data)
            self.assertEqual(data, old_val[1])

    @add_test_categories(["fork"])
    def test_qC(self):
        parent_pid, parent_tid, child_pid, child_tid = (
            self.start_fork_test(["fork", "thread:new", "stop"]))
        pidtids = [
            (parent_pid, parent_tid),
            (child_pid, child_tid),
        ]

        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hcp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $c#00",
                 {"direction": "send",
                  "regex": self.stop_regex.format(*pidtid),
                  },
                 ], True)

        self.add_threadinfo_collection_packets()
        ret = self.expect_gdbremote_sequence()
        self.reset_test_sequence()

        pidtids = set(self.parse_threadinfo_packets(ret))
        self.assertEqual(len(pidtids), 4)
        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $qC#00",
                 "send packet: $QCp{:x}.{:x}#00".format(*pidtid),
                 ], True)
        self.expect_gdbremote_sequence()

    @add_test_categories(["fork"])
    def test_T(self):
        parent_pid, parent_tid, child_pid, child_tid = (
            self.start_fork_test(["fork", "thread:new", "stop"]))
        pidtids = [
            (parent_pid, parent_tid),
            (child_pid, child_tid),
        ]

        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Hcp{}.{}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $c#00",
                 {"direction": "send",
                  "regex": self.stop_regex.format(*pidtid),
                  },
                 ], True)

        self.add_threadinfo_collection_packets()
        ret = self.expect_gdbremote_sequence()
        self.reset_test_sequence()

        pidtids = set(self.parse_threadinfo_packets(ret))
        self.assertEqual(len(pidtids), 4)
        max_pid = max(pid for pid, tid in pidtids)
        max_tid = max(tid for pid, tid in pidtids)
        bad_pidtids = (
            (max_pid, max_tid + 1, "E02"),
            (max_pid + 1, max_tid, "E01"),
            (max_pid + 1, max_tid + 1, "E01"),
        )

        for pidtid in pidtids:
            self.test_sequence.add_log_lines(
                [
                 # test explicit PID+TID
                 "read packet: $Tp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 # test implicit PID via Hg
                 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
                 "send packet: $OK#00",
                 "read packet: $T{:x}#00".format(max_tid + 1),
                 "send packet: $E02#00",
                 "read packet: $T{:x}#00".format(pidtid[1]),
                 "send packet: $OK#00",
                 ], True)
        for pid, tid, expected in bad_pidtids:
            self.test_sequence.add_log_lines(
                ["read packet: $Tp{:x}.{:x}#00".format(pid, tid),
                 "send packet: ${}#00".format(expected),
                 ], True)
        self.expect_gdbremote_sequence()