1""" 2Test that we are able to broadcast and receive progress events from lldb 3""" 4import lldb 5from lldbsuite.test.lldbtest import * 6from lldbsuite.test.decorators import * 7import lldbsuite.test.lldbutil as lldbutil 8import threading 9 10class TestProgressReporting(TestBase): 11 12 mydir = TestBase.compute_mydir(__file__) 13 14 eBroadcastBitStopProgressThread = (1 << 0) 15 16 def setUp(self): 17 TestBase.setUp(self) 18 self.progress_events = [] 19 20 def fetch_events(self): 21 event = lldb.SBEvent() 22 23 done = False 24 while not done: 25 if self.listener.WaitForEvent(1, event): 26 event_mask = event.GetType(); 27 if event.BroadcasterMatchesRef(self.test_broadcaster): 28 if event_mask & self.eBroadcastBitStopProgressThread: 29 done = True; 30 elif event.BroadcasterMatchesRef(self.progress_broadcaster): 31 ret_args = lldb.SBDebugger().GetProgressFromEvent(event); 32 self.assertGreater(len(ret_args), 1) 33 34 message = ret_args[0] 35 if message: 36 self.progress_events.append((message, event)) 37 38 def test_dwarf_symbol_loading_progress_report(self): 39 """Test that we are able to fetch dwarf symbol loading progress events""" 40 self.build() 41 42 self.listener = lldb.SBListener("lldb.progress.listener") 43 self.test_broadcaster = lldb.SBBroadcaster('lldb.broadcaster.test') 44 self.listener.StartListeningForEvents(self.test_broadcaster, 45 self.eBroadcastBitStopProgressThread) 46 47 self.progress_broadcaster = self.dbg.GetBroadcaster() 48 self.progress_broadcaster.AddListener(self.listener, lldb.SBDebugger.eBroadcastBitProgress) 49 50 listener_thread = threading.Thread(target=self.fetch_events) 51 listener_thread.start() 52 53 lldbutil.run_to_source_breakpoint(self, 'break here', lldb.SBFileSpec('main.c')) 54 55 self.test_broadcaster.BroadcastEventByType(self.eBroadcastBitStopProgressThread) 56 listener_thread.join() 57 58 self.assertGreater(len(self.progress_events), 0) 59