| """ |
| Test the functionality of interactive scripted processes |
| """ |
| |
| import lldb |
| import lldbsuite.test.lldbutil as lldbutil |
| from lldbsuite.test.decorators import * |
| from lldbsuite.test.lldbtest import * |
| import json, os |
| |
| |
| class TestInteractiveScriptedProcess(TestBase): |
| NO_DEBUG_INFO_TESTCASE = True |
| |
| def setUp(self): |
| # Call super's setUp(). |
| TestBase.setUp(self) |
| # Build and load test program |
| self.build() |
| self.runCmd("file " + self.getBuildArtifact("a.out"), CURRENT_EXECUTABLE_SET) |
| self.main_source_file = lldb.SBFileSpec("main.cpp") |
| self.script_module = "interactive_scripted_process" |
| self.script_file = self.script_module + ".py" |
| |
| @skipUnlessDarwin |
| @skipIfDarwin |
| def test_passthrough_launch(self): |
| """Test a simple pass-through process launch""" |
| self.passthrough_launch() |
| |
| lldbutil.run_break_set_by_source_regexp(self, "also break here") |
| self.assertEqual(self.mux_target.GetNumBreakpoints(), 2) |
| error = self.mux_process.Continue() |
| self.assertSuccess(error, "Resuming multiplexer scripted process") |
| self.assertTrue(self.mux_process.IsValid(), "Got a valid process") |
| |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| |
| @skipUnlessDarwin |
| @skipIfDarwin |
| def test_multiplexed_launch(self): |
| """Test a multiple interactive scripted process debugging""" |
| self.passthrough_launch() |
| self.assertEqual(self.dbg.GetNumTargets(), 2) |
| |
| driving_target = self.mux_process.GetScriptedImplementation().driving_target |
| self.assertTrue(driving_target.IsValid(), "Driving target is invalid") |
| |
| # Create a target for the multiplexed even scripted process |
| even_target = self.duplicate_target(driving_target) |
| self.assertTrue( |
| even_target.IsValid(), |
| "Couldn't duplicate driving target to launch multiplexed even scripted process", |
| ) |
| |
| class_name = f"{self.script_module}.MultiplexedScriptedProcess" |
| dictionary = {"driving_target_idx": self.dbg.GetIndexOfTarget(self.mux_target)} |
| |
| dictionary["parity"] = 0 |
| muxed_launch_info = self.get_launch_info(class_name, dictionary) |
| |
| # Launch Even Child Scripted Process |
| error = lldb.SBError() |
| even_process = even_target.Launch(muxed_launch_info, error) |
| self.assertTrue( |
| even_process, "Couldn't launch multiplexed even scripted process" |
| ) |
| self.multiplex(even_process) |
| |
| # Check that the even process started running |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), even_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| # Check that the even process stopped |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), even_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| |
| self.assertTrue(even_process.IsValid(), "Got a valid process") |
| self.assertState( |
| even_process.GetState(), lldb.eStateStopped, "Process is stopped" |
| ) |
| |
| # Create a target for the multiplexed odd scripted process |
| odd_target = self.duplicate_target(driving_target) |
| self.assertTrue( |
| odd_target.IsValid(), |
| "Couldn't duplicate driving target to launch multiplexed odd scripted process", |
| ) |
| |
| dictionary["parity"] = 1 |
| muxed_launch_info = self.get_launch_info(class_name, dictionary) |
| |
| # Launch Odd Child Scripted Process |
| error = lldb.SBError() |
| odd_process = odd_target.Launch(muxed_launch_info, error) |
| self.assertTrue(odd_process, "Couldn't launch multiplexed odd scripted process") |
| self.multiplex(odd_process) |
| |
| # Check that the odd process started running |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), odd_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| # Check that the odd process stopped |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), odd_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| |
| self.assertTrue(odd_process.IsValid(), "Got a valid process") |
| self.assertState( |
| odd_process.GetState(), lldb.eStateStopped, "Process is stopped" |
| ) |
| |
| # Set a breakpoint on the odd child process |
| bkpt = odd_target.BreakpointCreateBySourceRegex( |
| "also break here", self.main_source_file |
| ) |
| self.assertEqual(odd_target.GetNumBreakpoints(), 1) |
| self.assertTrue(bkpt, "Second breakpoint set on child scripted process") |
| self.assertEqual(bkpt.GetNumLocations(), 1, "Second breakpoint has 1 location") |
| |
| # Verify that the breakpoint was also set on the multiplexer & real target |
| self.assertEqual(self.mux_target.GetNumBreakpoints(), 2) |
| bkpt = self.mux_target.GetBreakpointAtIndex(1) |
| self.assertEqual( |
| bkpt.GetNumLocations(), 1, "Second breakpoint set on mux scripted process" |
| ) |
| self.assertTrue(bkpt.MatchesName("multiplexed_scripted_process_421")) |
| |
| self.assertGreater(driving_target.GetNumBreakpoints(), 1) |
| |
| # Resume execution on child process |
| error = odd_process.Continue() |
| self.assertSuccess(error, "Resuming odd child scripted process") |
| self.assertTrue(odd_process.IsValid(), "Got a valid process") |
| |
| # Since all the execution is asynchronous, the order in which events |
| # arrive is non-deterministic, so we need a data structure to make sure |
| # we received both the running and stopped event for each target. |
| |
| # Initialize the execution event "bingo book", that maps a process index |
| # to a dictionary that contains flags that are not set for the process |
| # events that we care about (running & stopped) |
| |
| execution_events = { |
| 1: {lldb.eStateRunning: False, lldb.eStateStopped: False}, |
| 2: {lldb.eStateRunning: False, lldb.eStateStopped: False}, |
| 3: {lldb.eStateRunning: False, lldb.eStateStopped: False}, |
| } |
| |
| def fetch_process_event(self, execution_events): |
| event = lldbutil.fetch_next_event( |
| self, |
| self.dbg.GetListener(), |
| lldb.SBProcess.GetBroadcasterClass(), |
| match_class=True, |
| ) |
| state = lldb.SBProcess.GetStateFromEvent(event) |
| self.assertIn(state, [lldb.eStateRunning, lldb.eStateStopped]) |
| event_process = lldb.SBProcess.GetProcessFromEvent(event) |
| self.assertTrue(event_process.IsValid()) |
| event_target = event_process.GetTarget() |
| event_target_idx = self.dbg.GetIndexOfTarget(event_target) |
| self.assertFalse( |
| execution_events[event_target_idx][state], |
| "Event already received for this process", |
| ) |
| execution_events[event_target_idx][state] = True |
| |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| |
| for _ in range((self.dbg.GetNumTargets() - 1) * 2): |
| fetch_process_event(self, execution_events) |
| |
| for target_index, event_states in execution_events.items(): |
| for state, is_set in event_states.items(): |
| self.assertTrue(is_set, f"Target {target_index} has state {state} set") |
| |
| def duplicate_target(self, driving_target): |
| exe = driving_target.executable.fullpath |
| triple = driving_target.triple |
| return self.dbg.CreateTargetWithFileAndTargetTriple(exe, triple) |
| |
| def get_launch_info(self, class_name, script_dict): |
| structured_data = lldb.SBStructuredData() |
| structured_data.SetFromJSON(json.dumps(script_dict)) |
| |
| launch_info = lldb.SBLaunchInfo(None) |
| launch_info.SetProcessPluginName("ScriptedProcess") |
| launch_info.SetScriptedProcessClassName(class_name) |
| launch_info.SetScriptedProcessDictionary(structured_data) |
| return launch_info |
| |
| def multiplex(self, muxed_process): |
| muxed_process.GetScriptedImplementation().multiplexer = ( |
| self.mux_process.GetScriptedImplementation() |
| ) |
| self.mux_process.GetScriptedImplementation().multiplexed_processes[ |
| muxed_process.GetProcessID() |
| ] = muxed_process |
| |
| def passthrough_launch(self): |
| """Test that a simple passthrough wrapper functions correctly""" |
| # First build the real target: |
| self.assertEqual(self.dbg.GetNumTargets(), 1) |
| real_target_id = 0 |
| real_target = self.dbg.GetTargetAtIndex(real_target_id) |
| lldbutil.run_break_set_by_source_regexp(self, "Break here") |
| self.assertEqual(real_target.GetNumBreakpoints(), 1) |
| |
| # Now source in the scripted module: |
| script_path = os.path.join(self.getSourceDir(), self.script_file) |
| self.runCmd(f"command script import '{script_path}'") |
| |
| self.mux_target = self.duplicate_target(real_target) |
| self.assertTrue(self.mux_target.IsValid(), "duplicate target succeeded") |
| |
| mux_class = f"{self.script_module}.MultiplexerScriptedProcess" |
| script_dict = {"driving_target_idx": real_target_id} |
| mux_launch_info = self.get_launch_info(mux_class, script_dict) |
| self.mux_process_listener = lldb.SBListener( |
| "lldb.test.interactive-scripted-process.listener" |
| ) |
| mux_launch_info.SetShadowListener(self.mux_process_listener) |
| |
| self.dbg.SetAsync(True) |
| error = lldb.SBError() |
| self.mux_process = self.mux_target.Launch(mux_launch_info, error) |
| self.assertSuccess(error, "Launched multiplexer scripted process") |
| self.assertTrue(self.mux_process.IsValid(), "Got a valid process") |
| |
| # Check that the mux process started running |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| # Check that the real process started running |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning) |
| |
| # Check that the real process stopped |
| event = lldbutil.fetch_next_event( |
| self, self.dbg.GetListener(), self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| # Check that the mux process stopped |
| event = lldbutil.fetch_next_event( |
| self, self.mux_process_listener, self.mux_process.GetBroadcaster() |
| ) |
| self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped) |
| |
| real_process = real_target.GetProcess() |
| self.assertTrue(real_process.IsValid(), "Got a valid process") |
| self.assertState( |
| real_process.GetState(), lldb.eStateStopped, "Process is stopped" |
| ) |
| |
| # This is a passthrough, so the two processes should have the same state: |
| # Check that we got the right threads: |
| self.assertEqual( |
| len(real_process.threads), |
| len(self.mux_process.threads), |
| "Same number of threads", |
| ) |
| for id in range(len(real_process.threads)): |
| real_pc = real_process.threads[id].frame[0].pc |
| mux_pc = self.mux_process.threads[id].frame[0].pc |
| self.assertEqual(real_pc, mux_pc, f"PC's equal for {id}") |
