maxjhandsome
/
llvm-project
forked from OSchip/llvm-project
1
0
Fork 0
Code Issues Pull Requests Packages Releases Wiki Activity

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

This patch introduces 2 new lldb utility functions:
- lldbutil.start_listening_from: This can be called in the test setup to
  create a listener and set it up for a specific event mask and add it
  to the user-provided broadcaster's list.
- lldbutil.fetch_next_event: This will use fetch a single event from the
  provided istener and return it if it matches the provided broadcaster.

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 these new helper functions.

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

Signed-off-by: Med Ismail Bennani <medismail.bennani@gmail.com>
This commit is contained in:
Med Ismail Bennani 2022-03-23 12:29:47 -07:00
parent 52f323d0f1
commit 9216baf87d
3 changed files with 46 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
lldb/packages/Python/lldbsuite/test/lldbutil.py
View File

@ -1255,6 +1255,29 @@ def expect_state_changes(test, listener, process, states, timeout=30):
lldb.SBProcess.GetStateFromEvent(event),
expected_state)
def start_listening_from(broadcaster, event_mask):
"""Creates a listener for a specific event mask and add it to the source broadcaster."""
listener = lldb.SBListener("lldb.test.listener")
broadcaster.AddListener(listener, event_mask)
return listener
def fetch_next_event(test, listener, broadcaster, timeout=10):
"""Fetch one event from the listener and return it if it matches the provided broadcaster.
Fails otherwise."""
event = lldb.SBEvent()
if listener.WaitForEvent(timeout, event):
if event.BroadcasterMatchesRef(broadcaster):
return event
test.fail("received event '%s' from unexpected broadcaster '%s'." %
(event.GetDescription(), event.GetBroadcaster().GetName()))
test.fail("couldn't fetch an event before reaching the timeout.")
# ===================================
# Utility functions related to Frames
# ===================================

53
lldb/test/API/functionalities/diagnostic_reporting/TestDiagnosticReporting.py
View File

@ -2,51 +2,25 @@
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
import lldbsuite.test.lldbutil as lldbutil
from lldbsuite.test.lldbtest import *
class TestDiagnosticReporting(TestBase):
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))
self.broadcaster = self.dbg.GetBroadcaster()
self.listener = lldbutil.start_listening_from(self.broadcaster,
lldb.SBDebugger.eBroadcastBitWarning |
lldb.SBDebugger.eBroadcastBitError)
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,17 +29,12 @@ 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.diagnostic_events), 1)
diagnostic_event = self.diagnostic_events[0]
event = lldbutil.fetch_next_event(self, self.listener, self.broadcaster)
diagnostic_data = lldb.SBDebugger.GetDiagnosticFromEvent(event)
self.assertEquals(
diagnostic_event.GetValueForKey("type").GetStringValue(100),
diagnostic_data.GetValueForKey("type").GetStringValue(100),
"warning")
self.assertEquals(
diagnostic_event.GetValueForKey("message").GetStringValue(100),
diagnostic_data.GetValueForKey("message").GetStringValue(100),
"unable to retrieve process ID from minidump file, setting process ID to 1"
)

50
lldb/test/API/functionalities/progress_reporting/TestProgressReporting.py
View File

@ -2,57 +2,31 @@
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
from lldbsuite.test.lldbtest import *
class TestProgressReporting(TestBase):
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))
self.broadcaster = self.dbg.GetBroadcaster()
self.listener = lldbutil.start_listening_from(self.broadcaster,
lldb.SBDebugger.eBroadcastBitProgress)
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()
event = lldbutil.fetch_next_event(self, self.listener, self.broadcaster)
ret_args = lldb.SBDebugger.GetProgressFromEvent(event)
self.assertGreater(len(ret_args), 0)
message = ret_args[0]
self.assertGreater(len(message), 0)
self.assertGreater(len(self.progress_events), 0)