TestThreadStates.py [plain text]
"""
Test thread states.
"""
import os, time
import unittest2
import lldb
from lldbtest import *
import lldbutil
class ThreadStateTestCase(TestBase):
mydir = os.path.join("functionalities", "thread", "state")
@unittest2.skipUnless(sys.platform.startswith("darwin"), "requires Darwin")
@dsym_test
def test_state_after_breakpoint_with_dsym(self):
"""Test thread state after breakpoint."""
self.buildDsym(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_breakpoint_test()
@dwarf_test
def test_state_after_breakpoint_with_dwarf(self):
"""Test thread state after breakpoint."""
self.buildDwarf(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_breakpoint_test()
@unittest2.skipUnless(sys.platform.startswith("darwin"), "requires Darwin")
@dsym_test
def test_state_after_continue_with_dsym(self):
"""Test thread state after continue."""
self.buildDsym(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_continue_test()
@dwarf_test
def test_state_after_continue_with_dwarf(self):
"""Test thread state after continue."""
self.buildDwarf(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_continue_test()
@unittest2.skipUnless(sys.platform.startswith("darwin"), "requires Darwin")
@dsym_test
def test_state_after_expression_with_dsym(self):
"""Test thread state after expression."""
self.buildDsym(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_continue_test()
@dwarf_test
def test_state_after_expression_with_dwarf(self):
"""Test thread state after expression."""
self.buildDwarf(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_state_after_continue_test()
@unittest2.skipUnless(sys.platform.startswith("darwin"), "requires Darwin")
@dsym_test
@unittest2.expectedFailure("llvm.org/pr16172") def test_process_interrupt_with_dsym(self):
"""Test process interrupt."""
self.buildDsym(dictionary=self.getBuildFlags(use_cpp11=False))
self.process_interrupt_test()
@dwarf_test
@unittest2.expectedFailure("llvm.org/pr16712") def test_process_interrupt_with_dwarf(self):
"""Test process interrupt."""
self.buildDwarf(dictionary=self.getBuildFlags(use_cpp11=False))
self.process_interrupt_test()
@unittest2.skipUnless(sys.platform.startswith("darwin"), "requires Darwin")
@dsym_test
@unittest2.expectedFailure("llvm.org/pr15824") def test_process_state_with_dsym(self):
"""Test thread states (comprehensive)."""
self.buildDsym(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_states_test()
@dwarf_test
@unittest2.expectedFailure("llvm.org/pr15824") def test_process_state_with_dwarf(self):
"""Test thread states (comprehensive)."""
self.buildDwarf(dictionary=self.getBuildFlags(use_cpp11=False))
self.thread_states_test()
def setUp(self):
TestBase.setUp(self)
self.break_1 = line_number('main.c', '// Set first breakpoint here')
self.break_2 = line_number('main.c', '// Set second breakpoint here')
def thread_state_after_breakpoint_test(self):
"""Test thread state after breakpoint."""
exe = os.path.join(os.getcwd(), "a.out")
self.runCmd("file " + exe, CURRENT_EXECUTABLE_SET)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_1, num_expected_locations=1)
self.expect("breakpoint list -f", "Breakpoint location shown correctly",
substrs = ["1: file = 'main.c', line = %d, locations = 1" % self.break_1])
self.runCmd("run", RUN_SUCCEEDED)
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
target = self.dbg.GetSelectedTarget()
process = target.GetProcess()
num_threads = process.GetNumThreads()
self.assertTrue(num_threads == 1, 'Number of expected threads and actual threads do not match.')
thread = process.GetThreadAtIndex(0)
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' during breakpoint 1.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' during breakpoint 1.")
self.runCmd("process kill")
def thread_state_after_continue_test(self):
"""Test thread state after continue."""
exe = os.path.join(os.getcwd(), "a.out")
self.runCmd("file " + exe, CURRENT_EXECUTABLE_SET)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_1, num_expected_locations=1)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_2, num_expected_locations=1)
self.expect("breakpoint list -f", "Breakpoint location shown correctly",
substrs = ["1: file = 'main.c', line = %d, locations = 1" % self.break_1])
self.runCmd("run", RUN_SUCCEEDED)
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
target = self.dbg.GetSelectedTarget()
process = target.GetProcess()
num_threads = process.GetNumThreads()
self.assertTrue(num_threads == 1, 'Number of expected threads and actual threads do not match.')
thread = process.GetThreadAtIndex(0)
self.dbg.SetAsync(True)
self.runCmd("continue")
time.sleep(1)
self.assertFalse(thread.IsStopped(), "Thread state is \'stopped\' when it should be running.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' when it should be running.")
self.dbg.SetAsync(False)
self.runCmd("process kill")
def thread_state_after_expression_test(self):
"""Test thread state after expression."""
exe = os.path.join(os.getcwd(), "a.out")
self.runCmd("file " + exe, CURRENT_EXECUTABLE_SET)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_1, num_expected_locations=1)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_2, num_expected_locations=1)
self.expect("breakpoint list -f", "Breakpoint location shown correctly",
substrs = ["1: file = 'main.c', line = %d, locations = 1" % self.break_1])
self.runCmd("run", RUN_SUCCEEDED)
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
target = self.dbg.GetSelectedTarget()
process = target.GetProcess()
num_threads = process.GetNumThreads()
self.assertTrue(num_threads == 1, 'Number of expected threads and actual threads do not match.')
thread = process.GetThreadAtIndex(0)
self.runCmd("expression g_test = 1")
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' after expression evaluation.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' after expression evaluation.")
self.runCmd("process continue")
def process_interrupt_test(self):
"""Test process interrupt and continue."""
exe = os.path.join(os.getcwd(), "a.out")
self.runCmd("file " + exe, CURRENT_EXECUTABLE_SET)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_1, num_expected_locations=1)
self.expect("breakpoint list -f", "Breakpoint location shown correctly",
substrs = ["1: file = 'main.c', line = %d, locations = 1" % self.break_1])
self.runCmd("run", RUN_SUCCEEDED)
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
target = self.dbg.GetSelectedTarget()
process = target.GetProcess()
num_threads = process.GetNumThreads()
self.assertTrue(num_threads == 1, 'Number of expected threads and actual threads do not match.')
self.dbg.SetAsync(True)
self.runCmd("continue")
time.sleep(1)
self.dbg.SetAsync(False)
self.runCmd("process interrupt")
self.expect("process status", STOPPED_DUE_TO_SIGNAL,
substrs = ['stopped',
'* thread #1',
'stop reason = signal'])
self.runCmd("expression g_test = 1")
self.runCmd("continue")
def thread_states_test(self):
"""Test thread states (comprehensive)."""
exe = os.path.join(os.getcwd(), "a.out")
self.runCmd("file " + exe, CURRENT_EXECUTABLE_SET)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_1, num_expected_locations=1)
lldbutil.run_break_set_by_file_and_line (self, "main.c", self.break_2, num_expected_locations=1)
self.expect("breakpoint list -f", "Breakpoint location shown correctly",
substrs = ["1: file = 'main.c', line = %d, locations = 1" % self.break_1,
"2: file = 'main.c', line = %d, locations = 1" % self.break_2])
self.runCmd("run", RUN_SUCCEEDED)
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
target = self.dbg.GetSelectedTarget()
process = target.GetProcess()
num_threads = process.GetNumThreads()
self.assertTrue(num_threads == 1, 'Number of expected threads and actual threads do not match.')
thread = process.GetThreadAtIndex(0)
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' during breakpoint 1.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' during breakpoint 1.")
self.dbg.SetAsync(True)
self.runCmd("continue")
time.sleep(1)
self.assertFalse(thread.IsStopped(), "Thread state is \'stopped\' when it should be running.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' when it should be running.")
self.dbg.SetAsync(False)
self.runCmd("process interrupt")
self.expect("process status", STOPPED_DUE_TO_SIGNAL,
substrs = ['stopped',
'* thread #1',
'stop reason = signal'])
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' after process stop.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' after process stop.")
self.runCmd("expression g_test = 1")
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' after expression evaluation.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' after expression evaluation.")
self.expect("process status", STOPPED_DUE_TO_SIGNAL,
substrs = ['stopped',
'* thread #1',
'stop reason = signal'])
self.runCmd("continue")
self.expect("thread list", STOPPED_DUE_TO_BREAKPOINT,
substrs = ['stopped',
'* thread #1',
'stop reason = breakpoint'])
self.assertTrue(thread.IsStopped(), "Thread state isn't \'stopped\' during breakpoint 2.")
self.assertFalse(thread.IsSuspended(), "Thread state is \'suspended\' during breakpoint 2.")
self.runCmd("continue")
self.assertTrue(process.GetState() == lldb.eStateExited, PROCESS_EXITED)
if __name__ == '__main__':
import atexit
lldb.SBDebugger.Initialize()
atexit.register(lambda: lldb.SBDebugger.Terminate())
unittest2.main()