Add thread count tests (#668)

* Add thread count tests

* Fix Linter

* Adding sleep to allow thread state to be captured accurately

tests/resources/system_tests/test_misc/single_thread.py

@@ -0,0 +1,2 @@
+
+print('check here')

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

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)