File: TestPlatformClient.py

package info (click to toggle)
swiftlang 6.0.3-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,519,992 kB
  • sloc: cpp: 9,107,863; ansic: 2,040,022; asm: 1,135,751; python: 296,500; objc: 82,456; f90: 60,502; lisp: 34,951; pascal: 19,946; sh: 18,133; perl: 7,482; ml: 4,937; javascript: 4,117; makefile: 3,840; awk: 3,535; xml: 914; fortran: 619; cs: 573; ruby: 573
file content (144 lines) | stat: -rw-r--r-- 5,408 bytes parent folder | download | duplicates (10)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import lldb
import binascii
import os
import time
from lldbsuite.test.lldbtest import *
from lldbsuite.test.decorators import *
from lldbsuite.test.gdbclientutils import *
from lldbsuite.test.lldbgdbclient import GDBRemoteTestBase


def hexlify(string):
    return binascii.hexlify(string.encode()).decode()


class TestPlatformClient(GDBRemoteTestBase):
    def test_process_list_with_all_users(self):
        """Test connecting to a remote linux platform"""

        class MyResponder(MockGDBServerResponder):
            def __init__(self):
                MockGDBServerResponder.__init__(self)
                self.currentQsProc = 0
                self.all_users = False

            def qfProcessInfo(self, packet):
                if "all_users:1" in packet:
                    self.all_users = True
                    name = hexlify("/a/test_process")
                    args = "-".join(
                        map(
                            hexlify,
                            ["/system/bin/sh", "-c", "/data/local/tmp/lldb-server"],
                        )
                    )
                    return (
                        "pid:10;ppid:1;uid:2;gid:3;euid:4;egid:5;name:"
                        + name
                        + ";args:"
                        + args
                        + ";"
                    )
                else:
                    self.all_users = False
                    return "E04"

            def qsProcessInfo(self):
                if self.all_users:
                    if self.currentQsProc == 0:
                        self.currentQsProc = 1
                        name = hexlify("/b/another_test_process")
                        # This intentionally has a badly encoded argument
                        args = "X".join(map(hexlify, ["/system/bin/ls", "--help"]))
                        return (
                            "pid:11;ppid:2;uid:3;gid:4;euid:5;egid:6;name:"
                            + name
                            + ";args:"
                            + args
                            + ";"
                        )
                    elif self.currentQsProc == 1:
                        self.currentQsProc = 0
                        return "E04"
                else:
                    return "E04"

        self.server.responder = MyResponder()

        try:
            self.runCmd("platform select remote-linux")
            self.runCmd("platform connect " + self.server.get_connect_url())
            self.assertTrue(self.dbg.GetSelectedPlatform().IsConnected())
            self.expect(
                "platform process list -x",
                substrs=[
                    "2 matching processes were found",
                    "test_process",
                    "another_test_process",
                ],
            )
            self.expect(
                "platform process list -xv",
                substrs=[
                    "PID    PARENT USER       GROUP      EFF USER   EFF GROUP  TRIPLE                         ARGUMENTS",
                    "10     1      2          3          4          5                                         /system/bin/sh -c /data/local/tmp/lldb-server",
                    "11     2      3          4          5          6",
                ],
            )
            self.expect(
                "platform process list -xv", substrs=["/system/bin/ls"], matching=False
            )
            self.expect(
                "platform process list",
                error=True,
                substrs=[
                    'error: no processes were found on the "remote-linux" platform'
                ],
            )
        finally:
            self.dbg.GetSelectedPlatform().DisconnectRemote()

    class TimeoutResponder(MockGDBServerResponder):
        """A mock server, which takes a very long time to compute the working
        directory."""

        def __init__(self):
            MockGDBServerResponder.__init__(self)

        def qGetWorkingDir(self):
            time.sleep(10)
            return hexlify("/foo/bar")

    def test_no_timeout(self):
        """Test that we honor the timeout setting. With a large enough timeout,
        we should get the CWD successfully."""

        self.server.responder = TestPlatformClient.TimeoutResponder()
        self.runCmd("settings set plugin.process.gdb-remote.packet-timeout 30")
        plat = lldb.SBPlatform("remote-linux")
        try:
            self.assertSuccess(
                plat.ConnectRemote(
                    lldb.SBPlatformConnectOptions(self.server.get_connect_url())
                )
            )
            self.assertEqual(plat.GetWorkingDirectory(), "/foo/bar")
        finally:
            plat.DisconnectRemote()

    def test_timeout(self):
        """Test that we honor the timeout setting. With a small timeout, CWD
        retrieval should fail."""

        self.server.responder = TestPlatformClient.TimeoutResponder()
        self.runCmd("settings set plugin.process.gdb-remote.packet-timeout 3")
        plat = lldb.SBPlatform("remote-linux")
        try:
            self.assertSuccess(
                plat.ConnectRemote(
                    lldb.SBPlatformConnectOptions(self.server.get_connect_url())
                )
            )
            self.assertIsNone(plat.GetWorkingDirectory())
        finally:
            plat.DisconnectRemote()