1"""
2Test the functionality of interactive scripted processes
3"""
4
5import lldb
6import lldbsuite.test.lldbutil as lldbutil
7from lldbsuite.test.lldbtest import *
8import json, os
9
10
11class TestInteractiveScriptedProcess(TestBase):
12
13    NO_DEBUG_INFO_TESTCASE = True
14
15    def setUp(self):
16        # Call super's setUp().
17        TestBase.setUp(self)
18        # Build and load test program
19        self.build()
20        self.runCmd("file " + self.getBuildArtifact("a.out"), CURRENT_EXECUTABLE_SET)
21        self.main_source_file = lldb.SBFileSpec("main.cpp")
22        self.script_module = "interactive_scripted_process"
23        self.script_file = self.script_module + ".py"
24
25    def test_passthrough_launch(self):
26        """Test a simple pass-through process launch"""
27        self.passthrough_launch()
28
29        lldbutil.run_break_set_by_source_regexp(self, "also break here")
30        self.assertEqual(self.mux_target.GetNumBreakpoints(), 2)
31        error = self.mux_process.Continue()
32        self.assertSuccess(error, "Resuming multiplexer scripted process")
33        self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
34
35        event = lldbutil.fetch_next_event(
36            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
37        )
38        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
39        event = lldbutil.fetch_next_event(
40            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
41        )
42        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
43
44    def test_multiplexed_launch(self):
45        """Test a multiple interactive scripted process debugging"""
46        self.passthrough_launch()
47        self.assertEqual(self.dbg.GetNumTargets(), 2)
48
49        driving_target = self.mux_process.GetScriptedImplementation().driving_target
50        self.assertTrue(driving_target.IsValid(), "Driving target is invalid")
51
52        # Create a target for the multiplexed even scripted process
53        even_target = self.duplicate_target(driving_target)
54        self.assertTrue(
55            even_target.IsValid(),
56            "Couldn't duplicate driving target to launch multiplexed even scripted process",
57        )
58
59        class_name = f"{self.script_module}.MultiplexedScriptedProcess"
60        dictionary = {"driving_target_idx": self.dbg.GetIndexOfTarget(self.mux_target)}
61
62        dictionary["parity"] = 0
63        muxed_launch_info = self.get_launch_info(class_name, dictionary)
64
65        # Launch Even Child Scripted Process
66        error = lldb.SBError()
67        even_process = even_target.Launch(muxed_launch_info, error)
68        self.assertTrue(
69            even_process, "Couldn't launch multiplexed even scripted process"
70        )
71        self.multiplex(even_process)
72
73        # Check that the even process started running
74        event = lldbutil.fetch_next_event(
75            self, self.dbg.GetListener(), even_process.GetBroadcaster()
76        )
77        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
78        # Check that the even process stopped
79        event = lldbutil.fetch_next_event(
80            self, self.dbg.GetListener(), even_process.GetBroadcaster()
81        )
82        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
83
84        self.assertTrue(even_process.IsValid(), "Got a valid process")
85        self.assertState(
86            even_process.GetState(), lldb.eStateStopped, "Process is stopped"
87        )
88
89        # Create a target for the multiplexed odd scripted process
90        odd_target = self.duplicate_target(driving_target)
91        self.assertTrue(
92            odd_target.IsValid(),
93            "Couldn't duplicate driving target to launch multiplexed odd scripted process",
94        )
95
96        dictionary["parity"] = 1
97        muxed_launch_info = self.get_launch_info(class_name, dictionary)
98
99        # Launch Odd Child Scripted Process
100        error = lldb.SBError()
101        odd_process = odd_target.Launch(muxed_launch_info, error)
102        self.assertTrue(odd_process, "Couldn't launch multiplexed odd scripted process")
103        self.multiplex(odd_process)
104
105        # Check that the odd process started running
106        event = lldbutil.fetch_next_event(
107            self, self.dbg.GetListener(), odd_process.GetBroadcaster()
108        )
109        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
110        # Check that the odd process stopped
111        event = lldbutil.fetch_next_event(
112            self, self.dbg.GetListener(), odd_process.GetBroadcaster()
113        )
114        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
115
116        self.assertTrue(odd_process.IsValid(), "Got a valid process")
117        self.assertState(
118            odd_process.GetState(), lldb.eStateStopped, "Process is stopped"
119        )
120
121        # Set a breakpoint on the odd child process
122        bkpt = odd_target.BreakpointCreateBySourceRegex(
123            "also break here", self.main_source_file
124        )
125        self.assertEqual(odd_target.GetNumBreakpoints(), 1)
126        self.assertTrue(bkpt, "Second breakpoint set on child scripted process")
127        self.assertEqual(bkpt.GetNumLocations(), 1, "Second breakpoint has 1 location")
128
129        # Verify that the breakpoint was also set on the multiplexer & real target
130        self.assertEqual(self.mux_target.GetNumBreakpoints(), 2)
131        bkpt = self.mux_target.GetBreakpointAtIndex(1)
132        self.assertEqual(
133            bkpt.GetNumLocations(), 1, "Second breakpoint set on mux scripted process"
134        )
135        self.assertTrue(bkpt.MatchesName("multiplexed_scripted_process_421"))
136
137        self.assertGreater(driving_target.GetNumBreakpoints(), 1)
138
139        # Resume execution on child process
140        error = odd_process.Continue()
141        self.assertSuccess(error, "Resuming odd child scripted process")
142        self.assertTrue(odd_process.IsValid(), "Got a valid process")
143
144        # Since all the execution is asynchronous, the order in which events
145        # arrive is non-deterministic, so we need a data structure to make sure
146        # we received both the running and stopped event for each target.
147
148        # Initialize the execution event "bingo book", that maps a process index
149        # to a dictionary that contains flags that are not set for the process
150        # events that we care about (running & stopped)
151
152        execution_events = {
153            1: {lldb.eStateRunning: False, lldb.eStateStopped: False},
154            2: {lldb.eStateRunning: False, lldb.eStateStopped: False},
155            3: {lldb.eStateRunning: False, lldb.eStateStopped: False},
156        }
157
158        def fetch_process_event(self, execution_events):
159            event = lldbutil.fetch_next_event(
160                self,
161                self.dbg.GetListener(),
162                lldb.SBProcess.GetBroadcasterClass(),
163                match_class=True,
164            )
165            state = lldb.SBProcess.GetStateFromEvent(event)
166            self.assertIn(state, [lldb.eStateRunning, lldb.eStateStopped])
167            event_process = lldb.SBProcess.GetProcessFromEvent(event)
168            self.assertTrue(event_process.IsValid())
169            event_target = event_process.GetTarget()
170            event_target_idx = self.dbg.GetIndexOfTarget(event_target)
171            self.assertFalse(
172                execution_events[event_target_idx][state],
173                "Event already received for this process",
174            )
175            execution_events[event_target_idx][state] = True
176
177        event = lldbutil.fetch_next_event(
178            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
179        )
180        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
181
182        event = lldbutil.fetch_next_event(
183            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
184        )
185        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
186
187        for _ in range((self.dbg.GetNumTargets() - 1) * 2):
188            fetch_process_event(self, execution_events)
189
190        for target_index, event_states in execution_events.items():
191            for state, is_set in event_states.items():
192                self.assertTrue(is_set, f"Target {target_index} has state {state} set")
193
194    def duplicate_target(self, driving_target):
195        exe = driving_target.executable.fullpath
196        triple = driving_target.triple
197        return self.dbg.CreateTargetWithFileAndTargetTriple(exe, triple)
198
199    def get_launch_info(self, class_name, script_dict):
200        structured_data = lldb.SBStructuredData()
201        structured_data.SetFromJSON(json.dumps(script_dict))
202
203        launch_info = lldb.SBLaunchInfo(None)
204        launch_info.SetProcessPluginName("ScriptedProcess")
205        launch_info.SetScriptedProcessClassName(class_name)
206        launch_info.SetScriptedProcessDictionary(structured_data)
207        return launch_info
208
209    def multiplex(self, muxed_process):
210        muxed_process.GetScriptedImplementation().multiplexer = (
211            self.mux_process.GetScriptedImplementation()
212        )
213        self.mux_process.GetScriptedImplementation().multiplexed_processes[
214            muxed_process.GetProcessID()
215        ] = muxed_process
216
217    def passthrough_launch(self):
218        """Test that a simple passthrough wrapper functions correctly"""
219        # First build the real target:
220        self.assertEqual(self.dbg.GetNumTargets(), 1)
221        real_target_id = 0
222        real_target = self.dbg.GetTargetAtIndex(real_target_id)
223        lldbutil.run_break_set_by_source_regexp(self, "Break here")
224        self.assertEqual(real_target.GetNumBreakpoints(), 1)
225
226        # Now source in the scripted module:
227        script_path = os.path.join(self.getSourceDir(), self.script_file)
228        self.runCmd(f"command script import '{script_path}'")
229
230        self.mux_target = self.duplicate_target(real_target)
231        self.assertTrue(self.mux_target.IsValid(), "duplicate target succeeded")
232
233        mux_class = f"{self.script_module}.MultiplexerScriptedProcess"
234        script_dict = {"driving_target_idx": real_target_id}
235        mux_launch_info = self.get_launch_info(mux_class, script_dict)
236        self.mux_process_listener = lldb.SBListener(
237            "lldb.test.interactive-scripted-process.listener"
238        )
239        mux_launch_info.SetShadowListener(self.mux_process_listener)
240
241        self.dbg.SetAsync(True)
242        error = lldb.SBError()
243        self.mux_process = self.mux_target.Launch(mux_launch_info, error)
244        self.assertSuccess(error, "Launched multiplexer scripted process")
245        self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
246
247        # Check that the mux process started running
248        event = lldbutil.fetch_next_event(
249            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
250        )
251        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
252        # Check that the real process started running
253        event = lldbutil.fetch_next_event(
254            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
255        )
256        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
257
258        # Check that the real process stopped
259        event = lldbutil.fetch_next_event(
260            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
261        )
262        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
263        # Check that the mux process stopped
264        event = lldbutil.fetch_next_event(
265            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
266        )
267        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
268
269        real_process = real_target.GetProcess()
270        self.assertTrue(real_process.IsValid(), "Got a valid process")
271        self.assertState(
272            real_process.GetState(), lldb.eStateStopped, "Process is stopped"
273        )
274
275        # This is a passthrough, so the two processes should have the same state:
276        # Check that we got the right threads:
277        self.assertEqual(
278            len(real_process.threads),
279            len(self.mux_process.threads),
280            "Same number of threads",
281        )
282        for id in range(len(real_process.threads)):
283            real_pc = real_process.threads[id].frame[0].pc
284            mux_pc = self.mux_process.threads[id].frame[0].pc
285            self.assertEqual(real_pc, mux_pc, f"PC's equal for {id}")
286