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, test_broadcaster): 21 listener = lldb.SBListener("lldb.progress.listener") 22 listener.StartListeningForEvents(test_broadcaster, 23 self.eBroadcastBitStopProgressThread) 24 25 progress_broadcaster = self.dbg.GetBroadcaster() 26 progress_broadcaster.AddListener(listener, lldb.SBDebugger.eBroadcastBitProgress) 27 28 event = lldb.SBEvent() 29 30 done = False 31 while not done: 32 if listener.WaitForEvent(1, event): 33 event_mask = event.GetType(); 34 if event.BroadcasterMatchesRef(test_broadcaster): 35 if event_mask & self.eBroadcastBitStopProgressThread: 36 done = True; 37 elif event.BroadcasterMatchesRef(progress_broadcaster): 38 message = lldb.SBDebugger().GetProgressFromEvent(event, 0, 0, 0, False); 39 if message: 40 self.progress_events.append((message, event)) 41 42 @skipUnlessDarwin 43 def test_dwarf_symbol_loading_progress_report(self): 44 """Test that we are able to fetch dwarf symbol loading progress events""" 45 self.build() 46 47 test_broadcaster = lldb.SBBroadcaster('lldb.broadcaster.test') 48 listener_thread = threading.Thread(target=self.fetch_events, 49 args=[test_broadcaster]) 50 listener_thread.start() 51 52 lldbutil.run_to_source_breakpoint(self, 'break here', lldb.SBFileSpec('main.c')) 53 54 test_broadcaster.BroadcastEventByType(self.eBroadcastBitStopProgressThread) 55 listener_thread.join() 56 57 self.assertTrue(len(self.progress_events) > 0) 58