MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-10-01 19:02:13 +07:00
Code Issues Packages Projects Releases Wiki Activity

PyDev: Added test for single notification on regular case.

Browse Source 
GitOrigin-RevId: 2f6262bf771f7f24a46cf87790b13f5786c85f66
This commit is contained in:
Elizaveta Shashkova
2019-06-07 18:11:10 +03:00
committed by intellij-monorepo-bot
parent f8ef1eaa95
commit 0dd489592b
5 changed files with 59 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/helpers/pydev/_pydevd_bundle/pydevd_comm.py
+1 -1
View File
@@ -107,6 +107,7 @@ from _pydev_bundle import _pydev_completer
from pydevd_tracing import get_exception_traceback_str
from _pydevd_bundle import pydevd_console
from _pydev_bundle.pydev_monkey import disable_trace_thread_modules, enable_trace_thread_modules
from _pydevd_bundle.pydevd_additional_thread_info import set_additional_thread_info
try:
import cStringIO as StringIO #may not always be available @UnusedImport
@@ -1086,7 +1087,6 @@ class InternalGetThreadStack(InternalThreadCommand):
@overrides(InternalThreadCommand.can_be_executed_by)
def can_be_executed_by(self, _thread_id):
from _pydevd_bundle.pydevd_additional_thread_info import set_additional_thread_info
timed_out = time.time() >= self._timeout
py_db = self._py_db()
python/helpers/pydev/tests_python/debugger_unittest.py
+20 -7
View File
@@ -95,6 +95,7 @@ REASON_CAUGHT_EXCEPTION = CMD_STEP_CAUGHT_EXCEPTION
REASON_UNCAUGHT_EXCEPTION = CMD_ADD_EXCEPTION_BREAK
REASON_STOP_ON_BREAKPOINT = CMD_SET_BREAK
REASON_THREAD_SUSPEND = CMD_THREAD_SUSPEND
REASON_STEP_INTO = CMD_STEP_INTO
REASON_STEP_INTO_MY_CODE = CMD_STEP_INTO_MY_CODE
REASON_STEP_OVER = CMD_STEP_OVER
@@ -200,17 +201,19 @@ class ReaderThread(threading.Thread):
else:
frame = sys._getframe().f_back.f_back
frame_info = ''
i = 3
while frame:
if frame.f_code.co_filename.endswith('debugger_unittest.py'):
frame = frame.f_back
continue
stack_msg = ' -- File "%s", line %s, in %s\n' % (frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name)
if 'run' == frame.f_code.co_name:
frame_info = stack_msg # Ok, found the writer thread 'run' method (show only that).
break
frame_info += stack_msg
frame = frame.f_back
i -= 1
if i == 0:
break
# Just print the first which is not debugger_unittest.py
break
frame = None
sys.stdout.write('Message returned in get_next_message(): %s -- ctx: %s, asked at:\n%s\n' % (unquote_plus(unquote_plus(msg)), context_message, frame_info))
return msg
@@ -385,7 +388,7 @@ class DebuggerRunner(object):
expected_returncode = writer.EXPECTED_RETURNCODE
if not isinstance(expected_returncode, (list, tuple)):
expected_returncode = (expected_returncode,)
if process.returncode not in expected_returncode:
self.fail_with_message('Expected process.returncode to be %s. Found: %s' % (
writer.EXPECTED_RETURNCODE, process.returncode), stdout, stderr, writer)
@@ -646,10 +649,20 @@ class AbstractWriterThread(threading.Thread):
msg = unquote_plus(unquote_plus(msg.split('"')[1]))
return msg, ctx
def get_current_stack_hit(self, thread_id):
def get_current_stack_hit(self, thread_id, **kwargs):
self.write_get_thread_stack(thread_id)
msg = self.wait_for_message(CMD_GET_THREAD_STACK)
return self._get_stack_as_hit(msg)
return self._get_stack_as_hit(msg, **kwargs)
def wait_for_single_notification_as_hit(self, reason=REASON_STOP_ON_BREAKPOINT, **kwargs):
dct = self.wait_for_json_message(CMD_THREAD_SUSPEND_SINGLE_NOTIFICATION)
assert dct['stop_reason'] == reason
line = kwargs.pop('line', None)
file = kwargs.pop('file', None)
assert not kwargs, 'Unexpected kwargs: %s' % (kwargs,)
return self.get_current_stack_hit(dct['thread_id'], line=line, file=file)
def wait_for_breakpoint_hit(self, reason=REASON_STOP_ON_BREAKPOINT, timeout=None, **kwargs):
'''
python/helpers/pydev/tests_python/resources/_debugger_case_import_imported.py
+2
View File
@@ -0,0 +1,2 @@
print('text 1')
print('text 2')
python/helpers/pydev/tests_python/resources/_debugger_case_import_main.py
+3
View File
@@ -0,0 +1,3 @@
import _debugger_case_import_imported # break here
print('TEST SUCEEDED')
python/helpers/pydev/tests_python/test_debugger.py
+33 -16
View File
@@ -859,6 +859,8 @@ def test_case_20(case_setup):
writer.finished_ok = True
@pytest.mark.skipif(not TEST_FLASK, reason='No flask available')
def test_case_flask(case_setup_flask):
with case_setup_flask.test_file(EXPECTED_RETURNCODE='any') as writer:
@@ -871,23 +873,15 @@ def test_case_flask(case_setup_flask):
time.sleep(2) # Give flask some time to get to startup before requesting the page
t.start()
dct = writer.wait_for_json_message(CMD_THREAD_SUSPEND_SINGLE_NOTIFICATION)
assert dct['stop_reason'] == REASON_STOP_ON_BREAKPOINT
writer.write_get_thread_stack(dct['thread_id'])
msg = writer.wait_for_get_thread_stack_message()
assert msg.thread.frame[0]['line'] == '5'
writer.write_get_frame(dct['thread_id'], msg.thread.frame[0]['id'])
hit = writer.wait_for_single_notification_as_hit(line=5)
writer.write_get_frame(hit.thread_id, hit.frame_id)
writer.wait_for_vars(['<var name="content" type="str"'])
writer.write_run_thread(dct['thread_id'])
writer.write_run_thread(hit.thread_id)
dct = writer.wait_for_json_message(CMD_THREAD_SUSPEND_SINGLE_NOTIFICATION)
assert dct['stop_reason'] == REASON_STOP_ON_BREAKPOINT
writer.write_get_thread_stack(dct['thread_id'])
msg = writer.wait_for_get_thread_stack_message()
assert msg.thread.frame[0]['line'] == '8'
writer.write_get_frame(dct['thread_id'], msg.thread.frame[0]['id'])
hit = writer.wait_for_single_notification_as_hit(line=8)
writer.write_get_frame(hit.thread_id, hit.frame_id)
writer.wait_for_vars(['<var name="content" type="str"'])
writer.write_run_thread(dct['thread_id'])
writer.write_run_thread(hit.thread_id)
for _ in xrange(10):
if hasattr(t, 'contents'):
@@ -2280,6 +2274,29 @@ def test_trace_dispatch_correct(case_setup):
writer.finished_ok = True
def test_case_single_notification_on_step(case_setup):
from tests_python.debugger_unittest import REASON_STEP_INTO
with case_setup.test_file('_debugger_case_import_main.py') as writer:
writer.write_multi_threads_single_notification(True)
writer.write_add_breakpoint(writer.get_line_index_with_content('break here'), '')
writer.write_make_initial_run()
hit = writer.wait_for_single_notification_as_hit()
writer.write_step_in(hit.thread_id)
hit = writer.wait_for_single_notification_as_hit(reason=REASON_STEP_INTO)
writer.write_step_in(hit.thread_id)
hit = writer.wait_for_single_notification_as_hit(reason=REASON_STEP_INTO)
writer.write_step_in(hit.thread_id)
hit = writer.wait_for_single_notification_as_hit(reason=REASON_STEP_INTO)
writer.write_run_thread(hit.thread_id)
writer.finished_ok = True
@pytest.mark.skipif(IS_JYTHON, reason='Jython can only have one thread stopped at each time.')
@pytest.mark.parametrize('check_single_notification', [True, False])
def test_run_pause_all_threads_single_notification(case_setup, check_single_notification):
@@ -2319,10 +2336,10 @@ def test_run_pause_all_threads_single_notification(case_setup, check_single_noti
writer.write_step_over(thread_id1)
if check_single_notification:
dct = writer.wait_for_json_message(CMD_THREAD_RESUME_SINGLE_NOTIFICATION)
dct = writer.wait_for_json_message(CMD_THREAD_RESUME_SINGLE_NOTIFICATION) # Note: prefer wait_for_single_notification_as_hit
assert dct['thread_id'] == thread_id1
dct = writer.wait_for_json_message(CMD_THREAD_SUSPEND_SINGLE_NOTIFICATION)
dct = writer.wait_for_json_message(CMD_THREAD_SUSPEND_SINGLE_NOTIFICATION) # Note: prefer wait_for_single_notification_as_hit
assert dct['thread_id'] == thread_id1
assert dct['stop_reason'] == REASON_STEP_OVER