1
2from lldbsuite.test.lldbtest import *
3import os
4import vscode
5import time
6
7
8class VSCodeTestCaseBase(TestBase):
9
10    NO_DEBUG_INFO_TESTCASE = True
11
12    def create_debug_adaptor(self, lldbVSCodeEnv=None):
13        '''Create the Visual Studio Code debug adaptor'''
14        self.assertTrue(os.path.exists(self.lldbVSCodeExec),
15                        'lldb-vscode must exist')
16        log_file_path = self.getBuildArtifact('vscode.txt')
17        self.vscode = vscode.DebugAdaptor(
18            executable=self.lldbVSCodeExec, init_commands=self.setUpCommands(),
19            log_file=log_file_path, env=lldbVSCodeEnv)
20
21    def build_and_create_debug_adaptor(self, lldbVSCodeEnv=None):
22        self.build()
23        self.create_debug_adaptor(lldbVSCodeEnv)
24
25    def set_source_breakpoints(self, source_path, lines, condition=None,
26                               hitCondition=None):
27        '''Sets source breakpoints and returns an array of strings containing
28           the breakpoint IDs ("1", "2") for each breakpoint that was set.
29        '''
30        response = self.vscode.request_setBreakpoints(
31            source_path, lines, condition=condition, hitCondition=hitCondition)
32        if response is None:
33            return []
34        breakpoints = response['body']['breakpoints']
35        breakpoint_ids = []
36        for breakpoint in breakpoints:
37            breakpoint_ids.append('%i' % (breakpoint['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 IDs
44           ("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            breakpoint_ids.append('%i' % (breakpoint['id']))
54        return breakpoint_ids
55
56    def waitUntil(self, condition_callback):
57        for _ in range(20):
58            if condition_callback():
59                return True
60            time.sleep(0.5)
61        return False
62
63    def verify_breakpoint_hit(self, breakpoint_ids):
64        '''Wait for the process we are debugging to stop, and verify we hit
65           any breakpoint location in the "breakpoint_ids" array.
66           "breakpoint_ids" should be a list of breakpoint ID strings
67           (["1", "2"]). The return value from self.set_source_breakpoints()
68           or self.set_function_breakpoints() can be passed to this function'''
69        stopped_events = self.vscode.wait_for_stopped()
70        for stopped_event in stopped_events:
71            if 'body' in stopped_event:
72                body = stopped_event['body']
73                if 'reason' not in body:
74                    continue
75                if body['reason'] != 'breakpoint':
76                    continue
77                if 'description' not in body:
78                    continue
79                # Descriptions for breakpoints will be in the form
80                # "breakpoint 1.1", so look for any description that matches
81                # ("breakpoint 1.") in the description field as verification
82                # that one of the breakpoint locations was hit. VSCode doesn't
83                # allow breakpoints to have multiple locations, but LLDB does.
84                # So when looking at the description we just want to make sure
85                # the right breakpoint matches and not worry about the actual
86                # location.
87                description = body['description']
88                for breakpoint_id in breakpoint_ids:
89                    match_desc = 'breakpoint %s.' % (breakpoint_id)
90                    if match_desc in description:
91                        return
92        self.assertTrue(False, "breakpoint not hit")
93
94    def verify_exception_breakpoint_hit(self, filter_label):
95        '''Wait for the process we are debugging to stop, and verify the stop
96           reason is 'exception' and that the description matches
97           'filter_label'
98        '''
99        stopped_events = self.vscode.wait_for_stopped()
100        for stopped_event in stopped_events:
101            if 'body' in stopped_event:
102                body = stopped_event['body']
103                if 'reason' not in body:
104                    continue
105                if body['reason'] != 'exception':
106                    continue
107                if 'description' not in body:
108                    continue
109                description = body['description']
110                if filter_label == description:
111                    return True
112        return False
113
114    def verify_commands(self, flavor, output, commands):
115        self.assertTrue(output and len(output) > 0, "expect console output")
116        lines = output.splitlines()
117        prefix = '(lldb) '
118        for cmd in commands:
119            found = False
120            for line in lines:
121                if line.startswith(prefix) and cmd in line:
122                    found = True
123                    break
124            self.assertTrue(found,
125                            "verify '%s' found in console output for '%s'" % (
126                                cmd, flavor))
127
128    def get_dict_value(self, d, key_path):
129        '''Verify each key in the key_path array is in contained in each
130           dictionary within "d". Assert if any key isn't in the
131           corresponding dictionary. This is handy for grabbing values from VS
132           Code response dictionary like getting
133           response['body']['stackFrames']
134        '''
135        value = d
136        for key in key_path:
137            if key in value:
138                value = value[key]
139            else:
140                self.assertTrue(key in value,
141                                'key "%s" from key_path "%s" not in "%s"' % (
142                                    key, key_path, d))
143        return value
144
145    def get_stackFrames_and_totalFramesCount(self, threadId=None, startFrame=None,
146                        levels=None, dump=False):
147        response = self.vscode.request_stackTrace(threadId=threadId,
148                                                  startFrame=startFrame,
149                                                  levels=levels,
150                                                  dump=dump)
151        if response:
152            stackFrames = self.get_dict_value(response, ['body', 'stackFrames'])
153            totalFrames = self.get_dict_value(response, ['body', 'totalFrames'])
154            self.assertTrue(totalFrames > 0,
155                    'verify totalFrames count is provided by extension that supports '
156                    'async frames loading')
157            return (stackFrames, totalFrames)
158        return (None, 0)
159
160    def get_stackFrames(self, threadId=None, startFrame=None, levels=None,
161                        dump=False):
162        (stackFrames, totalFrames) = self.get_stackFrames_and_totalFramesCount(
163                                                threadId=threadId,
164                                                startFrame=startFrame,
165                                                levels=levels,
166                                                dump=dump)
167        return stackFrames
168
169    def get_source_and_line(self, threadId=None, frameIndex=0):
170        stackFrames = self.get_stackFrames(threadId=threadId,
171                                           startFrame=frameIndex,
172                                           levels=1)
173        if stackFrames is not None:
174            stackFrame = stackFrames[0]
175            ['source', 'path']
176            if 'source' in stackFrame:
177                source = stackFrame['source']
178                if 'path' in source:
179                    if 'line' in stackFrame:
180                        return (source['path'], stackFrame['line'])
181        return ('', 0)
182
183    def get_stdout(self, timeout=0.0):
184        return self.vscode.get_output('stdout', timeout=timeout)
185
186    def get_console(self, timeout=0.0):
187        return self.vscode.get_output('console', timeout=timeout)
188
189    def collect_console(self, duration):
190        return self.vscode.collect_output('console', duration=duration)
191
192    def get_local_as_int(self, name, threadId=None):
193        value = self.vscode.get_local_variable_value(name, threadId=threadId)
194        if value.startswith('0x'):
195            return int(value, 16)
196        elif value.startswith('0'):
197            return int(value, 8)
198        else:
199            return int(value)
200
201    def set_local(self, name, value, id=None):
202        '''Set a top level local variable only.'''
203        return self.vscode.request_setVariable(1, name, str(value), id=id)
204
205    def set_global(self, name, value, id=None):
206        '''Set a top level global variable only.'''
207        return self.vscode.request_setVariable(2, name, str(value), id=id)
208
209    def stepIn(self, threadId=None, waitForStop=True):
210        self.vscode.request_stepIn(threadId=threadId)
211        if waitForStop:
212            return self.vscode.wait_for_stopped()
213        return None
214
215    def stepOver(self, threadId=None, waitForStop=True):
216        self.vscode.request_next(threadId=threadId)
217        if waitForStop:
218            return self.vscode.wait_for_stopped()
219        return None
220
221    def stepOut(self, threadId=None, waitForStop=True):
222        self.vscode.request_stepOut(threadId=threadId)
223        if waitForStop:
224            return self.vscode.wait_for_stopped()
225        return None
226
227    def continue_to_next_stop(self):
228        self.vscode.request_continue()
229        return self.vscode.wait_for_stopped()
230
231    def continue_to_breakpoints(self, breakpoint_ids):
232        self.vscode.request_continue()
233        self.verify_breakpoint_hit(breakpoint_ids)
234
235    def continue_to_exception_breakpoint(self, filter_label):
236        self.vscode.request_continue()
237        self.assertTrue(self.verify_exception_breakpoint_hit(filter_label),
238                        'verify we got "%s"' % (filter_label))
239
240    def continue_to_exit(self, exitCode=0):
241        self.vscode.request_continue()
242        stopped_events = self.vscode.wait_for_stopped()
243        self.assertEquals(len(stopped_events), 1,
244                        "stopped_events = {}".format(stopped_events))
245        self.assertEquals(stopped_events[0]['event'], 'exited',
246                        'make sure program ran to completion')
247        self.assertEquals(stopped_events[0]['body']['exitCode'], exitCode,
248                        'exitCode == %i' % (exitCode))
249
250    def attach(self, program=None, pid=None, waitFor=None, trace=None,
251               initCommands=None, preRunCommands=None, stopCommands=None,
252               exitCommands=None, attachCommands=None, coreFile=None,
253               disconnectAutomatically=True, terminateCommands=None,
254               postRunCommands=None, sourceMap=None):
255        '''Build the default Makefile target, create the VSCode debug adaptor,
256           and attach to the process.
257        '''
258        # Make sure we disconnect and terminate the VSCode debug adaptor even
259        # if we throw an exception during the test case.
260        def cleanup():
261            if disconnectAutomatically:
262                self.vscode.request_disconnect(terminateDebuggee=True)
263            self.vscode.terminate()
264
265        # Execute the cleanup function during test case tear down.
266        self.addTearDownHook(cleanup)
267        # Initialize and launch the program
268        self.vscode.request_initialize()
269        response = self.vscode.request_attach(
270            program=program, pid=pid, waitFor=waitFor, trace=trace,
271            initCommands=initCommands, preRunCommands=preRunCommands,
272            stopCommands=stopCommands, exitCommands=exitCommands,
273            attachCommands=attachCommands, terminateCommands=terminateCommands,
274            coreFile=coreFile, postRunCommands=postRunCommands,
275            sourceMap=sourceMap)
276        if not (response and response['success']):
277            self.assertTrue(response['success'],
278                            'attach failed (%s)' % (response['message']))
279
280    def launch(self, program=None, args=None, cwd=None, env=None,
281               stopOnEntry=False, disableASLR=True,
282               disableSTDIO=False, shellExpandArguments=False,
283               trace=False, initCommands=None, preRunCommands=None,
284               stopCommands=None, exitCommands=None, terminateCommands=None,
285               sourcePath=None, debuggerRoot=None, launchCommands=None,
286               sourceMap=None, disconnectAutomatically=True, runInTerminal=False,
287               expectFailure=False, postRunCommands=None):
288        '''Sending launch request to vscode
289        '''
290
291        # Make sure we disconnect and terminate the VSCode debug adapter,
292        # if we throw an exception during the test case
293        def cleanup():
294            if disconnectAutomatically:
295                self.vscode.request_disconnect(terminateDebuggee=True)
296            self.vscode.terminate()
297
298        # Execute the cleanup function during test case tear down.
299        self.addTearDownHook(cleanup)
300
301        # Initialize and launch the program
302        self.vscode.request_initialize()
303        response = self.vscode.request_launch(
304            program,
305            args=args,
306            cwd=cwd,
307            env=env,
308            stopOnEntry=stopOnEntry,
309            disableASLR=disableASLR,
310            disableSTDIO=disableSTDIO,
311            shellExpandArguments=shellExpandArguments,
312            trace=trace,
313            initCommands=initCommands,
314            preRunCommands=preRunCommands,
315            stopCommands=stopCommands,
316            exitCommands=exitCommands,
317            terminateCommands=terminateCommands,
318            sourcePath=sourcePath,
319            debuggerRoot=debuggerRoot,
320            launchCommands=launchCommands,
321            sourceMap=sourceMap,
322            runInTerminal=runInTerminal,
323            expectFailure=expectFailure,
324            postRunCommands=postRunCommands)
325
326        if expectFailure:
327            return response
328
329        if not (response and response['success']):
330            self.assertTrue(response['success'],
331                            'launch failed (%s)' % (response['message']))
332        # We need to trigger a request_configurationDone after we've successfully
333        # attached a runInTerminal process to finish initialization.
334        if runInTerminal:
335            self.vscode.request_configurationDone()
336        return response
337
338
339    def build_and_launch(self, program, args=None, cwd=None, env=None,
340                         stopOnEntry=False, disableASLR=True,
341                         disableSTDIO=False, shellExpandArguments=False,
342                         trace=False, initCommands=None, preRunCommands=None,
343                         stopCommands=None, exitCommands=None,
344                         terminateCommands=None, sourcePath=None,
345                         debuggerRoot=None, runInTerminal=False,
346                         disconnectAutomatically=True, postRunCommands=None,
347                         lldbVSCodeEnv=None):
348        '''Build the default Makefile target, create the VSCode debug adaptor,
349           and launch the process.
350        '''
351        self.build_and_create_debug_adaptor(lldbVSCodeEnv)
352        self.assertTrue(os.path.exists(program), 'executable must exist')
353
354        return self.launch(program, args, cwd, env, stopOnEntry, disableASLR,
355                    disableSTDIO, shellExpandArguments, trace,
356                    initCommands, preRunCommands, stopCommands, exitCommands,
357                    terminateCommands, sourcePath, debuggerRoot, runInTerminal=runInTerminal,
358                    disconnectAutomatically=disconnectAutomatically,
359                    postRunCommands=postRunCommands)
360