"""
Test SBDebugger.InterruptRequested and SBCommandInterpreter.WasInterrupted.
"""
import lldb
import lldbsuite.test.lldbutil as lldbutil
from lldbsuite.test.lldbtest import *
import threading
import os
class TestDebuggerInterruption(TestBase):
"""This test runs a command that starts up, rendevous with the test thread
using threading barriers, then checks whether it has been interrupted.
The command's first argument is either 'interp' or 'debugger', to test
InterruptRequested and WasInterrupted respectively.
The command has two modes, interrupt and check, the former is the one that
waits for an interrupt. Then latter just returns whether an interrupt was
requested. We use the latter to make sure we took down the flag correctly."""
NO_DEBUG_INFO_TESTCASE = True
class CommandRunner(threading.Thread):
"""This class is for running a command, and for making a thread to run the command on.
It gets passed the test it is working on behalf of, and most of the important
objects come from the test."""
def __init__(self, test):
super().__init__()
self.test = test
def rendevous(self):
# We smuggle out barriers and event to the runner thread using thread local data:
import interruptible
interruptible.local_data = interruptible.BarrierContainer(
self.test.before_interrupt_barrier,
self.test.after_interrupt_barrier,
self.test.event,
)
class DirectCommandRunner(CommandRunner):
""" "This version runs a single command using HandleCommand."""
def __init__(self, test, command):
super().__init__(test)
self.command = command
def run(self):
self.rendevous()
result = self.test.dbg.GetCommandInterpreter().HandleCommand(
self.command, self.test.result
)
if self.test.result_barrier:
self.test.result_barrier.wait()
class CommandInterpreterRunner(CommandRunner):
"""This version runs the CommandInterpreter and feeds the command to it."""
def __init__(self, test):
super().__init__(test)
def run(self):
self.rendevous()
test = self.test
# We will use files for debugger input and output:
# First write down the command:
with open(test.getBuildArtifact(test.in_filename), "w") as f:
f.write(f"{test.command}\n")
# Now set the debugger's stdout & stdin to our files, and run
# the CommandInterpreter:
with open(test.out_filename, "w") as outf, open(
test.in_filename, "r"
) as inf:
outsbf = lldb.SBFile(outf.fileno(), "w", False)
orig_outf = test.dbg.GetOutputFile()
error = test.dbg.SetOutputFile(outsbf)
test.assertSuccess(error, "Could not set outfile")
insbf = lldb.SBFile(inf.fileno(), "r", False)
orig_inf = test.dbg.GetOutputFile()
error = test.dbg.SetInputFile(insbf)
test.assertSuccess(error, "Could not set infile")
options = lldb.SBCommandInterpreterRunOptions()
options.SetPrintResults(True)
options.SetEchoCommands(False)
test.dbg.RunCommandInterpreter(True, False, options, 0, False, False)
test.dbg.GetOutputFile().Flush()
error = test.dbg.SetOutputFile(orig_outf)
test.assertSuccess(error, "Restored outfile")
test.dbg.SetInputFile(orig_inf)
test.assertSuccess(error, "Restored infile")
def command_setup(self, args):
"""Insert our command, if needed. Then set up event and barriers if needed.
Then return the command to run."""
self.interp = self.dbg.GetCommandInterpreter()
self.command_name = "interruptible_command"
self.cmd_result = lldb.SBCommandReturnObject()
if not "check" in args:
self.event = threading.Event()
self.result_barrier = threading.Barrier(2, timeout=10)
self.before_interrupt_barrier = threading.Barrier(2, timeout=10)
self.after_interrupt_barrier = threading.Barrier(2, timeout=10)
else:
self.event = None
self.result_barrier = None
self.before_interrupt_barrier = None
self.after_interrupt_barrier = None
if not self.interp.UserCommandExists(self.command_name):
# Make the command we're going to use - it spins calling WasInterrupted:
cmd_filename = "interruptible"
cmd_filename = os.path.join(self.getSourceDir(), "interruptible.py")
self.runCmd(f"command script import {cmd_filename}")
cmd_string = f"command script add {self.command_name} --class interruptible.WelcomeCommand"
self.runCmd(cmd_string)
if len(args) == 0:
command = self.command_name
else:
command = self.command_name + " " + args
return command
def run_single_command(self, command):
# Now start up a thread to run the command:
self.result.Clear()
self.runner = TestDebuggerInterruption.DirectCommandRunner(self, command)
self.runner.start()
def start_command_interp(self):
self.runner = TestDebuggerInterruption.CommandInterpreterRunner(self)
self.runner.start()
def check_text(self, result_text, interrupted):
if interrupted:
self.assertIn(
"Command was interrupted", result_text, "Got the interrupted message"
)
else:
self.assertIn(
"Command was not interrupted",
result_text,
"Got the not interrupted message",
)
def gather_output(self):
# Now wait for the interrupt to interrupt the command:
self.runner.join(10.0)
finished = not self.runner.is_alive()
# Don't leave the runner thread stranded if the interrupt didn't work.
if not finished:
self.event.set()
self.runner.join(10.0)
self.assertTrue(finished, "We did finish the command")
def check_result(self, interrupted=True):
self.gather_output()
self.check_text(self.result.GetOutput(), interrupted)
def check_result_output(self, interrupted=True):
self.gather_output()
buffer = ""
# Okay, now open the file for reading, and read.
with open(self.out_filename, "r") as f:
buffer = f.read()
self.assertNotEqual(len(buffer), 0, "No command data")
self.check_text(buffer, interrupted)
def debugger_interrupt_test(self, use_interrupt_requested):
"""Test that debugger interruption interrupts a command
running directly through HandleCommand.
If use_interrupt_requested is true, we'll check that API,
otherwise we'll check WasInterrupted. They should both do
the same thing."""
if use_interrupt_requested:
command = self.command_setup("debugger")
else:
command = self.command_setup("interp")
self.result = lldb.SBCommandReturnObject()
self.run_single_command(command)
# Okay now wait till the command has gotten started to issue the interrupt:
self.before_interrupt_barrier.wait()
# I'm going to do it twice here to test that it works as a counter:
self.dbg.RequestInterrupt()
self.dbg.RequestInterrupt()
def cleanup():
self.dbg.CancelInterruptRequest()
self.addTearDownHook(cleanup)
# Okay, now set both sides going:
self.after_interrupt_barrier.wait()
# Check that the command was indeed interrupted. First rendevous
# after the runner thread had a chance to execute the command:
self.result_barrier.wait()
self.assertTrue(self.result.Succeeded(), "Our command succeeded")
result_output = self.result.GetOutput()
self.check_result(True)
# Do it again to make sure that the counter is counting:
self.dbg.CancelInterruptRequest()
command = self.command_setup("debugger")
self.run_single_command(command)
# This time we won't even get to run the command, since HandleCommand
# checks for the interrupt state on entry, so we don't wait on the command
# barriers.
self.result_barrier.wait()
# Again check that we were
self.assertFalse(self.result.Succeeded(), "Our command was not allowed to run")
error_output = self.result.GetError()
self.assertIn(
"... Interrupted", error_output, "Command was cut short by interrupt"
)
# Now take down the flag, and make sure that we aren't interrupted:
self.dbg.CancelInterruptRequest()
# Now make sure that we really did take down the flag:
command = self.command_setup("debugger check")
self.run_single_command(command)
result_output = self.result.GetOutput()
self.check_result(False)
def test_debugger_interrupt_use_dbg(self):
self.debugger_interrupt_test(True)
def test_debugger_interrupt_use_interp(self):
self.debugger_interrupt_test(False)
def test_interp_doesnt_interrupt_debugger(self):
"""Test that interpreter interruption does not interrupt a command
running directly through HandleCommand.
If use_interrupt_requested is true, we'll check that API,
otherwise we'll check WasInterrupted. They should both do
the same thing."""
command = self.command_setup("debugger poll")
self.result = lldb.SBCommandReturnObject()
self.run_single_command(command)
# Now raise the debugger interrupt flag. It will also interrupt the command:
self.before_interrupt_barrier.wait()
self.dbg.GetCommandInterpreter().InterruptCommand()
self.after_interrupt_barrier.wait()
# Check that the command was indeed interrupted:
self.result_barrier.wait()
self.assertTrue(self.result.Succeeded(), "Our command succeeded")
result_output = self.result.GetOutput()
self.check_result(False)
def interruptible_command_test(self, use_interrupt_requested):
"""Test that interpreter interruption interrupts a command
running in the RunCommandInterpreter loop.
If use_interrupt_requested is true, we'll check that API,
otherwise we'll check WasInterrupted. They should both do
the same thing."""
self.out_filename = self.getBuildArtifact("output")
self.in_filename = self.getBuildArtifact("input")
# We're going to overwrite the input file, but we
# don't want data accumulating in the output file.
if os.path.exists(self.out_filename):
os.unlink(self.out_filename)
# You should be able to use either check method interchangeably:
if use_interrupt_requested:
self.command = self.command_setup("debugger") + "\n"
else:
self.command = self.command_setup("interp") + "\n"
self.start_command_interp()
# Now give the interpreter a chance to run this command up
# to the first barrier
self.before_interrupt_barrier.wait()
# Then issue the interrupt:
sent_interrupt = self.dbg.GetCommandInterpreter().InterruptCommand()
self.assertTrue(sent_interrupt, "Did send command interrupt.")
# Now give the command a chance to finish:
self.after_interrupt_barrier.wait()
self.check_result_output(True)
os.unlink(self.out_filename)
# Now send the check command, and make sure the flag is now down.
self.command = self.command_setup("interp check") + "\n"
self.start_command_interp()
self.check_result_output(False)
def test_interruptible_command_check_dbg(self):
self.interruptible_command_test(True)
def test_interruptible_command_check_interp(self):
self.interruptible_command_test(False)
def test_debugger_doesnt_interrupt_command(self):
"""Test that debugger interruption doesn't interrupt a command
running in the RunCommandInterpreter loop."""
self.out_filename = self.getBuildArtifact("output")
self.in_filename = self.getBuildArtifact("input")
# We're going to overwrite the input file, but we
# don't want data accumulating in the output file.
if os.path.exists(self.out_filename):
os.unlink(self.out_filename)
self.command = self.command_setup("interp poll") + "\n"
self.start_command_interp()
self.before_interrupt_barrier.wait()
self.dbg.RequestInterrupt()
def cleanup():
self.dbg.CancelInterruptRequest()
self.addTearDownHook(cleanup)
self.after_interrupt_barrier.wait()
self.check_result_output(False)
os.unlink(self.out_filename)
Codebase Browser
llvm