1from __future__ import print_function
2
3from lldbsuite.test.lldbtest import *
4import os
5import vscode
6
7
8class VSCodeTestCaseBase(TestBase):
9
10    def create_debug_adaptor(self):
11        '''Create the Visual Studio Code debug adaptor'''
12        self.assertTrue(os.path.exists(self.lldbVSCodeExec),
13                        'lldb-vscode must exist')
14        self.vscode = vscode.DebugAdaptor(
15            executable=self.lldbVSCodeExec, init_commands=Base.setUpCommands())
16
17    def build_and_create_debug_adaptor(self):
18        self.build()
19        self.create_debug_adaptor()
20
21    def set_source_breakpoints(self, source_path, lines, condition=None,
22                               hitCondition=None):
23        '''Sets source breakpoints and returns an array of strings containing
24           the breakpoint location IDs ("1.1", "1.2") for each breakpoint
25           that was set.
26        '''
27        response = self.vscode.request_setBreakpoints(
28            source_path, lines, condition=condition, hitCondition=hitCondition)
29        if response is None:
30            return []
31        breakpoints = response['body']['breakpoints']
32        breakpoint_ids = []
33        for breakpoint in breakpoints:
34            response_id = breakpoint['id']
35            bp_id = response_id >> 32
36            bp_loc_id = response_id & 0xffffffff
37            breakpoint_ids.append('%i.%i' % (bp_id, bp_loc_id))
38        return breakpoint_ids
39
40    def set_function_breakpoints(self, functions, condition=None,
41                                 hitCondition=None):
42        '''Sets breakpoints by function name given an array of function names
43           and returns an array of strings containing the breakpoint location
44           IDs ("1.1", "1.2") for each breakpoint that was set.
45        '''
46        response = self.vscode.request_setFunctionBreakpoints(
47            functions, condition=condition, hitCondition=hitCondition)
48        if response is None:
49            return []
50        breakpoints = response['body']['breakpoints']
51        breakpoint_ids = []
52        for breakpoint in breakpoints:
53            response_id = breakpoint['id']
54            bp_id = response_id >> 32
55            bp_loc_id = response_id & 0xffffffff
56            breakpoint_ids.append('%i.%i' % (bp_id, bp_loc_id))
57        return breakpoint_ids
58
59    def verify_breakpoint_hit(self, breakpoint_ids):
60        '''Wait for the process we are debugging to stop, and verify we hit
61           any breakpoint location in the "breakpoint_ids" array.
62           "breakpoint_ids" should be a list of breakpoint location ID strings
63           (["1.1", "2.1"]). The return value from
64           self.set_source_breakpoints() can be passed to this function'''
65        stopped_events = self.vscode.wait_for_stopped()
66        for stopped_event in stopped_events:
67            if 'body' in stopped_event:
68                body = stopped_event['body']
69                if 'reason' not in body:
70                    continue
71                if body['reason'] != 'breakpoint':
72                    continue
73                if 'description' not in body:
74                    continue
75                # Description is "breakpoint 1.1", so look for any location id
76                # ("1.1") in the description field as verification that one of
77                # the breakpoint locations was hit
78                description = body['description']
79                for breakpoint_id in breakpoint_ids:
80                    if breakpoint_id in description:
81                        return True
82        return False
83
84    def verify_exception_breakpoint_hit(self, filter_label):
85        '''Wait for the process we are debugging to stop, and verify the stop
86           reason is 'exception' and that the description matches
87           'filter_label'
88        '''
89        stopped_events = self.vscode.wait_for_stopped()
90        for stopped_event in stopped_events:
91            if 'body' in stopped_event:
92                body = stopped_event['body']
93                if 'reason' not in body:
94                    continue
95                if body['reason'] != 'exception':
96                    continue
97                if 'description' not in body:
98                    continue
99                description = body['description']
100                if filter_label == description:
101                    return True
102        return False
103
104    def verify_commands(self, flavor, output, commands):
105        self.assertTrue(output and len(output) > 0, "expect console output")
106        lines = output.splitlines()
107        prefix = '(lldb) '
108        for cmd in commands:
109            found = False
110            for line in lines:
111                if line.startswith(prefix) and cmd in line:
112                    found = True
113                    break
114            self.assertTrue(found,
115                            "verify '%s' found in console output for '%s'" % (
116                                cmd, flavor))
117
118    def get_dict_value(self, d, key_path):
119        '''Verify each key in the key_path array is in contained in each
120           dictionary within "d". Assert if any key isn't in the
121           corresponding dictionary. This is handy for grabbing values from VS
122           Code response dictionary like getting
123           response['body']['stackFrames']
124        '''
125        value = d
126        for key in key_path:
127            if key in value:
128                value = value[key]
129            else:
130                self.assertTrue(key in value,
131                                'key "%s" from key_path "%s" not in "%s"' % (
132                                    key, key_path, d))
133        return value
134
135    def get_stackFrames(self, threadId=None, startFrame=None, levels=None,
136                        dump=False):
137        response = self.vscode.request_stackTrace(threadId=threadId,
138                                                  startFrame=startFrame,
139                                                  levels=levels,
140                                                  dump=dump)
141        if response:
142            return self.get_dict_value(response, ['body', 'stackFrames'])
143        return None
144
145    def get_source_and_line(self, threadId=None, frameIndex=0):
146        stackFrames = self.get_stackFrames(threadId=threadId,
147                                           startFrame=frameIndex,
148                                           levels=1)
149        if stackFrames is not None:
150            stackFrame = stackFrames[0]
151            ['source', 'path']
152            if 'source' in stackFrame:
153                source = stackFrame['source']
154                if 'path' in source:
155                    if 'line' in stackFrame:
156                        return (source['path'], stackFrame['line'])
157        return ('', 0)
158
159    def get_stdout(self, timeout=0.0):
160        return self.vscode.get_output('stdout', timeout=timeout)
161
162    def get_console(self, timeout=0.0):
163        return self.vscode.get_output('console', timeout=timeout)
164
165    def get_local_as_int(self, name, threadId=None):
166        value = self.vscode.get_local_variable_value(name, threadId=threadId)
167        if value.startswith('0x'):
168            return int(value, 16)
169        elif value.startswith('0'):
170            return int(value, 8)
171        else:
172            return int(value)
173
174    def set_local(self, name, value, id=None):
175        '''Set a top level local variable only.'''
176        return self.vscode.request_setVariable(1, name, str(value), id=id)
177
178    def set_global(self, name, value, id=None):
179        '''Set a top level global variable only.'''
180        return self.vscode.request_setVariable(2, name, str(value), id=id)
181
182    def stepIn(self, threadId=None, waitForStop=True):
183        self.vscode.request_stepIn(threadId=threadId)
184        if waitForStop:
185            return self.vscode.wait_for_stopped()
186        return None
187
188    def stepOver(self, threadId=None, waitForStop=True):
189        self.vscode.request_next(threadId=threadId)
190        if waitForStop:
191            return self.vscode.wait_for_stopped()
192        return None
193
194    def stepOut(self, threadId=None, waitForStop=True):
195        self.vscode.request_stepOut(threadId=threadId)
196        if waitForStop:
197            return self.vscode.wait_for_stopped()
198        return None
199
200    def continue_to_next_stop(self):
201        self.vscode.request_continue()
202        return self.vscode.wait_for_stopped()
203
204    def continue_to_breakpoints(self, breakpoint_ids):
205        self.vscode.request_continue()
206        self.verify_breakpoint_hit(breakpoint_ids)
207
208    def continue_to_exception_breakpoint(self, filter_label):
209        self.vscode.request_continue()
210        self.assertTrue(self.verify_exception_breakpoint_hit(filter_label),
211                        'verify we got "%s"' % (filter_label))
212
213    def continue_to_exit(self, exitCode=0):
214        self.vscode.request_continue()
215        stopped_events = self.vscode.wait_for_stopped()
216        self.assertTrue(len(stopped_events) == 1,
217                        "expecting single 'exited' event")
218        self.assertTrue(stopped_events[0]['event'] == 'exited',
219                        'make sure program ran to completion')
220        self.assertTrue(stopped_events[0]['body']['exitCode'] == exitCode,
221                        'exitCode == %i' % (exitCode))
222
223    def attach(self, program=None, pid=None, waitFor=None, trace=None,
224               initCommands=None, preRunCommands=None, stopCommands=None,
225               exitCommands=None, attachCommands=None):
226        '''Build the default Makefile target, create the VSCode debug adaptor,
227           and attach to the process.
228        '''
229        # Make sure we disconnect and terminate the VSCode debug adaptor even
230        # if we throw an exception during the test case.
231        def cleanup():
232            self.vscode.request_disconnect(terminateDebuggee=True)
233            self.vscode.terminate()
234
235        # Execute the cleanup function during test case tear down.
236        self.addTearDownHook(cleanup)
237        # Initialize and launch the program
238        self.vscode.request_initialize()
239        response = self.vscode.request_attach(
240            program=program, pid=pid, waitFor=waitFor, trace=trace,
241            initCommands=initCommands, preRunCommands=preRunCommands,
242            stopCommands=stopCommands, exitCommands=exitCommands,
243            attachCommands=attachCommands)
244        if not (response and response['success']):
245            self.assertTrue(response['success'],
246                            'attach failed (%s)' % (response['message']))
247
248    def launch(self, program=None, args=None, cwd=None, env=None,
249               stopOnEntry=False, disableASLR=True,
250               disableSTDIO=False, shellExpandArguments=False,
251               trace=False, initCommands=None, preRunCommands=None,
252               stopCommands=None, exitCommands=None,sourcePath= None,
253               debuggerRoot=None, launchCommands=None):
254        '''Sending launch request to vscode
255        '''
256
257        # Make sure we disconnet and terminate the VSCode debug adaptor,
258        # if we throw an exception during the test case
259        def cleanup():
260            self.vscode.request_disconnect(terminateDebuggee=True)
261            self.vscode.terminate()
262
263        # Execute the cleanup function during test case tear down.
264        self.addTearDownHook(cleanup)
265
266        # Initialize and launch the program
267        self.vscode.request_initialize()
268        response = self.vscode.request_launch(
269            program,
270            args=args,
271            cwd=cwd,
272            env=env,
273            stopOnEntry=stopOnEntry,
274            disableASLR=disableASLR,
275            disableSTDIO=disableSTDIO,
276            shellExpandArguments=shellExpandArguments,
277            trace=trace,
278            initCommands=initCommands,
279            preRunCommands=preRunCommands,
280            stopCommands=stopCommands,
281            exitCommands=exitCommands,
282            sourcePath=sourcePath,
283            debuggerRoot=debuggerRoot,
284            launchCommands=launchCommands)
285        if not (response and response['success']):
286            self.assertTrue(response['success'],
287                            'launch failed (%s)' % (response['message']))
288
289    def build_and_launch(self, program, args=None, cwd=None, env=None,
290                         stopOnEntry=False, disableASLR=True,
291                         disableSTDIO=False, shellExpandArguments=False,
292                         trace=False, initCommands=None, preRunCommands=None,
293                         stopCommands=None, exitCommands=None,
294                         sourcePath=None, debuggerRoot=None):
295        '''Build the default Makefile target, create the VSCode debug adaptor,
296           and launch the process.
297        '''
298        self.build_and_create_debug_adaptor()
299        self.assertTrue(os.path.exists(program), 'executable must exist')
300
301        self.launch(program, args, cwd, env, stopOnEntry, disableASLR,
302                    disableSTDIO, shellExpandArguments, trace,
303                    initCommands, preRunCommands, stopCommands, exitCommands,
304                    sourcePath, debuggerRoot)
305