diff --git a/tests/resources/system_tests/test_misc/single_thread.py b/tests/resources/system_tests/test_misc/single_thread.py new file mode 100644 index 000000000..d4edea065 --- /dev/null +++ b/tests/resources/system_tests/test_misc/single_thread.py @@ -0,0 +1,2 @@ + +print('check here') diff --git a/tests/resources/system_tests/test_misc/three_threads.py b/tests/resources/system_tests/test_misc/three_threads.py new file mode 100644 index 000000000..445ac68dd --- /dev/null +++ b/tests/resources/system_tests/test_misc/three_threads.py @@ -0,0 +1,23 @@ +import threading +import time + + +stop = False + + +def worker(tid, offset): + i = 0 + global stop + while not stop: + time.sleep(0.01) + i += 1 + + +threads = [] +for i in [111, 222]: + thread = threading.Thread(target=worker, args=(i, len(threads))) + threads.append(thread) + thread.start() + +print('check here') +stop = True diff --git a/tests/system_tests/test_misc.py b/tests/system_tests/test_misc.py index da02d0627..efda4e219 100644 --- a/tests/system_tests/test_misc.py +++ b/tests/system_tests/test_misc.py @@ -1,6 +1,6 @@ import os import os.path - +import time from tests.helpers.resource import TestResources from . import (lifecycle_handshake, LifecycleTestsBase, DebugInfo) @@ -22,3 +22,47 @@ def test_with_no_output(self): cwd = os.path.dirname(filename) self.run_test_with_no_output( DebugInfo(filename=filename, cwd=cwd)) + + +class ThreadCountTests(LifecycleTestsBase): + def run_test_threads(self, debug_info, bp_filename, bp_line, count): + breakpoints = [{ + 'source': { + 'path': bp_filename + }, + 'breakpoints': [{ + 'line': bp_line + }] + }] + with self.start_debugging(debug_info) as dbg: + session = dbg.session + with session.wait_for_event('stopped') as result: + lifecycle_handshake( + session, debug_info.starttype, + breakpoints=breakpoints, + threads=True) + # Give extra time for thread state to be captured + time.sleep(1) + event = result['msg'] + tid = event.body['threadId'] + req_threads = session.send_request('threads') + req_threads.wait() + threads = req_threads.resp.body['threads'] + + session.send_request('continue', threadId=tid) + + self.assertEqual(count, len(threads)) + + def test_single_thread(self): + filename = TEST_FILES.resolve('single_thread.py') + cwd = os.path.dirname(filename) + self.run_test_threads( + DebugInfo(filename=filename, cwd=cwd), + bp_filename=filename, bp_line=2, count=1) + + def test_multi_thread(self): + filename = TEST_FILES.resolve('three_threads.py') + cwd = os.path.dirname(filename) + self.run_test_threads( + DebugInfo(filename=filename, cwd=cwd), + bp_filename=filename, bp_line=22, count=3)