1"""
2Test lldb-vscode setBreakpoints request
3"""
4
5
6import unittest2
7import vscode
8from lldbsuite.test.decorators import *
9from lldbsuite.test.lldbtest import *
10from lldbsuite.test import lldbutil
11import lldbvscode_testcase
12import os
13import shutil
14import subprocess
15import tempfile
16import threading
17import time
18
19
20def spawn_and_wait(program, delay):
21    if delay:
22        time.sleep(delay)
23    process = subprocess.Popen([program],
24                               stdin=subprocess.PIPE,
25                               stdout=subprocess.PIPE,
26                               stderr=subprocess.PIPE)
27    process.wait()
28
29
30class TestVSCode_attach(lldbvscode_testcase.VSCodeTestCaseBase):
31
32    mydir = TestBase.compute_mydir(__file__)
33
34    def set_and_hit_breakpoint(self, continueToExit=True):
35        source = 'main.c'
36        breakpoint1_line = line_number(source, '// breakpoint 1')
37        lines = [breakpoint1_line]
38        # Set breakoint in the thread function so we can step the threads
39        breakpoint_ids = self.set_source_breakpoints(source, lines)
40        self.assertEqual(len(breakpoint_ids), len(lines),
41                         "expect correct number of breakpoints")
42        self.continue_to_breakpoints(breakpoint_ids)
43        if continueToExit:
44            self.continue_to_exit()
45
46
47    @skipIfWindows
48    @skipIfNetBSD # Hangs on NetBSD as well
49    def test_by_pid(self):
50        '''
51            Tests attaching to a process by process ID.
52        '''
53        self.build_and_create_debug_adaptor()
54        program = self.getBuildArtifact("a.out")
55        self.process = subprocess.Popen([program],
56                                        stdin=subprocess.PIPE,
57                                        stdout=subprocess.PIPE,
58                                        stderr=subprocess.PIPE)
59        self.attach(pid=self.process.pid)
60        self.set_and_hit_breakpoint(continueToExit=True)
61
62    @skipIfWindows
63    @skipIfNetBSD # Hangs on NetBSD as well
64    def test_by_name(self):
65        '''
66            Tests attaching to a process by process name.
67        '''
68        self.build_and_create_debug_adaptor()
69        orig_program = self.getBuildArtifact("a.out")
70        # Since we are going to attach by process name, we need a unique
71        # process name that has minimal chance to match a process that is
72        # already running. To do this we use tempfile.mktemp() to give us a
73        # full path to a location where we can copy our executable. We then
74        # run this copy to ensure we don't get the error "more that one
75        # process matches 'a.out'".
76        program = tempfile.mktemp()
77        shutil.copyfile(orig_program, program)
78        shutil.copymode(orig_program, program)
79
80        # Use a file as a synchronization point between test and inferior.
81        pid_file_path = lldbutil.append_to_process_working_directory(self,
82            "pid_file_%d" % (int(time.time())))
83
84        def cleanup():
85            if os.path.exists(program):
86                os.unlink(program)
87            self.run_platform_command("rm %s" % (pid_file_path))
88        # Execute the cleanup function during test case tear down.
89        self.addTearDownHook(cleanup)
90
91        popen = self.spawnSubprocess(program, [pid_file_path])
92        self.addTearDownHook(self.cleanupSubprocesses)
93
94        pid = lldbutil.wait_for_file_on_target(self, pid_file_path)
95
96        self.attach(program=program)
97        self.set_and_hit_breakpoint(continueToExit=True)
98
99    @skipUnlessDarwin
100    @skipIfDarwin
101    @skipIfNetBSD # Hangs on NetBSD as well
102    def test_by_name_waitFor(self):
103        '''
104            Tests attaching to a process by process name and waiting for the
105            next instance of a process to be launched, ingoring all current
106            ones.
107        '''
108        self.build_and_create_debug_adaptor()
109        program = self.getBuildArtifact("a.out")
110        self.spawn_thread = threading.Thread(target=spawn_and_wait,
111                                             args=(program, 1.0,))
112        self.spawn_thread.start()
113        self.attach(program=program, waitFor=True)
114        self.set_and_hit_breakpoint(continueToExit=True)
115
116    @skipIfWindows
117    @skipIfDarwin
118    @skipIfNetBSD # Hangs on NetBSD as well
119    def test_commands(self):
120        '''
121            Tests the "initCommands", "preRunCommands", "stopCommands",
122            "exitCommands", and "attachCommands" that can be passed during
123            attach.
124
125            "initCommands" are a list of LLDB commands that get executed
126            before the targt is created.
127            "preRunCommands" are a list of LLDB commands that get executed
128            after the target has been created and before the launch.
129            "stopCommands" are a list of LLDB commands that get executed each
130            time the program stops.
131            "exitCommands" are a list of LLDB commands that get executed when
132            the process exits
133            "attachCommands" are a list of LLDB commands that get executed and
134            must have a valid process in the selected target in LLDB after
135            they are done executing. This allows custom commands to create any
136            kind of debug session.
137        '''
138        self.build_and_create_debug_adaptor()
139        program = self.getBuildArtifact("a.out")
140        # Here we just create a target and launch the process as a way to test
141        # if we are able to use attach commands to create any kind of a target
142        # and use it for debugging
143        attachCommands = [
144            'target create -d "%s"' % (program),
145            'process launch'
146        ]
147        initCommands = ['target list', 'platform list']
148        preRunCommands = ['image list a.out', 'image dump sections a.out']
149        stopCommands = ['frame variable', 'bt']
150        exitCommands = ['expr 2+3', 'expr 3+4']
151        self.attach(program=program,
152                    attachCommands=attachCommands,
153                    initCommands=initCommands,
154                    preRunCommands=preRunCommands,
155                    stopCommands=stopCommands,
156                    exitCommands=exitCommands)
157
158        # Get output from the console. This should contain both the
159        # "initCommands" and the "preRunCommands".
160        output = self.get_console()
161        # Verify all "initCommands" were found in console output
162        self.verify_commands('initCommands', output, initCommands)
163        # Verify all "preRunCommands" were found in console output
164        self.verify_commands('preRunCommands', output, preRunCommands)
165
166        functions = ['main']
167        breakpoint_ids = self.set_function_breakpoints(functions)
168        self.assertEquals(len(breakpoint_ids), len(functions),
169                        "expect one breakpoint")
170        self.continue_to_breakpoints(breakpoint_ids)
171        output = self.get_console(timeout=1.0)
172        self.verify_commands('stopCommands', output, stopCommands)
173
174        # Continue after launch and hit the "pause()" call and stop the target.
175        # Get output from the console. This should contain both the
176        # "stopCommands" that were run after we stop.
177        self.vscode.request_continue()
178        time.sleep(0.5)
179        self.vscode.request_pause()
180        self.vscode.wait_for_stopped()
181        output = self.get_console(timeout=1.0)
182        self.verify_commands('stopCommands', output, stopCommands)
183
184        # Continue until the program exits
185        self.continue_to_exit()
186        # Get output from the console. This should contain both the
187        # "exitCommands" that were run after the second breakpoint was hit
188        output = self.get_console(timeout=1.0)
189        self.verify_commands('exitCommands', output, exitCommands)
190