import time
import unittest
import pexpect
from helper import Debugger
class MultithreadTestCase(unittest.TestCase):
def setUp(self):
self.debugger = Debugger(path='./examples/target/debug/mt')
def test_multithreaded_app_running(self):
self.debugger.cmd(
'run',
'thread 1 spawn',
'thread 2 spawn',
'sum2: 199990000',
'sum1: 49995000',
'total 249985000',
)
def test_multithreaded_breakpoints(self):
self.debugger.cmd('break mt.rs:6', 'New breakpoint')
self.debugger.cmd('break mt.rs:24', 'New breakpoint')
self.debugger.cmd('break mt.rs:36', 'New breakpoint')
self.debugger.cmd('break mt.rs:14', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at', '6 let jh1 = thread::spawn(sum1);')
self.debugger.cmd(
'continue',
'thread 1 spawn',
'thread 2 spawn',
'Hit breakpoint 3 at',
'36 let mut sum2 = 0;',
)
self.debugger.cmd('continue', 'Hit breakpoint 2 at', '24 let mut sum = 0;')
self.debugger.cmd('continue', 'Hit breakpoint 4 at', '14 println!("total {}", sum1 + sum2);')
self.debugger.cmd('continue', 'total 249985000')
def test_multithreaded_backtrace(self):
self.debugger.cmd('break mt.rs:24', 'New breakpoint')
self.debugger.cmd('run', 'thread 1 spawn', 'Hit breakpoint 1 at', '24 let mut sum = 0;')
self.debugger.cmd('backtrace', 'mt::sum1', 'new::thread_start')
def test_multithreaded_trace(self):
self.debugger.cmd('break mt.rs:36', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at', '36 let mut sum2 = 0;')
self.debugger.cmd(
'backtrace all',
'thread',
'mt::main',
'thread',
'clock_nanosleep',
'thread',
'mt::sum2',
)
def test_multithreaded_quit(self):
self.debugger.cmd('break mt.rs:36', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at', '36 let mut sum2 = 0;')
self.debugger.cmd('quit')
time.sleep(2)
self.assertFalse(self.debugger.is_alive())
def test_thread_info(self):
self.debugger.cmd('break mt.rs:40', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at')
self.debugger.cmd('thread info', '#1 thread id', '#2 thread id', '#3 thread id')
self.debugger.cmd('thread current', '#3 thread id')
def test_thread_switch(self):
self.debugger.cmd('break mt.rs:40', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at')
self.debugger.cmd('thread current', '#3 thread id')
self.debugger.cmd('thread switch 2', 'Thread #2 brought into focus')
self.debugger.cmd('thread current', '#2 thread id')
while True:
try:
self.debugger.expect_in_output('24 let mut sum = 0;', timeout=1)
except pexpect.ExceptionPexpect:
self.debugger.cmd('step')
else:
break
self.debugger.cmd('step', '25 for i in 0..10000')
self.debugger.cmd('var locals', 'sum = i32(0)')
def test_thread_switch_frame_switch(self):
self.debugger.cmd('break mt.rs:40', 'New breakpoint')
self.debugger.cmd('run', 'Hit breakpoint 1 at')
self.debugger.cmd('thread current', '#3 thread id')
self.debugger.cmd('thread switch 2', 'Thread #2 brought into focus')
self.debugger.cmd('thread current', '#2 thread id')
self.debugger.cmd('frame switch 2', 'switch to #2')
self.debugger.cmd('var locals', 'sum3_jh = JoinHandle<i32> {')