squiid/llvm
1
0
Fork 0
Code Issues Pull requests Activity

[lldb/test] Add events listener helper class to lldbtest

Browse source 
This patch introduces a generic helper class that will listen for
event in a background thread and match it against a source broadcaster.

If the event received matches the source broadcaster, the event is
queued up in a list that the user can access later on.

The motivation behind this is to easily test new kinds of events
(i.e. Swift type-system progress events). However, this patch also
updates `TestProgressReporting.py` and `TestDiagnosticReporting.py`
to make use of this new helper class.

Differential Revision: https://reviews.llvm.org/D121977

Signed-off-by: Med Ismail Bennani <medismail.bennani@gmail.com>
This commit is contained in:
Med Ismail Bennani 2022-03-18 16:35:24 -07:00
parent 2d8b6a4784
commit 09ff41a087
3 changed files with 90 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
lldb/packages/Python/lldbsuite/test/eventlistener.py Normal file
View file

@ -0,0 +1,72 @@
import threading
import lldb
from lldbsuite.test.lldbtest import *
class EventListenerTestBase(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)
events = []
event_mask = None
event_data_extractor = None
def setUp(self):
TestBase.setUp(self)
self.src_broadcaster = self.dbg.GetBroadcaster()
self.broadcaster = lldb.SBBroadcaster('lldb.test.broadcaster')
self.listener = lldb.SBListener("lldb.test.listener")
self.listener.StartListeningForEvents(self.broadcaster,
self.eBroadcastBitStopListenerThread)
self.src_broadcaster.AddListener(self.listener, self.event_mask)
self.listener_thread = threading.Thread(target=self._fetch_events)
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 = 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):
# NOTE: https://wiki.python.org/moin/FromFunctionToMethod
#
# When assigning the `event_data_extractor` function pointer
# to the `EventListenerTestBase` instance, it turns the
# function object into an instance method which subsequently
# passes `self` as an extra argument.
# However, because most of the event data extractor
# functions are static, passing the `self` argument makes
# the number of passed arguments exceeds the function definition
# This is why we unwrap the function from the instance
# method object calling `__func__` instead.
ret_args = self.event_data_extractor.__func__(event);
if not ret_args:
continue
self.events.append(ret_args)

51
lldb/test/API/functionalities/diagnostic_reporting/TestDiagnosticReporting.py
View file

@ -2,51 +2,20 @@
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 EventListenerTestBase
class TestDiagnosticReporting(TestBase):
class TestDiagnosticReporting(EventListenerTestBase):
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))
event_mask = lldb.SBDebugger.eBroadcastBitWarning | lldb.SBDebugger.eBroadcastBitError
event_data_extractor = 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 +24,9 @@ class TestDiagnosticReporting(TestBase):
self.process = self.target.LoadCore(
self.getBuildArtifact("minidump.core"))
self.test_broadcaster.BroadcastEventByType(
self.eBroadcastBitStopDiagnosticThread)
listener_thread.join()
self.assertEquals(len(self.events), 1)
self.assertEquals(len(self.diagnostic_events), 1)
diagnostic_event = self.diagnostic_events[0]
diagnostic_event = self.events[0]
self.assertEquals(
diagnostic_event.GetValueForKey("type").GetStringValue(100),
"warning")

55
lldb/test/API/functionalities/progress_reporting/TestProgressReporting.py
View file

@ -2,57 +2,22 @@
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):
import lldbsuite.test.lldbutil as lldbutil
from lldbsuite.test.lldbtest import *
from lldbsuite.test.eventlistener import EventListenerTestBase
class TestProgressReporting(EventListenerTestBase):
mydir = TestBase.compute_mydir(__file__)
eBroadcastBitStopProgressThread = (1 << 0)
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))
event_mask = lldb.SBDebugger.eBroadcastBitProgress
event_data_extractor = 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.events), 0)