Index: lldb/packages/Python/lldbsuite/test/eventlistener.py =================================================================== --- /dev/null +++ lldb/packages/Python/lldbsuite/test/eventlistener.py @@ -0,0 +1,57 @@ +import threading + +import lldb +from lldbsuite.test.lldbtest import * + +class EventListener(TestBase): + + """ + Base class for lldb event listener tests. + + This class will setup and start an event listener for the test to use. + + If the event received matches the source broadcaster, the event is + queued up in a list that the user can access later on. + """ + NO_DEBUG_INFO_TESTCASE = True + + eBroadcastBitStopListenerThread = (1 << 0) + fetched_events = [] + + + def setUp(self, src_broadcaster, event_mask, event_data_extractor): + self.src_broadcaster = src_broadcaster + self.broadcaster = lldb.SBBroadcaster('lldb.test.broadcaster') + self.listener = lldb.SBListener("lldb.test.listener") + self.listener.StartListeningForEvents(self.broadcaster, + self.eBroadcastBitStopListenerThread) + + src_broadcaster.AddListener(self.listener, event_mask) + + self.listener_thread = threading.Thread(target=self._fetch_events, + args=(event_data_extractor,)) + self.listener_thread.start() + + def tearDown(self): + # Broadcast a eBroadcastBitStopListenerThread` event so the background + # thread stops listening to events, then join the background thread. + self.broadcaster.BroadcastEventByType(self.eBroadcastBitStopListenerThread) + self.listener_thread.join() + TestBase.tearDown(self) + + def _fetch_events(self, event_data_extractor): + event = lldb.SBEvent() + + done = False + while not done: + if self.listener.GetNextEvent(event): + event_mask = event.GetType(); + if event.BroadcasterMatchesRef(self.broadcaster): + if event_mask & self.eBroadcastBitStopListenerThread: + done = True; + elif event.BroadcasterMatchesRef(self.src_broadcaster): + ret_args = event_data_extractor(event); + if not ret_args: + continue + + self.fetched_events.append(ret_args) Index: lldb/test/API/functionalities/diagnostic_reporting/TestDiagnosticReporting.py =================================================================== --- lldb/test/API/functionalities/diagnostic_reporting/TestDiagnosticReporting.py +++ lldb/test/API/functionalities/diagnostic_reporting/TestDiagnosticReporting.py @@ -2,51 +2,24 @@ Test that we are able to broadcast and receive diagnostic events from lldb """ import lldb -from lldbsuite.test.lldbtest import * -from lldbsuite.test.decorators import * + import lldbsuite.test.lldbutil as lldbutil -import threading +from lldbsuite.test.lldbtest import * +from lldbsuite.test.eventlistener import EventListener -class TestDiagnosticReporting(TestBase): +class TestDiagnosticReporting(EventListener): mydir = TestBase.compute_mydir(__file__) - eBroadcastBitStopDiagnosticThread = (1 << 0) - def setUp(self): - TestBase.setUp(self) - self.diagnostic_events = [] - - def fetch_events(self): - event = lldb.SBEvent() - - done = False - while not done: - if self.listener.WaitForEvent(1, event): - event_mask = event.GetType() - if event.BroadcasterMatchesRef(self.test_broadcaster): - if event_mask & self.eBroadcastBitStopDiagnosticThread: - done = True - elif event.BroadcasterMatchesRef(self.diagnostic_broadcaster): - self.diagnostic_events.append( - lldb.SBDebugger.GetDiagnosticFromEvent(event)) + TestBase.setUp(self) # we need this to be able to access `self.dbg` + event_mask = lldb.SBDebugger.eBroadcastBitWarning | lldb.SBDebugger.eBroadcastBitError + EventListener.setUp(self, self.dbg.GetBroadcaster(), event_mask, + lldb.SBDebugger.GetDiagnosticFromEvent) def test_dwarf_symbol_loading_diagnostic_report(self): """Test that we are able to fetch diagnostic events""" - self.listener = lldb.SBListener("lldb.diagnostic.listener") - self.test_broadcaster = lldb.SBBroadcaster('lldb.broadcaster.test') - self.listener.StartListeningForEvents( - self.test_broadcaster, self.eBroadcastBitStopDiagnosticThread) - - self.diagnostic_broadcaster = self.dbg.GetBroadcaster() - self.diagnostic_broadcaster.AddListener( - self.listener, lldb.SBDebugger.eBroadcastBitWarning) - self.diagnostic_broadcaster.AddListener( - self.listener, lldb.SBDebugger.eBroadcastBitError) - - listener_thread = threading.Thread(target=self.fetch_events) - listener_thread.start() self.yaml2obj("minidump.yaml", self.getBuildArtifact("minidump.core")) @@ -55,13 +28,9 @@ self.process = self.target.LoadCore( self.getBuildArtifact("minidump.core")) - self.test_broadcaster.BroadcastEventByType( - self.eBroadcastBitStopDiagnosticThread) - listener_thread.join() - - self.assertEquals(len(self.diagnostic_events), 1) + self.assertEquals(len(self.fetched_events), 1) - diagnostic_event = self.diagnostic_events[0] + diagnostic_event = self.fetched_events[0] self.assertEquals( diagnostic_event.GetValueForKey("type").GetStringValue(100), "warning") Index: lldb/test/API/functionalities/progress_reporting/TestProgressReporting.py =================================================================== --- lldb/test/API/functionalities/progress_reporting/TestProgressReporting.py +++ lldb/test/API/functionalities/progress_reporting/TestProgressReporting.py @@ -2,57 +2,26 @@ Test that we are able to broadcast and receive progress events from lldb """ import lldb -from lldbsuite.test.lldbtest import * -from lldbsuite.test.decorators import * + import lldbsuite.test.lldbutil as lldbutil -import threading -class TestProgressReporting(TestBase): +from lldbsuite.test.lldbtest import * +from lldbsuite.test.eventlistener import EventListener + - mydir = TestBase.compute_mydir(__file__) +class TestProgressReporting(EventListener): - eBroadcastBitStopProgressThread = (1 << 0) + mydir = TestBase.compute_mydir(__file__) def setUp(self): - TestBase.setUp(self) - self.progress_events = [] - - def fetch_events(self): - event = lldb.SBEvent() - - done = False - while not done: - if self.listener.WaitForEvent(1, event): - event_mask = event.GetType(); - if event.BroadcasterMatchesRef(self.test_broadcaster): - if event_mask & self.eBroadcastBitStopProgressThread: - done = True; - elif event.BroadcasterMatchesRef(self.progress_broadcaster): - ret_args = lldb.SBDebugger().GetProgressFromEvent(event); - self.assertGreater(len(ret_args), 1) - - message = ret_args[0] - if message: - self.progress_events.append((message, event)) + TestBase.setUp(self) # we need this to be able to access `self.dbg` + EventListener.setUp(self, self.dbg.GetBroadcaster(), + lldb.SBDebugger.eBroadcastBitProgress, + lldb.SBDebugger.GetProgressFromEvent) def test_dwarf_symbol_loading_progress_report(self): """Test that we are able to fetch dwarf symbol loading progress events""" self.build() - self.listener = lldb.SBListener("lldb.progress.listener") - self.test_broadcaster = lldb.SBBroadcaster('lldb.broadcaster.test') - self.listener.StartListeningForEvents(self.test_broadcaster, - self.eBroadcastBitStopProgressThread) - - self.progress_broadcaster = self.dbg.GetBroadcaster() - self.progress_broadcaster.AddListener(self.listener, lldb.SBDebugger.eBroadcastBitProgress) - - listener_thread = threading.Thread(target=self.fetch_events) - listener_thread.start() - lldbutil.run_to_source_breakpoint(self, 'break here', lldb.SBFileSpec('main.c')) - - self.test_broadcaster.BroadcastEventByType(self.eBroadcastBitStopProgressThread) - listener_thread.join() - - self.assertGreater(len(self.progress_events), 0) + self.assertGreater(len(self.fetched_events), 0)